26628352543

Committed 29 May 2026 09:02AM UTC coverage: 66.325% (-0.1%) from 66.439%

Build # 26628352543

Build Type

push

github

Committed by

web-flow

Commit Message

enhance(cli, task): Interactive Ctrl+C drain with force-quit escalation (#687)

When `jp` finishes its main work, it waits for background tasks (e.g.
title generation) before exiting. Previously this was a silent, blocking
call with no user feedback and no way to interrupt it.

This commit replaces that with an interactive drain loop. If background
tasks are still running after 1 second, a `⏱ Finishing background tasks…
Ns` line is printed to stderr (TTY only). The first Ctrl+C
(SIGINT/SIGTERM) switches the line to a 2-second countdown and signals
graceful cancellation via the existing soft-cancel token. A second
Ctrl+C — or SIGQUIT — escalates to a hard force-quit that aborts the
`JoinSet` immediately and drops any pending workspace mutations.

The escalation path required a second `force_token` on `TaskHandler`.
The soft-cancel token already existed but was only fired internally; it
is now also exposed via `cancel_token()` so the drain loop can signal it
externally. `is_empty()` is added to short-circuit the whole drain when
there is nothing to wait for.

As a related fix, `TitleGeneratorTask` now accepts an `is_tty` flag and
suppresses the OSC-2 terminal-title sequence when the process is not
attached to a TTY. Without this guard, the escape bytes leaked into
captured pipes or CI logs when running `jp` non-interactively.

---------

Signed-off-by: Jean Mertz <git@jeanmertz.com>

Coverage Stats

11 of 86 new or added lines in 4 files covered. (12.79%)

11 existing lines in 3 files now uncovered.

31992 of 48235 relevant lines covered (66.33%)

267.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

3.57

/crates/jp_task/src/handler.rs

use std::{error::Error, time::Duration};

use jp_workspace::Workspace;
use tokio::task::{JoinError, JoinSet};
use tokio_util::sync::CancellationToken;
use tracing::{debug, trace, warn};

use crate::Task;

#[derive(Debug, Default)]
pub struct TaskHandler {
    tasks: JoinSet<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>>,
    /// Soft-cancellation signal. Firing it asks each task's `run()` to
    /// return promptly; well-behaved tasks then proceed to their `sync()`
    /// phase under [`TaskHandler::sync`]. Tasks that don't observe the
    /// token are force-aborted after the grace window.
    cancel_token: CancellationToken,
    /// Hard-cancellation signal. Firing it short-circuits both the soft
    /// wait and the grace window in [`TaskHandler::sync`], force-aborts
    /// the `JoinSet`, and skips the workspace-sync iteration entirely.
    /// Tasks that had completed their `run()` cleanly lose their pending
    /// workspace mutation.
    force_token: CancellationToken,
}

impl TaskHandler {
    /// Returns `true` if no tasks are currently live.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.tasks.is_empty()
    }

    /// Returns a clone of the soft-cancellation token.
    ///
    /// Cancelling the token signals every task's `run()` to stop. The
    /// `sync()` phase still runs for tasks that returned cleanly.
    #[must_use]
    pub fn cancel_token(&self) -> CancellationToken {
        self.cancel_token.clone()
    }

    /// Returns a clone of the hard-cancellation token.
    ///
    /// Cancelling the token force-aborts the `JoinSet` and skips the
    /// workspace-sync iteration. Pending workspace mutations are dropped.
    #[must_use]
    pub fn force_token(&self) -> CancellationToken {
        self.force_token.clone()
    }

    pub fn spawn(&mut self, task: impl Task) {
        let name = task.name();
        debug!(name, "Spawning task.");
        let token = self.cancel_token.child_token();
        self.tasks.spawn(async move {
            let mut task = Box::new(task).run(token);
            let now = tokio::time::Instant::now();
            loop {
                jp_macro::select!(
                    biased,
                    tokio::time::sleep(Duration::from_millis(500)),
                    |_wake| {
                        trace!(name, elapsed_ms = %now.elapsed().as_millis(), "Task running...");
                    },
                    &mut task,
                    |v| {
                        debug!(name, elapsed_ms = %now.elapsed().as_millis(), "Task completed.");
                        return v;
                    }
                );
            }
        });
    }

    pub async fn sync(
        &mut self,
        workspace: &mut Workspace,
        timeout: Duration,
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        if self.tasks.is_empty() {
            return Ok(());
        }

        let mut tasks: Vec<Box<dyn Task>> = Vec::new();
        self.wait_for_tasks(timeout, &mut tasks, false).await;

        // Grace window for stragglers that didn't observe the soft
        // cancellation signal.
        self.wait_for_tasks(Duration::from_secs(2), &mut tasks, true)
            .await;

        // Force quit: drop accumulated results without applying them.
        if self.force_token.is_cancelled() {
            warn!(
                count = tasks.len(),
                "Force-quit requested; skipping workspace sync for collected tasks."
            );
            return Ok(());
        }

        for task in tasks {
            if let Err(error) = task.sync(workspace).await {
                tracing::error!(%error, "Error syncing background task.");
            }
        }

        Ok(())
    }

    async fn wait_for_tasks(
        &mut self,
        timeout: Duration,
        tasks: &mut Vec<Box<dyn Task>>,
        shutdown: bool,
    ) {
        let timeout = tokio::time::sleep(timeout);
        tokio::pin!(timeout);

        loop {
            jp_macro::select!(
                biased,
                self.force_token.cancelled(),
                |_force| {
                    // Force quit: abort everything immediately. The
                    // grace pass observes the same signal and exits via
                    // the empty-JoinSet branch.
                    warn!("Force-quit requested. Aborting background tasks.");
                    self.tasks.shutdown().await;
                    break;
                },
                self.cancel_token.cancelled(),
                |_cancel| if (!shutdown) {
                    // Soft cancellation fired externally during the soft
                    // wait: stop waiting and let the grace pass collect
                    // any tasks still in flight.
                    break;
                },
                &mut timeout,
                |_wake| {
                    if shutdown {
                        warn!("Tasks did not respond to cancellation signal. Forcing shutdown.");
                        self.tasks.shutdown().await;
                    } else {
                        warn!(
                            "Task finalization timed out. Signalling cancellation to remaining \
                             tasks."
                        );
                        self.cancel_token.cancel();
                    }
                    break;
                },
                self.tasks.join_next(),
                |task| {
                    match task {
                        Some(task) => handle_task_completion(task, tasks),
                        None => break,
                    }
                },
            );
        }
    }
}

#[expect(clippy::type_complexity)]
fn handle_task_completion(
    result: Result<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>, JoinError>,
    tasks: &mut Vec<Box<dyn Task>>,
) {
    match result {
        Ok(Ok(task)) => tasks.push(task),
        Ok(Err(error)) => tracing::error!(%error, "Background task failed."),
        Err(error) => tracing::error!(%error, "Error waiting for background task to complete."),
    }
}

1	use std::{error::Error, time::Duration};
2
3	use jp_workspace::Workspace;
4	use tokio::task::{JoinError, JoinSet};
5	use tokio_util::sync::CancellationToken;
6	use tracing::{debug, trace, warn};
7
8	use crate::Task;
9
10	#[derive(Debug, Default)]
11	pub struct TaskHandler {
12	tasks: JoinSet<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>>,
13	/// Soft-cancellation signal. Firing it asks each task's `run()` to
14	/// return promptly; well-behaved tasks then proceed to their `sync()`
15	/// phase under [`TaskHandler::sync`]. Tasks that don't observe the
16	/// token are force-aborted after the grace window.
17	cancel_token: CancellationToken,
18	/// Hard-cancellation signal. Firing it short-circuits both the soft
19	/// wait and the grace window in [`TaskHandler::sync`], force-aborts
20	/// the `JoinSet`, and skips the workspace-sync iteration entirely.
21	/// Tasks that had completed their `run()` cleanly lose their pending
22	/// workspace mutation.
23	force_token: CancellationToken,
24	}
25
26	impl TaskHandler {
27	/// Returns `true` if no tasks are currently live.
28	#[must_use]
29	pub fn is_empty(&self) -> bool {	2✔
30	self.tasks.is_empty()	2✔
31	}	2✔
32
33	/// Returns a clone of the soft-cancellation token.
34	///
35	/// Cancelling the token signals every task's `run()` to stop. The
36	/// `sync()` phase still runs for tasks that returned cleanly.
37	#[must_use]
NEW 38	pub fn cancel_token(&self) -> CancellationToken {	×
NEW 39	self.cancel_token.clone()	×
NEW 40	}	×
41
42	/// Returns a clone of the hard-cancellation token.
43	///
44	/// Cancelling the token force-aborts the `JoinSet` and skips the
45	/// workspace-sync iteration. Pending workspace mutations are dropped.
46	#[must_use]
NEW 47	pub fn force_token(&self) -> CancellationToken {	×
NEW 48	self.force_token.clone()	×
NEW 49	}	×
50
51	pub fn spawn(&mut self, task: impl Task) {	×
52	let name = task.name();	×
53	debug!(name, "Spawning task.");	×
54	let token = self.cancel_token.child_token();	×
55	self.tasks.spawn(async move {	×
56	let mut task = Box::new(task).run(token);	×
57	let now = tokio::time::Instant::now();	×
58	loop {
59	jp_macro::select!(	×
60	biased,
61	tokio::time::sleep(Duration::from_millis(500)),	×
62	\|_wake\| {
63	trace!(name, elapsed_ms = %now.elapsed().as_millis(), "Task running...");	×
64	},
65	&mut task,	×
66	\|v\| {
67	debug!(name, elapsed_ms = %now.elapsed().as_millis(), "Task completed.");	×
68	return v;	×
69	}
70	);
71	}
72	});	×
73	}	×
74
UNCOV 75	pub async fn sync(	×
UNCOV 76	&mut self,	×
UNCOV 77	workspace: &mut Workspace,	×
UNCOV 78	timeout: Duration,	×
UNCOV 79	) -> Result<(), Box<dyn Error + Send + Sync>> {	×
UNCOV 80	if self.tasks.is_empty() {	×
UNCOV 81	return Ok(());	×
82	}	×
83
84	let mut tasks: Vec<Box<dyn Task>> = Vec::new();	×
85	self.wait_for_tasks(timeout, &mut tasks, false).await;	×
86
87	// Grace window for stragglers that didn't observe the soft
88	// cancellation signal.
89	self.wait_for_tasks(Duration::from_secs(2), &mut tasks, true)	×
90	.await;	×
91
92	// Force quit: drop accumulated results without applying them.
NEW 93	if self.force_token.is_cancelled() {	×
NEW 94	warn!(	×
NEW 95	count = tasks.len(),	×
96	"Force-quit requested; skipping workspace sync for collected tasks."
97	);
NEW 98	return Ok(());	×
NEW 99	}	×
100
101	for task in tasks {	×
102	if let Err(error) = task.sync(workspace).await {	×
103	tracing::error!(%error, "Error syncing background task.");	×
104	}	×
105	}
106
107	Ok(())	×
UNCOV 108	}	×
109
110	async fn wait_for_tasks(	×
111	&mut self,	×
112	timeout: Duration,	×
113	tasks: &mut Vec<Box<dyn Task>>,	×
114	shutdown: bool,	×
115	) {	×
116	let timeout = tokio::time::sleep(timeout);	×
117	tokio::pin!(timeout);	×
118
119	loop {
120	jp_macro::select!(	×
121	biased,
NEW 122	self.force_token.cancelled(),	×
123	\|_force\| {
124	// Force quit: abort everything immediately. The
125	// grace pass observes the same signal and exits via
126	// the empty-JoinSet branch.
NEW 127	warn!("Force-quit requested. Aborting background tasks.");	×
NEW 128	self.tasks.shutdown().await;	×
NEW 129	break;	×
130	},
NEW 131	self.cancel_token.cancelled(),	×
NEW 132	\|_cancel\| if (!shutdown) {	×
133	// Soft cancellation fired externally during the soft
134	// wait: stop waiting and let the grace pass collect
135	// any tasks still in flight.
NEW 136	break;	×
137	},
UNCOV 138	&mut timeout,	×
139	\|_wake\| {
140	if shutdown {	×
141	warn!("Tasks did not respond to cancellation signal. Forcing shutdown.");	×
142	self.tasks.shutdown().await;	×
143	} else {
144	warn!(	×
145	"Task finalization timed out. Signalling cancellation to remaining \
146	tasks."
147	);
148	self.cancel_token.cancel();	×
149	}
150	break;	×
151	},
152	self.tasks.join_next(),	×
153	\|task\| {
154	match task {	×
155	Some(task) => handle_task_completion(task, tasks),	×
156	None => break,	×
157	}
158	},
159	);
160	}
161	}	×
162	}
163
164	#[expect(clippy::type_complexity)]
165	fn handle_task_completion(	×
166	result: Result<Result<Box<dyn Task>, Box<dyn Error + Send + Sync>>, JoinError>,	×
167	tasks: &mut Vec<Box<dyn Task>>,	×
168	) {	×
169	match result {	×
170	Ok(Ok(task)) => tasks.push(task),	×
171	Ok(Err(error)) => tracing::error!(%error, "Background task failed."),	×
172	Err(error) => tracing::error!(%error, "Error waiting for background task to complete."),	×
173	}
174	}	×

dcdpr / jp / 26628352543

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous