From 5ad130c94f1645edb4630d277aab25f91e06439c Mon Sep 17 00:00:00 2001 From: daniel Date: Tue, 8 Sep 2026 00:28:13 +0100 Subject: [PATCH 1/2] feat(compose): expose source-attributed Runlet progress --- Cargo.lock | 4 +- crates/agentkit-tool-compose/Cargo.toml | 2 +- crates/agentkit-tool-compose/README.md | 29 ++ crates/agentkit-tool-compose/src/lib.rs | 7 +- .../src/runlet_backend.rs | 176 +++++++- .../agentkit-tool-compose/src/tests/runlet.rs | 386 +++++++++++++++++- 6 files changed, 596 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cc57ddd..3fc7739 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3111,9 +3111,9 @@ dependencies = [ [[package]] name = "runlet" -version = "0.5.0" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "927f293f868ee447cfc55b1e5220f1717a907c566f45d836c8810e5155b3b6c8" +checksum = "057e0864428c5a79a68683942d3750d05e9ffae541aa0185fde9ae1c87eca100" dependencies = [ "hex", "regex", diff --git a/crates/agentkit-tool-compose/Cargo.toml b/crates/agentkit-tool-compose/Cargo.toml index cd3361e..bd1439d 100644 --- a/crates/agentkit-tool-compose/Cargo.toml +++ b/crates/agentkit-tool-compose/Cargo.toml @@ -14,7 +14,7 @@ agentkit-core = { version = "0.10.5", path = "../agentkit-core" } agentkit-tools-core = { version = "0.10.5", path = "../agentkit-tools-core" } async-trait.workspace = true mlua = { workspace = true, optional = true } -runlet = { version = "0.5.0", optional = true } +runlet = { version = "0.6.0", optional = true } serde = { workspace = true, features = ["derive"] } serde_json.workspace = true serde_toon2 = { version = "0.2.0", optional = true } diff --git a/crates/agentkit-tool-compose/README.md b/crates/agentkit-tool-compose/README.md index 41f6e10..d62770e 100644 --- a/crates/agentkit-tool-compose/README.md +++ b/crates/agentkit-tool-compose/README.md @@ -56,3 +56,32 @@ With the `toon` feature enabled, renders uniform object lists as a header plus one row per element — smaller than JSON for the list-shaped values compose scripts tend to return. The tool description gains a note explaining the format so the model can read it. + +## Scoped Runlet progress + +A custom `ComposeBackend` can delegate to +`RunletBackend.execute_with_progress(run, sink, capacity).await` instead of +`RunletBackend.execute(run).await`. `sink` is a host-owned bounded +`tokio::sync::mpsc::Sender`; `capacity` is a `NonZeroUsize` +bounding each execution's metadata queue. The unit backend and `BackendRun` +remain unchanged. + +Each received envelope starts observation of one compiled execution and carries +its exact `parent_call_id`, process-local unique `incarnation`, compiled +`source_digest`, and `healed` flag. Poll `RunletProgress::try_recv` in the host's +own scoped task until `RunletProgressEnd`. No observer threads or callbacks are +created. A full/dropped sink means the execution is unobserved; a dropped +per-execution receiver or overflowing queue never blocks execution. + +Events contain only typed Runlet metadata, not inputs, outputs, or error text. +Raw runtime completion is withheld until the compose host checks its outcome. +`Interrupted` and `Incomplete` invalidate the entire observed incarnation; +`Lagged` means state beyond the received prefix is unknown. Approval replay +gets a fresh incarnation. Dropping the execution future invalidates observation +even if its existing blocking runtime is still completing. + +Byte spans refer only to the digest-matching compiled source. In particular, +never map `healed` spans onto the submitted script. Source text is not included +because it can contain secrets; if matching source is unavailable, display +metadata without source snippets. Node IDs are execution-local, and neither a +tool name nor an absent event proves which call is running. diff --git a/crates/agentkit-tool-compose/src/lib.rs b/crates/agentkit-tool-compose/src/lib.rs index ee876d4..95bf80c 100644 --- a/crates/agentkit-tool-compose/src/lib.rs +++ b/crates/agentkit-tool-compose/src/lib.rs @@ -36,7 +36,7 @@ mod runlet_backend; #[cfg(feature = "lua")] pub use lua::LuaBackend; #[cfg(feature = "runlet")] -pub use runlet_backend::RunletBackend; +pub use runlet_backend::{RunletBackend, RunletProgress, RunletProgressEnd}; pub const COMPOSE_TOOL_NAME: &str = "compose"; @@ -222,6 +222,11 @@ pub struct ChildDispatcher { } impl ChildDispatcher { + /// Exact compose call that owns this dispatcher, including approval replays. + pub fn parent_call_id(&self) -> &ToolCallId { + &self.parent_call_id + } + pub fn is_cancelled(&self) -> bool { self.cancellation .as_ref() diff --git a/crates/agentkit-tool-compose/src/runlet_backend.rs b/crates/agentkit-tool-compose/src/runlet_backend.rs index 6e822f4..4438895 100644 --- a/crates/agentkit-tool-compose/src/runlet_backend.rs +++ b/crates/agentkit-tool-compose/src/runlet_backend.rs @@ -26,6 +26,120 @@ use crate::{ BackendRun, CallKey, ComposeBackend, ComposeOutcome, DispatchError, render_catalog_shapes, }; +// Single writer: the async execute guard commits host outcome, or invalidates on +// drop/unwind. The blocking executor and consumer only read/share this atomic. +// No new locks, callbacks, or observation workers; executor ownership is unchanged. +type ProgressSink = ( + tokio::sync::mpsc::Sender, + std::num::NonZeroUsize, + Arc, +); +static NEXT_INCARNATION: AtomicU64 = AtomicU64::new(1); +const ACTIVE: u8 = 0; +const SUCCEEDED: u8 = 1; +const FAILED: u8 = 2; +const INTERRUPTED: u8 = 3; +const INCOMPLETE: u8 = 4; + +struct ProgressGuard(Arc); +impl Drop for ProgressGuard { + fn drop(&mut self) { + let _ = self + .0 + .compare_exchange(ACTIVE, INCOMPLETE, Ordering::Release, Ordering::Relaxed); + } +} + +/// Host-authoritative terminal or observation gap. A gap never stops execution. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RunletProgressEnd { + /// Compose completed successfully, and the complete runtime stream was read. + Succeeded, + /// Compose failed, and the complete runtime stream was read. + Failed, + /// Approval suspended execution. Invalidate all observations of this incarnation. + Interrupted, + /// Execution future dropped/cancelled, or publication ended abnormally. + /// Invalidate all observations of this incarnation. + Incomplete, + /// Queue overflow: state after the received prefix is unknown. + Lagged, +} + +/// A Started envelope for one successfully compiled execution. +/// +/// Delivered with `try_send` into a host-owned bounded Tokio channel. A full or +/// dropped sink leaves the execution unobserved; absence never proves inactivity. +/// The host polls this receiver in its own scoped task; no worker is spawned. +/// Dropping either receiver cannot block or fail execution. +/// +/// Events contain no arguments, outputs, error text, or tool-name ownership guesses. +/// Node IDs and sequences are local to `(parent_call_id, incarnation)`. Unobserved +/// expressions have unknown state. Raw runtime Finished is suppressed: only the +/// host-authoritative end below establishes compose completion. +pub struct RunletProgress { + /// Exact parent compose call, not a tool-name match. + pub parent_call_id: agentkit_core::ToolCallId, + /// Unique within this loaded library's process lifetime (not durable identity). + /// Approval replay and concurrent executions receive fresh incarnations. + pub incarnation: u64, + /// SHA-256 of exact compiled source bytes, including auto-healing. + pub source_digest: String, + /// If true, NEVER map byte spans onto the submitted source. Source text is + /// deliberately omitted: scripts can contain secrets. Without independently + /// available digest-matching source, display metadata without source snippets. + pub healed: bool, + receiver: runlet::ProgressReceiver, + state: Arc, + ended: Option, + cancellation: Option, +} + +impl RunletProgress { + /// Nonblocking poll. `Ok(None)` means still open, not completion. + /// Poll until terminal; Interrupted/Incomplete invalidate the entire prefix. + pub fn try_recv(&mut self) -> Result, RunletProgressEnd> { + if let Some(end) = self.ended { + return Err(end); + } + let state = self.state.load(Ordering::Acquire); + let invalid = match state { + ACTIVE if self.cancellation.as_ref().is_some_and(|c| c.is_cancelled()) => { + Some(RunletProgressEnd::Incomplete) + } + INTERRUPTED => Some(RunletProgressEnd::Interrupted), + INCOMPLETE => Some(RunletProgressEnd::Incomplete), + _ => None, + }; + if let Some(end) = invalid { + self.ended = Some(end); + return Err(end); + } + match self.receiver.try_recv() { + Ok(Some(event)) if matches!(event.change, runlet::ProgressChange::Finished(_)) => { + // Do not expose runtime success before the host checks approval. + self.try_recv() + } + Ok(event) => Ok(event), + Err(runlet::ProgressRecvError::Closed) if state == ACTIVE => Ok(None), + Err(error) => { + let end = match error { + runlet::ProgressRecvError::Closed if state == SUCCEEDED => { + RunletProgressEnd::Succeeded + } + runlet::ProgressRecvError::Closed if state == FAILED => { + RunletProgressEnd::Failed + } + runlet::ProgressRecvError::Lagged => RunletProgressEnd::Lagged, + _ => RunletProgressEnd::Incomplete, + }; + self.ended = Some(end); + Err(end) + } + } + } +} + /// Loop concurrency defaults for compose runs. Each active iteration pins one /// OS thread while its tool call blocks on the async executor, so these are /// deliberately far below runlet's own defaults. @@ -293,6 +407,41 @@ impl ComposeBackend for RunletBackend { } async fn execute(&self, run: BackendRun) -> Result { + self.execute_inner(run, None).await + } +} + +impl RunletBackend { + /// Executes with bounded, payload-free progress. See [`RunletProgress`]. + pub async fn execute_with_progress( + &self, + run: BackendRun, + sink: tokio::sync::mpsc::Sender, + capacity: std::num::NonZeroUsize, + ) -> Result { + let guard = ProgressGuard(Arc::new(std::sync::atomic::AtomicU8::new(ACTIVE))); + let cancellation = run.cancellation.clone(); + let result = self + .execute_inner(run, Some((sink, capacity, guard.0.clone()))) + .await; + let state = if cancellation.as_ref().is_some_and(|c| c.is_cancelled()) { + INCOMPLETE + } else { + match &result { + Ok(_) => SUCCEEDED, + Err(ComposeOutcome::Interrupted(_)) => INTERRUPTED, + Err(_) => FAILED, + } + }; + guard.0.store(state, Ordering::Release); + result + } + + async fn execute_inner( + &self, + run: BackendRun, + progress: Option, + ) -> Result { if std::env::var_os("COMPOSE_RUNLET_DEBUG").is_some() { eprintln!("[compose-runlet] executing program:\n{}\n---", run.script); } @@ -380,6 +529,8 @@ impl ComposeBackend for RunletBackend { ))) })?; + let parent_call_id = run.dispatcher.parent_call_id().clone(); + let cancellation = run.cancellation.clone(); let script = run.script.clone(); let result = tokio::task::spawn_blocking(move || { let (program, heal_notes) = match runtime.compile(&script) { @@ -407,8 +558,29 @@ impl ComposeBackend for RunletBackend { } } }; - runtime - .run(&program) + let execution = if let Some((sink, capacity, state)) = progress { + let (sender, receiver) = runlet::progress_channel(capacity.get()); + // Exhaustion disables observation rather than reusing an identity. + if let Ok(incarnation) = + NEXT_INCARNATION + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1)) + { + let _ = sink.try_send(RunletProgress { + parent_call_id, + incarnation, + source_digest: program.source_digest.clone(), + healed: program.source != script, + receiver, + state, + ended: None, + cancellation, + }); + } + runtime.run_with_progress(&program, sender) + } else { + runtime.run(&program) + }; + execution .map(|execution| (execution, heal_notes)) .map_err(RunletRunError::Run) }) diff --git a/crates/agentkit-tool-compose/src/tests/runlet.rs b/crates/agentkit-tool-compose/src/tests/runlet.rs index ed6aace..1d10dde 100644 --- a/crates/agentkit-tool-compose/src/tests/runlet.rs +++ b/crates/agentkit-tool-compose/src/tests/runlet.rs @@ -320,7 +320,8 @@ async fn compose_is_not_callable_from_runlet() { #[tokio::test] async fn nested_approval_interrupts_and_resumes_with_replay() { - let compose = ComposeTool::new(ComposeConfig::default()).with_backend(RunletBackend); + let (backend, mut progress_receiver) = progress_backend(1024); + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); let first = EchoTool::new(); let gated = ApprovalEchoTool::new(); let first_calls = first.calls.clone(); @@ -331,7 +332,7 @@ async fn nested_approval_interrupts_and_resumes_with_replay() { let permissions: Arc = Arc::new(RequireApproval); let req = request( "a = echo({ value: 1 })\n\ - b = approval_echo({ value: a.value + 1 })\n\ + b = boundary { return approval_echo({ value: a.value + 1 }) } catch err { return { value: 0 } }\n\ return b", Value::Null, ); @@ -361,6 +362,18 @@ async fn nested_approval_interrupts_and_resumes_with_replay() { // dispatching again; only the approved call executes. assert_eq!(first_calls.load(Ordering::SeqCst), 1); assert_eq!(gated_calls.load(Ordering::SeqCst), 1); + let mut first = progress_receiver.try_recv().unwrap(); + let mut replay = progress_receiver.try_recv().unwrap(); + assert_eq!(first.parent_call_id, replay.parent_call_id); + assert_ne!(first.incarnation, replay.incarnation); + assert_eq!( + collect_progress(&mut first).1, + crate::RunletProgressEnd::Interrupted + ); + assert_eq!( + collect_progress(&mut replay).1, + crate::RunletProgressEnd::Succeeded + ); } #[tokio::test] @@ -465,3 +478,372 @@ async fn prelude_intrinsics_and_folds_run_locally_without_consuming_call_budget( other => panic!("unexpected outcome: {other:?}"), } } + +struct ProgressBackend { + sink: tokio::sync::mpsc::Sender, + capacity: std::num::NonZeroUsize, +} + +#[async_trait::async_trait] +impl crate::ComposeBackend for ProgressBackend { + fn name(&self) -> &'static str { + RunletBackend.name() + } + fn description(&self, catalog: Option<&[ToolSpec]>) -> String { + RunletBackend.description(catalog) + } + fn script_description(&self) -> &'static str { + RunletBackend.script_description() + } + async fn execute(&self, run: crate::BackendRun) -> Result { + RunletBackend + .execute_with_progress(run, self.sink.clone(), self.capacity) + .await + } +} + +fn progress_backend( + capacity: usize, +) -> ( + ProgressBackend, + tokio::sync::mpsc::Receiver, +) { + let (sink, receiver) = tokio::sync::mpsc::channel(4); + ( + ProgressBackend { + sink, + capacity: std::num::NonZeroUsize::new(capacity).unwrap(), + }, + receiver, + ) +} + +fn collect_progress( + progress: &mut crate::RunletProgress, +) -> (Vec, crate::RunletProgressEnd) { + let mut events = Vec::new(); + loop { + match progress.try_recv() { + Ok(Some(event)) => events.push(event), + Ok(None) => panic!("execution was already awaited"), + Err(end) => return (events, end), + } + } +} + +#[tokio::test] +async fn progress_namespaces_concurrent_parents_and_orders_repeated_calls() { + let (backend, mut receiver) = progress_backend(1024); + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); + let executor: Arc = Arc::new(BasicToolExecutor::from_registry( + ToolRegistry::new().with(compose).with(EchoTool::new()), + )); + let script = "a = echo({ value: 1 })\nb = echo({ value: a.value + 1 })\nreturn b"; + let run = |id: &'static str| { + let executor = executor.clone(); + async move { + let owned = owned_context(executor.clone(), Arc::new(AllowAllPermissions)); + let mut ctx = owned.borrowed(); + let mut req = request(script, Value::Null); + req.call_id = ToolCallId::new(id); + assert!(matches!( + executor.execute(req, &mut ctx).await, + ToolExecutionOutcome::Completed(_) + )); + } + }; + tokio::join!(run("parent-a"), run("parent-b")); + let mut a = receiver.try_recv().unwrap(); + let mut b = receiver.try_recv().unwrap(); + assert_ne!(a.parent_call_id, b.parent_call_id); + assert!( + (a.parent_call_id == ToolCallId::new("parent-a") + && b.parent_call_id == ToolCallId::new("parent-b")) + || (a.parent_call_id == ToolCallId::new("parent-b") + && b.parent_call_id == ToolCallId::new("parent-a")) + ); + assert_ne!(a.incarnation, b.incarnation); + assert_eq!(a.source_digest, b.source_digest); + assert!(!a.healed); + for progress in [&mut a, &mut b] { + let (events, end) = collect_progress(progress); + assert_eq!(end, crate::RunletProgressEnd::Succeeded); + let mut first_succeeded = None; + let mut second_running = None; + for event in events { + if let runlet::ProgressChange::NodeUpdated(node) = event.change { + if node.span.start == script.find("echo").unwrap() + && node.state == runlet::ProgressState::Succeeded + { + first_succeeded = Some(event.sequence); + } + if node.span.start == script.rfind("echo").unwrap() + && node.state == runlet::ProgressState::Running + { + second_running = Some(event.sequence); + } + } + } + assert!(first_succeeded.unwrap() < second_running.unwrap()); + } +} + +#[tokio::test] +async fn progress_overflow_and_dropped_sink_do_not_fail_execution() { + for drop_sink in [false, true] { + let (backend, mut receiver) = progress_backend(1); + if drop_sink { + receiver.close(); + } + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); + let executor: Arc = Arc::new(BasicToolExecutor::from_registry( + ToolRegistry::new().with(compose), + )); + let owned = owned_context(executor.clone(), Arc::new(AllowAllPermissions)); + let mut ctx = owned.borrowed(); + let result = tokio::time::timeout( + Duration::from_secs(5), + executor.execute(request("return 1 + 2", Value::Null), &mut ctx), + ) + .await + .unwrap(); + assert!(matches!(result, ToolExecutionOutcome::Completed(_))); + if !drop_sink { + let mut progress = receiver.try_recv().unwrap(); + assert_eq!( + collect_progress(&mut progress).1, + crate::RunletProgressEnd::Lagged + ); + } + } +} + +#[tokio::test] +async fn progress_healing_identifies_compiled_not_submitted_source() { + let (backend, mut receiver) = progress_backend(1024); + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); + let executor: Arc = Arc::new(BasicToolExecutor::from_registry( + ToolRegistry::new().with(compose), + )); + let owned = owned_context(executor.clone(), Arc::new(AllowAllPermissions)); + let mut ctx = owned.borrowed(); + let script = "if true { x = 1 }\nreturn 2"; + assert!(matches!( + executor + .execute(request(script, Value::Null), &mut ctx) + .await, + ToolExecutionOutcome::Completed(_) + )); + let mut progress = receiver.try_recv().unwrap(); + assert!(progress.healed); + let healed = runlet::heal(script).unwrap(); + let runtime = runlet::Runtime::builder().with_prelude().build().unwrap(); + assert_eq!( + progress.source_digest, + runtime.compile(&healed.source).unwrap().source_digest + ); + assert_eq!( + collect_progress(&mut progress).1, + crate::RunletProgressEnd::Succeeded + ); +} + +#[derive(Clone)] +struct ProgressGate { + spec: ToolSpec, + entered: Arc, + release: Arc, + finished: Arc, +} + +impl ProgressGate { + fn new() -> Self { + Self { + spec: ToolSpec::new( + "progress_gate", + "hold at a real child boundary", + json!({"type":"object"}), + ), + entered: Arc::new(tokio::sync::Notify::new()), + release: Arc::new(tokio::sync::Notify::new()), + finished: Arc::new(tokio::sync::Notify::new()), + } + } +} + +#[async_trait::async_trait] +impl Tool for ProgressGate { + fn spec(&self) -> &ToolSpec { + &self.spec + } + async fn invoke( + &self, + request: ToolRequest, + _ctx: &mut ToolContext<'_>, + ) -> Result { + self.entered.notify_one(); + self.release.notified().await; + self.finished.notify_one(); + Ok(ToolResult::new(ToolResultPart::success( + request.call_id, + ToolOutput::structured(json!(null)), + ))) + } +} + +#[tokio::test] +async fn progress_future_drop_invalidates_and_consumer_drop_is_harmless() { + for disposition in 0..3 { + let abort = disposition == 2; + let (backend, mut receiver) = progress_backend(1024); + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); + let gate = ProgressGate::new(); + let executor: Arc = Arc::new(BasicToolExecutor::from_registry( + ToolRegistry::new().with(compose).with(gate.clone()), + )); + let task = tokio::spawn(async move { + let owned = owned_context(executor.clone(), Arc::new(AllowAllPermissions)); + let mut ctx = owned.borrowed(); + executor + .execute(request("return progress_gate({})", Value::Null), &mut ctx) + .await + }); + tokio::time::timeout(Duration::from_secs(5), gate.entered.notified()) + .await + .unwrap(); + let mut progress = receiver.try_recv().unwrap(); + assert!(matches!(progress.try_recv(), Ok(Some(_)))); + if abort { + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + assert_eq!( + progress.try_recv(), + Err(crate::RunletProgressEnd::Incomplete) + ); + gate.release.notify_one(); + // Explicitly release existing blocking execution, not an observer worker. + tokio::time::timeout(Duration::from_secs(5), gate.finished.notified()) + .await + .unwrap(); + } else { + if disposition == 1 { + assert!( + std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || { + let _consumer = progress; + panic!("consumer failure outside the executor"); + })) + .is_err() + ); + } else { + drop(progress); + } + gate.release.notify_one(); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(5), task) + .await + .unwrap() + .unwrap(), + ToolExecutionOutcome::Completed(_) + )); + } + } +} + +#[tokio::test] +async fn progress_external_cancellation_invalidates_before_runtime_finishes() { + let (backend, mut receiver) = progress_backend(1024); + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); + let gate = ProgressGate::new(); + let controller = agentkit_core::CancellationController::new(); + let cancellation = controller.handle().checkpoint(); + let executor: Arc = Arc::new(BasicToolExecutor::from_registry( + ToolRegistry::new().with(compose).with(gate.clone()), + )); + let task = tokio::spawn(async move { + let mut owned = owned_context(executor.clone(), Arc::new(AllowAllPermissions)); + owned.cancellation = Some(cancellation); + let mut ctx = owned.borrowed(); + executor + .execute(request("return progress_gate({})", Value::Null), &mut ctx) + .await + }); + tokio::time::timeout(Duration::from_secs(5), gate.entered.notified()) + .await + .unwrap(); + let mut progress = receiver.try_recv().unwrap(); + controller.interrupt(); + assert_eq!( + progress.try_recv(), + Err(crate::RunletProgressEnd::Incomplete) + ); + gate.release.notify_one(); + let _ = tokio::time::timeout(Duration::from_secs(5), task) + .await + .unwrap() + .unwrap(); + assert_eq!( + progress.try_recv(), + Err(crate::RunletProgressEnd::Incomplete) + ); +} + +#[tokio::test] +async fn progress_full_host_sink_is_unobserved_and_events_are_payload_free() { + let (backend, mut receiver) = progress_backend(1024); + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); + let executor: Arc = Arc::new(BasicToolExecutor::from_registry( + ToolRegistry::new().with(compose).with(EchoTool::new()), + )); + for i in 0..5 { + let owned = owned_context(executor.clone(), Arc::new(AllowAllPermissions)); + let mut ctx = owned.borrowed(); + let mut req = request( + "return echo({ secret: input.secret })", + json!({"secret":"private-input-output-marker"}), + ); + req.call_id = ToolCallId::new(format!("bounded-parent-{i}")); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(5), executor.execute(req, &mut ctx)) + .await + .unwrap(), + ToolExecutionOutcome::Completed(_) + )); + } + for _ in 0..4 { + let mut progress = receiver.try_recv().unwrap(); + let (events, end) = collect_progress(&mut progress); + assert_eq!(end, crate::RunletProgressEnd::Succeeded); + let serialized = serde_json::to_string(&events).unwrap(); + assert!(!serialized.contains("private-input-output-marker")); + assert!(!serialized.contains("secret")); + assert!(!serialized.contains("echo")); + } + assert!(receiver.try_recv().is_err()); +} + +#[tokio::test] +async fn progress_failure_omits_runtime_error_text() { + let (backend, mut receiver) = progress_backend(1024); + let compose = ComposeTool::new(ComposeConfig::default()).with_backend(backend); + let executor: Arc = Arc::new(BasicToolExecutor::from_registry( + ToolRegistry::new().with(compose), + )); + let owned = owned_context(executor.clone(), Arc::new(AllowAllPermissions)); + let mut ctx = owned.borrowed(); + let outcome = executor + .execute( + request( + "return fail(\"PRIVATE_CODE\", \"private-error-marker\")", + Value::Null, + ), + &mut ctx, + ) + .await; + assert!(!matches!(outcome, ToolExecutionOutcome::Completed(_))); + let mut progress = receiver.try_recv().unwrap(); + let (events, end) = collect_progress(&mut progress); + assert_eq!(end, crate::RunletProgressEnd::Failed); + let serialized = serde_json::to_string(&events).unwrap(); + assert!(!serialized.contains("PRIVATE_CODE")); + assert!(!serialized.contains("private-error-marker")); +} From e941120c8fb683b0f4e032ab9b0531bb45a61528 Mon Sep 17 00:00:00 2001 From: daniel Date: Tue, 8 Sep 2026 08:55:40 +0100 Subject: [PATCH 2/2] chore(compose): prepare version 0.10.11 --- Cargo.lock | 2 +- crates/agentkit-tool-compose/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3fc7739..b260653 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -422,7 +422,7 @@ dependencies = [ [[package]] name = "agentkit-tool-compose" -version = "0.10.10" +version = "0.10.11" dependencies = [ "agentkit-core", "agentkit-tools-core", diff --git a/crates/agentkit-tool-compose/Cargo.toml b/crates/agentkit-tool-compose/Cargo.toml index bd1439d..4dd9366 100644 --- a/crates/agentkit-tool-compose/Cargo.toml +++ b/crates/agentkit-tool-compose/Cargo.toml @@ -4,7 +4,7 @@ homepage.workspace = true name = "agentkit-tool-compose" readme = "README.md" repository.workspace = true -version = "0.10.10" +version = "0.10.11" edition.workspace = true license.workspace = true rust-version.workspace = true