From ec52906b676c05c19654dd4343e3de6e398fa5aa Mon Sep 17 00:00:00 2001 From: Lifei Zhou Date: Fri, 9 Oct 2026 13:34:02 +1100 Subject: [PATCH] make Emitter sync --- crates/goose-agent/src/inference.rs | 21 ++++----- crates/goose-agent/src/machine.rs | 2 +- crates/goose-agent/src/operation.rs | 6 +-- crates/goose-agent/src/tool.rs | 4 +- .../agents/state_machine/ops_bang_shell.rs | 2 +- .../agents/state_machine/ops_compaction.rs | 35 ++++++--------- .../src/agents/state_machine/ops_doctor.rs | 4 +- .../state_machine/ops_empty_response.rs | 3 +- .../state_machine/ops_foreground_subagent.rs | 14 +++--- .../src/agents/state_machine/ops_maxturns.rs | 2 +- .../src/agents/state_machine/ops_recipe.rs | 38 ++++++++-------- .../src/agents/state_machine/ops_retry.rs | 16 +++---- .../src/agents/state_machine/ops_skills.rs | 13 +++--- .../src/agents/state_machine/ops_status.rs | 5 +-- .../src/agents/state_machine/ops_steer.rs | 2 +- .../src/agents/state_machine/ops_stop_hook.rs | 4 +- .../agents/state_machine/ops_toolcalling.rs | 45 ++++++++----------- .../agents/state_machine/ops_unknown_tool.rs | 2 +- .../goose/src/agents/state_machine/session.rs | 12 ++--- crates/goose/tests/state_machine_api.rs | 4 +- 20 files changed, 100 insertions(+), 134 deletions(-) diff --git a/crates/goose-agent/src/inference.rs b/crates/goose-agent/src/inference.rs index bce9dff0d3..92abd1405f 100644 --- a/crates/goose-agent/src/inference.rs +++ b/crates/goose-agent/src/inference.rs @@ -335,21 +335,16 @@ impl<'a, S: MaybeSync, E: InferenceEffect> InferenceRunner<'a, S, E> { effects } - async fn emit_message(&self, message: Message, emit: &Emitter) { + fn emit_message(&self, message: Message, emit: &Emitter) { let message = message.with_generated_id_if_missing(); self.output().accumulator.push(message.clone()); - emit.message(message).await; + emit.message(message); } - async fn error_outcome( - &self, - err: &ProviderError, - emit: &Emitter, - ) -> Result> { + fn error_outcome(&self, err: &ProviderError, emit: &Emitter) -> Result> { tracing::Span::current().record("error.type", err.telemetry_type()); tracing::error!("LLM provider error: {err}"); - self.emit_message(Message::from_provider_error(err), emit) - .await; + self.emit_message(Message::from_provider_error(err), emit); applied(self.take_output()) } } @@ -453,7 +448,7 @@ impl Inference for InferenceRunner<'_, S let mut stream = match stream { Ok(stream) => stream, - Err(err) => return self.error_outcome(&err, emit).await, + Err(err) => return self.error_outcome(&err, emit), }; let requested_model = self.model_config.model_name.clone(); @@ -475,7 +470,7 @@ impl Inference for InferenceRunner<'_, S while let Some(result) = stream.next().await { let (msg_opt, usage_opt) = match result { Ok(chunk) => chunk, - Err(err) => return self.error_outcome(&err, emit).await, + Err(err) => return self.error_outcome(&err, emit), }; if let Some(usage) = usage_opt { record_chat_usage(&tracing::Span::current(), &usage); @@ -497,7 +492,7 @@ impl Inference for InferenceRunner<'_, S if chunk.content.is_empty() && !chunk.metadata.output_token_limit_reached { self.output().accumulator.push(chunk); } else { - self.emit_message(chunk, emit).await; + self.emit_message(chunk, emit); } } } @@ -547,7 +542,7 @@ impl Inference for InferenceRunner<'_, S output.accumulator.clear(); message }; - self.emit_message(message, emit).await; + self.emit_message(message, emit); } applied(self.take_output()) } diff --git a/crates/goose-agent/src/machine.rs b/crates/goose-agent/src/machine.rs index ba5cc0dfd6..4750377248 100644 --- a/crates/goose-agent/src/machine.rs +++ b/crates/goose-agent/src/machine.rs @@ -236,7 +236,7 @@ where .and_then(|conversation| messages_since_kickoff(conversation).ok()) .and_then(interrupted_response); if let Some(response) = unanswered { - let effects = vec![E::from(emit.message(response).await)]; + let effects = vec![E::from(emit.message(response))]; session = self .save(runtime, session_id, session, effects, emit) .await?; diff --git a/crates/goose-agent/src/operation.rs b/crates/goose-agent/src/operation.rs index f6d4645d0c..53360da06b 100644 --- a/crates/goose-agent/src/operation.rs +++ b/crates/goose-agent/src/operation.rs @@ -260,13 +260,13 @@ impl Emitter { Self { tx, cancel } } - pub async fn emit(&self, event: AgentEvent) { + pub fn emit(&self, event: AgentEvent) { let _ = self.tx.send(event); } - pub async fn message(&self, message: Message) -> Message { + pub fn message(&self, message: Message) -> Message { let message = message.with_generated_id_if_missing(); - self.emit(AgentEvent::Message(message.clone())).await; + self.emit(AgentEvent::Message(message.clone())); message } diff --git a/crates/goose-agent/src/tool.rs b/crates/goose-agent/src/tool.rs index 0f91325930..5579d82b08 100644 --- a/crates/goose-agent/src/tool.rs +++ b/crates/goose-agent/src/tool.rs @@ -311,7 +311,7 @@ where ) -> Vec { let response = self.response.lock().unwrap().take(); match response { - Some(response) => vec![E::from(emit.message(response).await)], + Some(response) => vec![E::from(emit.message(response))], None => Vec::new(), } } @@ -391,7 +391,7 @@ where .unwrap() .take() .expect("tool responses retained"); - let message = emit.message(message).await; + let message = emit.message(message); applied([E::from(message)]) } } diff --git a/crates/goose/src/agents/state_machine/ops_bang_shell.rs b/crates/goose/src/agents/state_machine/ops_bang_shell.rs index 7c4f05383e..76b867cc96 100644 --- a/crates/goose/src/agents/state_machine/ops_bang_shell.rs +++ b/crates/goose/src/agents/state_machine/ops_bang_shell.rs @@ -68,7 +68,7 @@ impl Operation for BangShellOperation { ); let request = Message::assistant() .with_tool_request(format!("bang_shell_{}", uuid::Uuid::now_v7()), Ok(call)); - let request = emit.message(request).await; + let request = emit.message(request); applied([request.into()]) } diff --git a/crates/goose/src/agents/state_machine/ops_compaction.rs b/crates/goose/src/agents/state_machine/ops_compaction.rs index 68a48a7bd4..e87ab9ffa8 100644 --- a/crates/goose/src/agents/state_machine/ops_compaction.rs +++ b/crates/goose/src/agents/state_machine/ops_compaction.rs @@ -98,7 +98,7 @@ impl CompactionOperation { } } - async fn command_error( + fn command_error( conversation: &Conversation, message: String, emit: &Emitter, @@ -115,8 +115,8 @@ impl CompactionOperation { let response = Message::assistant() .with_text(message) .with_visibility(true, false); - emit.message(command).await; - let response = emit.message(response).await; + emit.message(command); + let response = emit.message(response); yielded_with([ ConversationEffect::SetMessageVisibility { message_id, @@ -128,10 +128,7 @@ impl CompactionOperation { ]) } - async fn clear( - conversation: &Conversation, - emit: &Emitter, - ) -> Result> { + fn clear(conversation: &Conversation, emit: &Emitter) -> Result> { let command = messages_since_kickoff(conversation)? .first() .cloned() @@ -140,8 +137,8 @@ impl CompactionOperation { let response = Message::assistant() .with_text("Conversation cleared") .with_visibility(true, false); - let command = emit.message(command).await; - let response = emit.message(response).await; + let command = emit.message(command); + let response = emit.message(response); yielded_with([ Conversation::default().into(), command.into(), @@ -164,7 +161,7 @@ impl Operation for CompactionOperation { emit: &Emitter, ) -> Result> { match command.command { - "clear" => return Self::clear(conversation, emit).await, + "clear" => return Self::clear(conversation, emit), "compact" => {} _ => return not_applicable(), } @@ -188,7 +185,7 @@ impl Operation for CompactionOperation { Ok(result) => result, Err(error) => { span.record("error.type", "compaction_error"); - return Self::command_error(conversation, error.to_string(), emit).await; + return Self::command_error(conversation, error.to_string(), emit); } }; let compacted = result.conversation; @@ -203,8 +200,8 @@ impl Operation for CompactionOperation { let response = Message::assistant() .with_text("Compaction complete") .with_visibility(true, false); - emit.message(command).await; - let response = emit.message(response).await; + emit.message(command); + let response = emit.message(response); yielded_with([ GooseEffect::CompactConversation { conversation: compacted, @@ -296,13 +293,11 @@ impl Operation for CompactionOperation { "Exceeded auto-compact threshold of {threshold_percentage}%. \ Performing auto-compaction..." ), - )) - .await; + )); emit.message(Message::assistant().with_system_notification( SystemNotificationType::ThinkingMessage, COMPACTION_THINKING_TEXT, - )) - .await; + )); let span = chat_span( self.provider.as_ref(), @@ -327,8 +322,7 @@ impl Operation for CompactionOperation { emit.message(Message::assistant().with_system_notification( SystemNotificationType::InlineMessage, "Compaction complete", - )) - .await; + )); applied([GooseEffect::CompactConversation { conversation: compacted, usage: Some(usage), @@ -339,8 +333,7 @@ impl Operation for CompactionOperation { emit.message(Message::assistant().with_text(format!( "Ran into this error trying to compact: {e}.\n\n\ Please try again or create a new session" - ))) - .await; + ))); yielded() } } diff --git a/crates/goose/src/agents/state_machine/ops_doctor.rs b/crates/goose/src/agents/state_machine/ops_doctor.rs index 07d82d6a64..8fdc3c7bfc 100644 --- a/crates/goose/src/agents/state_machine/ops_doctor.rs +++ b/crates/goose/src/agents/state_machine/ops_doctor.rs @@ -57,8 +57,8 @@ impl Operation for DoctorOperation { if result.role == Role::Assistant { let command_message = command_message.with_visibility(true, false); let result = result.with_visibility(true, false); - emit.message(command_message).await; - let result = emit.message(result).await; + emit.message(command_message); + let result = emit.message(result); return yielded_with([ ConversationEffect::SetMessageVisibility { message_id, diff --git a/crates/goose/src/agents/state_machine/ops_empty_response.rs b/crates/goose/src/agents/state_machine/ops_empty_response.rs index 2b970d3c6e..b7a5bcd129 100644 --- a/crates/goose/src/agents/state_machine/ops_empty_response.rs +++ b/crates/goose/src/agents/state_machine/ops_empty_response.rs @@ -44,8 +44,7 @@ impl Operation for EmptyResponseOperation { .id .clone() .ok_or_else(|| anyhow!("Persisted empty-response fallback has no id"))?; - emit.message(fallback.clone().with_visibility(true, true)) - .await; + emit.message(fallback.clone().with_visibility(true, true)); applied([ConversationEffect::SetMessageVisibility { message_id, user_visible: true, diff --git a/crates/goose/src/agents/state_machine/ops_foreground_subagent.rs b/crates/goose/src/agents/state_machine/ops_foreground_subagent.rs index 5f4337d37a..0c759819ed 100644 --- a/crates/goose/src/agents/state_machine/ops_foreground_subagent.rs +++ b/crates/goose/src/agents/state_machine/ops_foreground_subagent.rs @@ -98,7 +98,7 @@ fn readable_output(output: &str) -> String { } } -async fn subagent_result_message( +fn subagent_result_message( subagent_id: &str, outcome: SubagentOutcome, others_waiting: bool, @@ -123,7 +123,7 @@ async fn subagent_result_message( ), SubagentOutcome::Cancelled => return None, }; - emit.message(inline_notice(notice)).await; + emit.message(inline_notice(notice)); let mut message = Message::user().with_text(text).with_visibility(false, true); message .metadata @@ -207,8 +207,7 @@ impl Operation for ForegroundSubagentOperation { let Some(subagent_id) = stopped.subagent_ids.remove(&task_id) else { continue; }; - if let Some(message) = subagent_result_message(&subagent_id, outcome, true, emit).await - { + if let Some(message) = subagent_result_message(&subagent_id, outcome, true, emit) { effects.push(message); delivered.insert(subagent_id); } @@ -252,14 +251,13 @@ impl Operation for ForegroundSubagentOperation { { SubagentStart::HasOutcome(outcome) => { if let Some(message) = - subagent_result_message(subagent_id, outcome, pending.len() > 1, emit).await + subagent_result_message(subagent_id, outcome, pending.len() > 1, emit) { return applied([message]); } } SubagentStart::Started { task, run } => { - emit.message(inline_notice(start_notice(subagent_id, task.as_deref()))) - .await; + emit.message(inline_notice(start_notice(subagent_id, task.as_deref()))); let handle = running.tasks.spawn(run.in_current_span()); running .subagent_ids @@ -281,7 +279,7 @@ impl Operation for ForegroundSubagentOperation { .subagent_ids .remove(&task_id) .ok_or_else(|| anyhow!("Unknown foreground subagent task {task_id}"))?; - match subagent_result_message(&subagent_id, outcome, pending.len() > 1, emit).await { + match subagent_result_message(&subagent_id, outcome, pending.len() > 1, emit) { Some(message) => applied([message]), None => not_applicable(), } diff --git a/crates/goose/src/agents/state_machine/ops_maxturns.rs b/crates/goose/src/agents/state_machine/ops_maxturns.rs index e1c0fcbf5a..cf0f03a90f 100644 --- a/crates/goose/src/agents/state_machine/ops_maxturns.rs +++ b/crates/goose/src/agents/state_machine/ops_maxturns.rs @@ -62,7 +62,7 @@ impl Operation for MaxTurnsOperation { } let message = Message::assistant().with_text(MAX_TURNS_MESSAGE); - let message = emit.message(message).await; + let message = emit.message(message); yielded_with([message.into()]) } } diff --git a/crates/goose/src/agents/state_machine/ops_recipe.rs b/crates/goose/src/agents/state_machine/ops_recipe.rs index b51fb3f73e..b1d013ce42 100644 --- a/crates/goose/src/agents/state_machine/ops_recipe.rs +++ b/crates/goose/src/agents/state_machine/ops_recipe.rs @@ -67,7 +67,7 @@ impl RecipeOperation { self.response.lock().unwrap().take() } - async fn command_error( + fn command_error( &self, conversation: &Conversation, message: String, @@ -85,8 +85,8 @@ impl RecipeOperation { let response = Message::assistant() .with_text(message) .with_visibility(true, false); - emit.message(command).await; - let response = emit.message(response).await; + emit.message(command); + let response = emit.message(response); yielded_with([ ConversationEffect::SetMessageVisibility { message_id, @@ -112,7 +112,7 @@ impl Operation for RecipeOperation { emit: &Emitter, ) -> Vec { match self.take_response() { - Some(response) => vec![emit.message(response).await.into()], + Some(response) => vec![emit.message(response).into()], None => Vec::new(), } } @@ -130,14 +130,16 @@ impl Operation for RecipeOperation { ) { Ok(Some(recipe)) => recipe, Ok(None) => return not_applicable(), - Err(error) => return self.command_error(conversation, error, emit).await, + Err(error) => return self.command_error(conversation, error, emit), }; if let Some(response) = recipe.response.clone() { if let Err(error) = FinalOutputTool::try_new(response) { - return self - .command_error(conversation, format!("Recipe is not valid: {error}"), emit) - .await; + return self.command_error( + conversation, + format!("Recipe is not valid: {error}"), + emit, + ); } } @@ -193,13 +195,11 @@ impl Operation for RecipeOperation { }; if !self.provider.supports_builtin_tools() { - return self - .command_error( - conversation, - structured_output_unsupported_message(self.provider.get_name()), - emit, - ) - .await; + return self.command_error( + conversation, + structured_output_unsupported_message(self.provider.get_name()), + emit, + ); } let messages = messages_since_kickoff(conversation)?; @@ -222,7 +222,7 @@ impl Operation for RecipeOperation { )])), request.metadata.as_ref(), ); - let response = emit.message(response).await; + let response = emit.message(response); return applied([response.into()]); } if FinalOutputTool::has_unanswered_siblings(messages, &request.id) { @@ -297,14 +297,14 @@ impl Operation for RecipeOperation { .await; } let response = self.take_response().expect("recipe response held"); - let response = emit.message(response).await; + let response = emit.message(response); return applied([response.into()]); } if let Some(output) = FinalOutputTool::successful_output(messages) { if last_effective_role(messages)? == EffectiveRole::Tool { let message = Message::assistant().with_text(output); - let message = emit.message(message).await; + let message = emit.message(message); return applied([message.into()]); } return not_applicable(); @@ -314,7 +314,7 @@ impl Operation for RecipeOperation { let message = Message::user() .with_text(FINAL_OUTPUT_CONTINUATION_MESSAGE) .agent_only(); - let message = emit.message(message).await; + let message = emit.message(message); return applied([message.into()]); } diff --git a/crates/goose/src/agents/state_machine/ops_retry.rs b/crates/goose/src/agents/state_machine/ops_retry.rs index 6e472f97eb..a90c69a724 100644 --- a/crates/goose/src/agents/state_machine/ops_retry.rs +++ b/crates/goose/src/agents/state_machine/ops_retry.rs @@ -139,8 +139,8 @@ impl Operation for RetryOperation { .ok_or_else(|| anyhow!("Persisted slash command message has no id"))?; let command_message = command_message.with_visibility(true, false); let response = response.with_visibility(true, false); - emit.message(command_message).await; - let response = emit.message(response).await; + emit.message(command_message); + let response = emit.message(response); let mut effects = vec![ ConversationEffect::SetMessageVisibility { @@ -195,8 +195,7 @@ impl Operation for RetryOperation { emit.message(Message::assistant().with_system_notification( SystemNotificationType::InlineMessage, format!("Goal: {goal}"), - )) - .await; + )); return applied([message.into()]); } } @@ -213,8 +212,7 @@ impl Operation for RetryOperation { emit.message(Message::assistant().with_system_notification( SystemNotificationType::InlineMessage, format!("Grind: {grind}"), - )) - .await; + )); return applied([message.into()]); } @@ -235,7 +233,7 @@ impl Operation for RetryOperation { let success = match success { Ok(success) => success, Err(error) => { - let message = emit.message(retry_error(&error.to_string())).await; + let message = emit.message(retry_error(&error.to_string())); return applied([message.into()]); } }; @@ -257,7 +255,7 @@ impl Operation for RetryOperation { "retry_max_exceeded", &format!("Max retries ({}) exceeded", retry_config.max_retries), ); - let message = emit.message(message).await; + let message = emit.message(message); return applied([message.into()]); } @@ -267,7 +265,7 @@ impl Operation for RetryOperation { .map(Duration::from_secs) .unwrap_or(self.on_failure_timeout); if let Err(error) = execute_on_failure_command_with_timeout(command, timeout).await { - let message = emit.message(retry_error(&error.to_string())).await; + let message = emit.message(retry_error(&error.to_string())); return applied([message.into()]); } } diff --git a/crates/goose/src/agents/state_machine/ops_skills.rs b/crates/goose/src/agents/state_machine/ops_skills.rs index 97c51c74b6..0faf2d7e7c 100644 --- a/crates/goose/src/agents/state_machine/ops_skills.rs +++ b/crates/goose/src/agents/state_machine/ops_skills.rs @@ -219,7 +219,7 @@ impl SkillOperation { Some(session) } - async fn command_response( + fn command_response( conversation: &Conversation, message: String, emit: &Emitter, @@ -236,8 +236,8 @@ impl SkillOperation { let response = Message::assistant() .with_text(message) .with_visibility(true, false); - emit.message(command).await; - let response = emit.message(response).await; + emit.message(command); + let response = emit.message(response); yielded_with([ ConversationEffect::SetMessageVisibility { message_id, @@ -270,8 +270,7 @@ impl Operation for SkillOperation { &session.working_dir, )), emit, - ) - .await; + ); } let prompt = match crate::slash_commands::skill_slash_command::resolve_command( @@ -281,7 +280,7 @@ impl Operation for SkillOperation { ) { Ok(Some(prompt)) => prompt, Ok(None) => return not_applicable(), - Err(error) => return Self::command_response(conversation, error, emit).await, + Err(error) => return Self::command_response(conversation, error, emit), }; let command_message = messages_since_kickoff(conversation)? .first() @@ -426,7 +425,7 @@ impl Operation for SkillOperation { }; response.add_tool_response_with_metadata(request.id, result, request.metadata.as_ref()); } - let response = emit.message(response).await; + let response = emit.message(response); applied([response.into()]) } } diff --git a/crates/goose/src/agents/state_machine/ops_status.rs b/crates/goose/src/agents/state_machine/ops_status.rs index b7241695d5..e0937a26bc 100644 --- a/crates/goose/src/agents/state_machine/ops_status.rs +++ b/crates/goose/src/agents/state_machine/ops_status.rs @@ -69,9 +69,8 @@ impl Operation for StatusOperation { .id .clone() .ok_or_else(|| anyhow!("Persisted slash command message has no id"))?; - emit.message(command_message.with_visibility(true, false)) - .await; - let response = emit.message(response).await; + emit.message(command_message.with_visibility(true, false)); + let response = emit.message(response); yielded_with([ ConversationEffect::SetMessageVisibility { message_id, diff --git a/crates/goose/src/agents/state_machine/ops_steer.rs b/crates/goose/src/agents/state_machine/ops_steer.rs index 27e6eeabb8..51513132af 100644 --- a/crates/goose/src/agents/state_machine/ops_steer.rs +++ b/crates/goose/src/agents/state_machine/ops_steer.rs @@ -83,7 +83,7 @@ impl Operation for SteerOperation { let mut drained = Vec::with_capacity(pending.len()); for message in pending { - drained.push(emit.message(message).await); + drained.push(emit.message(message)); } *self.drained.lock().unwrap() = drained.clone(); for message in drained { diff --git a/crates/goose/src/agents/state_machine/ops_stop_hook.rs b/crates/goose/src/agents/state_machine/ops_stop_hook.rs index 3ee99c866c..bc03385474 100644 --- a/crates/goose/src/agents/state_machine/ops_stop_hook.rs +++ b/crates/goose/src/agents/state_machine/ops_stop_hook.rs @@ -106,12 +106,12 @@ impl Operation for StopHookOperation { + 1; if blocks > self.block_cap { let warning = block_cap_warning(&plugin, self.block_cap); - let warning = emit.message(warning).await; + let warning = emit.message(warning); yielded_with([warning.into()]) } else { let mut denial = denial_context_message(&plugin, &reason); self.set_message_meta(&mut denial, DENIED, serde_json::json!(true)); - emit.message(denial_notification(&plugin)).await; + emit.message(denial_notification(&plugin)); applied([denial.into()]) } } diff --git a/crates/goose/src/agents/state_machine/ops_toolcalling.rs b/crates/goose/src/agents/state_machine/ops_toolcalling.rs index 035ead7901..0c4821383d 100644 --- a/crates/goose/src/agents/state_machine/ops_toolcalling.rs +++ b/crates/goose/src/agents/state_machine/ops_toolcalling.rs @@ -475,7 +475,7 @@ impl ToolExecutionOperation { .await } - async fn command_response( + fn command_response( conversation: &Conversation, message: String, emit: &Emitter, @@ -492,8 +492,8 @@ impl ToolExecutionOperation { let response = Message::assistant() .with_text(message) .with_visibility(true, false); - emit.message(command).await; - let response = emit.message(response).await; + emit.message(command); + let response = emit.message(response); yielded_with([ ConversationEffect::SetMessageVisibility { message_id, @@ -524,8 +524,7 @@ impl ToolExecutionOperation { conversation, format!("Extension '{filter}' not found"), emit, - ) - .await; + ); } } @@ -546,7 +545,7 @@ impl ToolExecutionOperation { output.push('\n'); } } - Self::command_response(conversation, output, emit).await + Self::command_response(conversation, output, emit) } async fn run_prompt( @@ -562,8 +561,7 @@ impl ToolExecutionOperation { conversation, "Prompt name argument is required".to_string(), emit, - ) - .await; + ); }; let prompts = self .lease(session) @@ -581,8 +579,7 @@ impl ToolExecutionOperation { conversation, format!("Prompt '{prompt_name}' not found"), emit, - ) - .await; + ); }; if params.get(1) == Some(&"--info") { @@ -601,7 +598,7 @@ impl ToolExecutionOperation { output.push('\n'); } } - return Self::command_response(conversation, output, emit).await; + return Self::command_response(conversation, output, emit); } let arguments: HashMap<_, _> = params @@ -623,7 +620,7 @@ impl ToolExecutionOperation { { Ok(result) => result, Err(error) => { - return Self::command_response(conversation, error.to_string(), emit).await; + return Self::command_response(conversation, error.to_string(), emit); } }; @@ -655,8 +652,7 @@ impl ToolExecutionOperation { message.role ), emit, - ) - .await; + ); } effects.push(message.with_visibility(false, true).into()); } @@ -665,8 +661,7 @@ impl ToolExecutionOperation { conversation, format!("Prompt '{prompt_name}' returned no messages"), emit, - ) - .await; + ); } applied(effects) } @@ -823,8 +818,7 @@ impl Operation for ToolExecutionOperation { ) -> Vec { let (actions, response) = self.take_batch(); if let Some(response) = &response { - emit.emit(AgentEvent::Message(response.user_visible_content())) - .await; + emit.emit(AgentEvent::Message(response.user_visible_content())); } actions .into_iter() @@ -975,7 +969,7 @@ impl Operation for ToolExecutionOperation { request.metadata.as_ref(), ); } - let response = emit.message(response).await; + let response = emit.message(response); return applied([response.into()]); }; @@ -1015,7 +1009,7 @@ impl Operation for ToolExecutionOperation { request.metadata.as_ref(), ); } - let response = emit.message(response).await; + let response = emit.message(response); return applied([response.into()]); } @@ -1088,8 +1082,7 @@ impl Operation for ToolExecutionOperation { emit.emit(AgentEvent::McpNotification(( request_id.clone(), notification, - ))) - .await; + ))); } } let metadata = requests @@ -1102,21 +1095,19 @@ impl Operation for ToolExecutionOperation { .record(&request_id, output, metadata); } ToolStreamItem::Message(msg) => { - emit.emit(AgentEvent::McpNotification((request_id, msg))) - .await; + emit.emit(AgentEvent::McpNotification((request_id, msg))); } ToolStreamItem::ActionRequired(msg) => { let msg = msg.with_generated_id_if_missing(); self.batch.lock().unwrap().actions.push(msg.clone()); - emit.message(msg).await; + emit.message(msg); } } } let (actions, response) = self.take_batch(); let response = response.ok_or_else(|| anyhow!("tool batch produced no responses"))?; - emit.emit(AgentEvent::Message(response.user_visible_content())) - .await; + emit.emit(AgentEvent::Message(response.user_visible_content())); applied(actions.into_iter().chain([response]).map(GooseEffect::from)) } } diff --git a/crates/goose/src/agents/state_machine/ops_unknown_tool.rs b/crates/goose/src/agents/state_machine/ops_unknown_tool.rs index 313b95d085..db26946839 100644 --- a/crates/goose/src/agents/state_machine/ops_unknown_tool.rs +++ b/crates/goose/src/agents/state_machine/ops_unknown_tool.rs @@ -192,7 +192,7 @@ impl Operation for UnknownToolOperation { response.add_tool_response_with_metadata(request.id, result, metadata.as_ref()); } - let response = emit.message(response).await; + let response = emit.message(response); applied([response.into()]) } } diff --git a/crates/goose/src/agents/state_machine/session.rs b/crates/goose/src/agents/state_machine/session.rs index 7ae1823b82..32af60d831 100644 --- a/crates/goose/src/agents/state_machine/session.rs +++ b/crates/goose/src/agents/state_machine/session.rs @@ -121,7 +121,7 @@ impl EffectHandler for SessionManager { GooseEffect::Conversation(ConversationEffect::AppendMessage(message)) => { if contains_tool_confirmation_request(message) { // Responses can arrive immediately, so publish only after the persistence pass. - emit.emit(AgentEvent::Message(message.clone())).await; + emit.emit(AgentEvent::Message(message.clone())); } if let Some(usage) = message .metadata @@ -133,8 +133,7 @@ impl EffectHandler for SessionManager { emit.emit(AgentEvent::MessageUsage { message_id: message.id.clone(), usage, - }) - .await; + }); } } GooseEffect::Conversation(ConversationEffect::ReplaceConversation( @@ -144,12 +143,9 @@ impl EffectHandler for SessionManager { emit.emit(AgentEvent::HistoryReplaced(replaced_history( conversation, &effects[index + 1..], - ))) - .await; - } - GooseEffect::RecordUsage(usage) => { - emit.emit(AgentEvent::Usage(usage.clone())).await + ))); } + GooseEffect::RecordUsage(usage) => emit.emit(AgentEvent::Usage(usage.clone())), _ => {} } } diff --git a/crates/goose/tests/state_machine_api.rs b/crates/goose/tests/state_machine_api.rs index 347db6c48a..b4693eda38 100644 --- a/crates/goose/tests/state_machine_api.rs +++ b/crates/goose/tests/state_machine_api.rs @@ -68,9 +68,7 @@ impl Inference for TestInference { .find(|message| message.role == rmcp::model::Role::User) .map(Message::as_concat_text) .unwrap(); - let message = emit - .message(Message::assistant().with_text(format!("{prompt} answered"))) - .await; + let message = emit.message(Message::assistant().with_text(format!("{prompt} answered"))); yielded_with([ GooseEffect::from(message), GooseEffect::RecordUsage(ProviderUsage::new(