captchaforge 0.2.36

//! Speculative artifact pre-flight.
//!
//! The chain currently runs solvers strictly sequentially: each
//! solver finishes (success or failure) before the next starts.
//! Some solvers are expensive to *prepare* — the VLM solver spends
//! ~1-2s capturing a high-resolution screenshot before it can even
//! send the model request. The audio solver spends ~500ms
//! downloading + decoding the challenge audio. If the cheaper
//! solver before them succeeds, that prep work is wasted; if it
//! fails, the chain pays the prep cost on the critical path.
//!
//! [`PreflightHandle`] inverts that: the chain spawns the
//! expensive read-only prep work AS SOON AS the captcha is
//! detected, in parallel with the cheap solvers. Each prep task
//! is a [`PreflightHandle`] returning some artifact (screenshot
//! bytes, decoded audio, DOM snapshot). When the chain later
//! decides it needs the artifact, it `await`s the in-flight
//! handle — which has often already completed and returns
//! immediately. If the chain decides it doesn't need the artifact
//! (cheap solver won), it drops the handle, which cancels the
//! task.
//!
//! ## Why read-only?
//!
//! Two solvers that BOTH click the page would step on each other.
//! Speculation only works for tasks that don't mutate the page:
//! screenshot capture, DOM evals, audio-element extraction, token
//! polling. Anything that calls `dispatchMouseEvent` is sequential
//! by nature.
//!
//! ## Cancellation
//!
//! Dropping a [`PreflightHandle`] aborts the underlying task via
//! [`tokio::task::JoinHandle::abort`]. The artifact-producing
//! future doesn't need to be cancellation-aware — tokio yields at
//! every `.await` and the cancel takes effect at the next yield
//! point.

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::task::JoinHandle;

/// A future-of-T that has been spawned onto the runtime ahead of
/// time. The chain calls `.await` only when it actually needs the
/// artifact — typically after a cheaper solver has failed.
///
/// Implements [`Future`] so callers use it identically to a
/// non-speculative future. Drop = cancel.
pub struct PreflightHandle<T> {
    handle: Option<JoinHandle<T>>,
}

impl<T> PreflightHandle<T>
where
    T: Send + 'static,
{
    /// Spawn `fut` onto the current tokio runtime and return a
    /// handle the chain can `.await` later.
    ///
    /// Panics if called outside a tokio runtime — same constraint
    /// as `tokio::spawn`. The chain always runs inside one.
    pub fn spawn<F>(fut: F) -> Self
    where
        F: Future<Output = T> + Send + 'static,
    {
        Self {
            handle: Some(tokio::spawn(fut)),
        }
    }

    /// Cancel the in-flight task without awaiting its result.
    /// Idempotent — second call is a no-op. Useful when the chain
    /// learns it doesn't need the artifact (cheap solver won)
    /// before the natural drop point.
    pub fn cancel(mut self) {
        if let Some(h) = self.handle.take() {
            h.abort();
        }
    }

    /// True iff the spawned task has finished — useful for
    /// "should I bother awaiting, or is the artifact already here?"
    /// telemetry checks. Not load-bearing for correctness.
    pub fn is_finished(&self) -> bool {
        self.handle
            .as_ref()
            .map(|h| h.is_finished())
            .unwrap_or(true)
    }
}

impl<T> Future for PreflightHandle<T>
where
    T: Send + 'static,
{
    /// `Result<T, PreflightError>` so callers can distinguish
    /// "the spawn task panicked" from "T was an error". For
    /// captchaforge's use case, T is itself usually a `Result`,
    /// so the user sees nested Results — that's the price of
    /// preserving the panic vs business-error distinction.
    type Output = Result<T, PreflightError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let Some(h) = self.handle.as_mut() else {
            return Poll::Ready(Err(PreflightError::Cancelled));
        };
        match Pin::new(h).poll(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(Ok(v)) => {
                // Drop the handle so subsequent polls return
                // Cancelled rather than panicking the joinhandle.
                self.handle = None;
                Poll::Ready(Ok(v))
            }
            Poll::Ready(Err(join_err)) => {
                self.handle = None;
                Poll::Ready(Err(if join_err.is_cancelled() {
                    PreflightError::Cancelled
                } else {
                    PreflightError::Panicked
                }))
            }
        }
    }
}

impl<T> Drop for PreflightHandle<T> {
    fn drop(&mut self) {
        if let Some(h) = self.handle.take() {
            // Drop = cancel. tokio yields at every .await so this
            // takes effect at the next yield point inside the
            // spawned task.
            h.abort();
        }
    }
}

/// Why a [`PreflightHandle`] resolved without a valid artifact.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum PreflightError {
    /// The handle was cancelled (explicitly or via drop) before
    /// the spawned task finished.
    Cancelled,
    /// The spawned task panicked.
    Panicked,
}

impl std::fmt::Display for PreflightError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            PreflightError::Cancelled => write!(f, "preflight task was cancelled"),
            PreflightError::Panicked => write!(f, "preflight task panicked"),
        }
    }
}

impl std::error::Error for PreflightError {}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    #[tokio::test]
    async fn handle_returns_value_when_spawned_task_completes() {
        let h = PreflightHandle::spawn(async { 42_i32 });
        let v = h.await;
        assert_eq!(v, Ok(42));
    }

    #[tokio::test]
    async fn handle_returns_value_for_long_task_when_awaited_after_completion() {
        let h = PreflightHandle::spawn(async {
            tokio::time::sleep(Duration::from_millis(50)).await;
            "ok"
        });
        // Caller does other work first.
        tokio::time::sleep(Duration::from_millis(100)).await;
        // Task has already completed; await is essentially free.
        let v = h.await;
        assert_eq!(v, Ok("ok"));
    }

    #[tokio::test]
    async fn explicit_cancel_aborts_task_and_subsequent_await_errors() {
        let started = Arc::new(AtomicBool::new(false));
        let started2 = started.clone();
        let h = PreflightHandle::spawn(async move {
            started2.store(true, Ordering::SeqCst);
            tokio::time::sleep(Duration::from_secs(60)).await;
            "completed"
        });
        // Give it a moment to start.
        tokio::time::sleep(Duration::from_millis(10)).await;
        assert!(started.load(Ordering::SeqCst));
        h.cancel();
        // After cancel, the spawned task is killed. We can't
        // await the same handle (consumed), so this assertion is
        // implicit: no panic, no leak.
    }

    #[tokio::test]
    async fn drop_aborts_task() {
        let started = Arc::new(AtomicBool::new(false));
        let still_running = Arc::new(AtomicBool::new(true));
        {
            let s2 = started.clone();
            let r2 = still_running.clone();
            let _h = PreflightHandle::spawn(async move {
                s2.store(true, Ordering::SeqCst);
                // Long sleep — will be cancelled.
                tokio::time::sleep(Duration::from_secs(60)).await;
                r2.store(false, Ordering::SeqCst);
            });
            // Give it a moment to start.
            tokio::time::sleep(Duration::from_millis(10)).await;
            // _h drops at end of this block.
        }
        // Wait long enough that an un-cancelled task would have
        // had a chance to update `still_running` to false. Since
        // we cancelled, it stays true.
        tokio::time::sleep(Duration::from_millis(50)).await;
        assert!(started.load(Ordering::SeqCst), "task should have started");
        assert!(
            still_running.load(Ordering::SeqCst),
            "cancelled task must not run to completion"
        );
    }

    #[tokio::test]
    async fn is_finished_reflects_task_state() {
        let h = PreflightHandle::spawn(async {
            tokio::time::sleep(Duration::from_millis(20)).await;
            7_u32
        });
        assert!(!h.is_finished(), "should not be finished immediately");
        tokio::time::sleep(Duration::from_millis(60)).await;
        assert!(h.is_finished(), "should be finished after sleep");
        let v = h.await;
        assert_eq!(v, Ok(7));
    }

    #[tokio::test]
    async fn polling_after_panic_returns_panicked_error() {
        let h: PreflightHandle<i32> = PreflightHandle::spawn(async {
            #[allow(clippy::panic)]
            {
                panic!("boom");
            }
        });
        let v = h.await;
        assert_eq!(v, Err(PreflightError::Panicked));
    }

    #[test]
    fn preflight_error_implements_display_and_error_traits() {
        // std::error::Error bound — important for `?`-propagation
        // through anyhow::Result in caller code.
        let cancelled: Box<dyn std::error::Error> = Box::new(PreflightError::Cancelled);
        let panicked: Box<dyn std::error::Error> = Box::new(PreflightError::Panicked);
        assert!(cancelled.to_string().contains("cancelled"));
        assert!(panicked.to_string().contains("panicked"));
    }
}