fix: unexpected error message when ctrl+c is entered while running recipe with sub-recipes

[lifeizhou-ap]

Handle CLI CTRL+C gracefully to cancel the tasks

fixes issue #3553

[Copilot]

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

[Copilot]

Pull Request Overview

This PR fixes error message suppression when Ctrl+C is pressed during task execution. The changes implement proper cancellation handling to prevent spurious error messages that occur when channels are closed due to task cancellation rather than actual errors.

Key changes:

• Added cancellation token support throughout the task execution pipeline
• Implemented error message suppression logic when tasks are cancelled and channels are closed
• Added cancellation handling in the CLI session to properly clean up subagent executions

Reviewed Changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
workers.rs	Added conditional error logging to suppress "channel closed" errors when task is cancelled
tasks_manager.rs	Added cancellation token tracking and management functionality
task_execution_tracker.rs	Added cancellation state tracking and error suppression logic
subagent_execute_task_tool.rs	Integrated cancellation token support into task execution flow
lib/mod.rs	Updated function signatures to accept cancellation tokens
executor/mod.rs	Added cancellation handling setup and tokio::select! for graceful cancellation
agent.rs	Added public method to cancel all subagent executions
session/mod.rs	Added Ctrl+C handler to cancel subagent executions

[Copilot]

The should_suppress_error method is marked as public but appears to be an internal implementation detail. Consider making this method private or documenting its public API contract if it's intended for external use.

Suggested change

	pub fn should_suppress_error(
	fn should_suppress_error(

[Copilot]

The cleanup of cancelled tokens happens on every registration. For high-frequency registrations, this could become inefficient. Consider implementing a periodic cleanup strategy or cleanup threshold to avoid O(n) operations on every registration.

[lifeizhou-ap]

this is not high-frequency registrations

[Copilot]

Using string matching on error messages is fragile and could break if the error message format changes. Consider using pattern matching on the error type instead: matches!(e, tokio::sync::mpsc::error::SendError(_))

Suggested change

	&& e.to_string().contains("channel closed");
	&& matches!(e, tokio::sync::mpsc::error::SendError(_));

[michaelneale]

just flagging this PR #3554 which may collide with this (but not sure)

[lifeizhou-ap]

just flagging this PR #3554 which may collide with this (but not sure)

oh! thank you @michaelneale for the heads-up

[wendytang]

Should PR #3550 and PR #3554 Be Combined?
Analysis: Can They Coexist?
Looking at both PRs, they have overlapping but different approaches to cancellation. Let me analyze the compatibility:

My Recommendation: Enhanced PR #3554
I recommend Option 1 (Enhanced PR #3554) because:

Advantages:
✅ Single source of truth: One cancellation token flows through the entire system
✅ Consistent behavior: All operations respect the same cancellation signal
✅ Future-proof: New subsystems automatically get cancellation support
✅ Clean architecture: Clear separation between session management and execution
✅ Comprehensive: Handles both general agent operations AND sub-recipe specifics
Implementation Strategy:
Start with PR #3554's foundation
Add the sub-recipe cancellation token propagation from PR #3550
Remove the redundant cancel_all_subagent_executions() method
Ensure all subsystems receive and respect the same cancellation token


Option 1: Enhanced PR #3554 (Recommended)
Build on PR #3554's general approach but add the granular sub-recipe handling from PR #3550:


// Core Agent interface (from PR #3554)
pub async fn reply(
    &self,
    messages: &[Message],
    session: Option<SessionConfig>,
    cancel_token: Option<CancellationToken>, // ✅ General cancellation
) -> Result<BoxStream<'_, Result<AgentEvent>>>

// Enhanced session handling
async fn process_agent_response(&mut self, interactive: bool) -> Result<()> {
    let cancel_token = CancellationToken::new();
    let cancel_token_clone = cancel_token.clone();
    
    // ✅ Pass cancellation to both general agent AND sub-recipes
    self.agent.set_cancellation_token(cancel_token.clone()).await;
    
    let mut stream = self
        .agent
        .reply(&self.messages, session_config.clone(), Some(cancel_token))
        .await?;
        
    tokio::select! {
        // ... existing stream handling ...
        _ = tokio::signal::ctrl_c() => {
            // ✅ Single cancellation point that handles everything
            cancel_token_clone.cancel();
            drop(stream);
            // ... handle interruption ...
        }
    }
}
Enhanced Agent Implementation:

impl Agent {
    // ✅ Store cancellation token for sub-systems to use
    async fn set_cancellation_token(&self, token: CancellationToken) {
        // Pass to tasks manager
        self.tasks_manager.set_cancellation_token(token.clone()).await;
        
        // Pass to any other long-running subsystems
        // self.extension_manager.set_cancellation_token(token.clone()).await;
    }
    
    pub async fn reply(
        &self,
        messages: &[Message],
        session: Option<SessionConfig>,
        cancel_token: Option<CancellationToken>,
    ) -> Result<BoxStream<'_, Result<AgentEvent>>> {
        // ✅ Check cancellation in main loop (from PR #3554)
        loop {
            if cancel_token.as_ref().is_some_and(|t| t.is_cancelled()) {
                break;
            }
            
            // ... existing logic ...
            // Sub-recipes will automatically respect the same token
        }
    }
}
Enhanced TasksManager:

impl TasksManager {
    // ✅ Unified cancellation token management
    pub async fn set_cancellation_token(&self, token: CancellationToken) {
        let mut current_token = self.cancellation_token.write().await;
        *current_token = Some(token);
        
        // Cancel any existing executions
        self.cancel_all_executions().await;
    }
    
    pub async fn register_execution(&self, execution_token: CancellationToken) {
        // ✅ Link execution token to global cancellation
        if let Some(global_token) = &*self.cancellation_token.read().await {
            let linked_token = CancellationToken::new();
            
            // Cancel execution if either global OR execution-specific cancellation occurs
            tokio::spawn({
                let global = global_token.clone();
                let execution = execution_token.clone();
                let linked = linked_token.clone();
                async move {
                    tokio::select! {
                        _ = global.cancelled() => linked.cancel(),
                        _ = execution.cancelled() => linked.cancel(),
                    }
                }
            });
        }
        
        // Store the linked token
        let mut tokens = self.active_tokens.write().await;
        tokens.push(linked_token);
    }
}

[wendytang]

nice! definitely think this is needed.

@DOsinga's #3554 seems to be an overarching solution for clean ups, we could propagate the main agent cancellation token from to the subagent

[lifeizhou-ap]

I created a new PR #3599 to reuse the cancellation token created in agent.reply