diff --git a/compiler/rustc_infer/src/traits/mod.rs b/compiler/rustc_infer/src/traits/mod.rs
index 86aae77adcf8e..ae91abb62a943 100644
--- a/compiler/rustc_infer/src/traits/mod.rs
+++ b/compiler/rustc_infer/src/traits/mod.rs
@@ -18,6 +18,7 @@ use rustc_middle::traits::solve::Certainty;
 pub use rustc_middle::traits::*;
 use rustc_middle::ty::{self, Ty, TyCtxt, Upcast};
 use rustc_span::Span;
+use rustc_type_ir::solve::fulfill::FulfillmentObligation;
 use thin_vec::ThinVec;
 
 pub use self::engine::{FromSolverError, ScrubbedTraitError, TraitEngine, TraitErrors};
@@ -92,6 +93,28 @@ pub type PolyTraitObligation<'tcx> = Obligation<'tcx, ty::PolyTraitPredicate<'tc
 
 pub type PredicateObligations<'tcx> = ThinVec<PredicateObligation<'tcx>>;
 
+impl<'tcx> FulfillmentObligation<TyCtxt<'tcx>> for PredicateObligation<'tcx> {
+    fn as_goal(&self) -> solve::Goal<'tcx, ty::Predicate<'tcx>> {
+        Obligation::as_goal(self)
+    }
+
+    fn span(&self) -> Span {
+        self.cause.span
+    }
+
+    fn recursion_depth(&self) -> usize {
+        self.recursion_depth
+    }
+
+    fn set_recursion_depth(&mut self, depth: usize) {
+        self.recursion_depth = depth;
+    }
+
+    fn set_predicate(&mut self, predicate: ty::Predicate<'tcx>) {
+        self.predicate = predicate;
+    }
+}
+
 impl<'tcx> PredicateObligation<'tcx> {
     /// Flips the polarity of the inner predicate.
     ///
diff --git a/compiler/rustc_next_trait_solver/src/solve/fulfill.rs b/compiler/rustc_next_trait_solver/src/solve/fulfill.rs
new file mode 100644
index 0000000000000..73f5d29040b34
--- /dev/null
+++ b/compiler/rustc_next_trait_solver/src/solve/fulfill.rs
@@ -0,0 +1,301 @@
+use rustc_type_ir::solve::fulfill::FulfillmentObligation;
+use rustc_type_ir::solve::{Certainty, Goal, NoSolution};
+use rustc_type_ir::{InferCtxtLike as _, Interner};
+use thin_vec::ThinVec;
+
+use super::fast_path::compute_goal_fast_path;
+use super::{
+    GoalEvaluation, GoalStalledOn, HasChanged, SolverDelegate, SolverDelegateEvalExt as _,
+};
+
+#[derive(Debug, Clone)]
+pub enum NextSolverError<O> {
+    TrueError(O),
+    Ambiguity(O),
+    Overflow(O),
+}
+
+// FIXME: Do we need to use a `ThinVec` here?
+type PendingObligations<I, O> = ThinVec<(O, Option<GoalStalledOn<I>>)>;
+
+#[derive(Debug)]
+struct ObligationStorage<I: Interner, O> {
+    /// Obligations which resulted in overflow in fulfillment itself.
+    ///
+    /// We cannot eagerly return these as errors, so we instead store them here
+    /// to avoid recomputing them each time `try_evaluate_obligations` is called.
+    /// This also allows the frontend to construct the correct error for them.
+    overflowed: Vec<O>,
+
+    pending: PendingObligations<I, O>,
+}
+
+impl<I: Interner, O> Default for ObligationStorage<I, O> {
+    fn default() -> Self {
+        Self { overflowed: Vec::new(), pending: ThinVec::new() }
+    }
+}
+
+impl<I: Interner, O> ObligationStorage<I, O> {
+    fn register(&mut self, obligation: O, stalled_on: Option<GoalStalledOn<I>>) {
+        self.pending.push((obligation, stalled_on));
+    }
+
+    fn has_pending_obligations(&self) -> bool {
+        !self.pending.is_empty() || !self.overflowed.is_empty()
+    }
+
+    fn clone_pending(&self) -> ThinVec<O>
+    where
+        O: Clone,
+    {
+        let mut obligations =
+            self.pending.iter().map(|(obligation, _)| obligation.clone()).collect::<ThinVec<_>>();
+
+        obligations.extend(self.overflowed.iter().cloned());
+        obligations
+    }
+
+    fn clone_pending_filtered<F>(&self, mut filter: F) -> ThinVec<O>
+    where
+        O: Clone,
+        F: FnMut(&O, &Option<GoalStalledOn<I>>) -> bool,
+    {
+        let mut obligations = self
+            .pending
+            .iter()
+            .filter_map(|(obligation, stalled_on)| {
+                filter(obligation, stalled_on).then(|| obligation.clone())
+            })
+            .collect::<ThinVec<_>>();
+
+        obligations.extend(self.overflowed.iter().cloned());
+        obligations
+    }
+
+    fn drain_pending<F>(&mut self, mut filter: F) -> ThinVec<O>
+    where
+        F: FnMut(&O, &Option<GoalStalledOn<I>>) -> bool,
+    {
+        let (drained, pending): (PendingObligations<I, O>, PendingObligations<I, O>) =
+            std::mem::take(&mut self.pending)
+                .into_iter()
+                .partition(|(obligation, stalled_on)| filter(obligation, stalled_on));
+
+        self.pending = pending;
+
+        drained.into_iter().map(|(obligation, _)| obligation).collect()
+    }
+
+    #[cold]
+    #[inline(never)]
+    fn collect_remaining_errors<E>(
+        &mut self,
+        map: impl FnMut(NextSolverError<O>) -> E,
+    ) -> ThinVec<E> {
+        self.pending
+            .drain(..)
+            .map(|(obligation, _)| NextSolverError::Ambiguity(obligation))
+            .chain(self.overflowed.drain(..).map(NextSolverError::Overflow))
+            .map(map)
+            .collect()
+    }
+}
+
+/// A fulfillment engine using the new trait solver.
+///
+/// This is mostly identical to how `evaluate_all` works inside of the solver,
+/// except that it is possible to add new obligations later and the frontend
+/// needs to retain its obligation representation for diagnostics.
+///
+/// It is also likely that we want to use different data structures here, as
+/// fulfillment deals with far more root goals than `evaluate_all`.
+#[derive(Debug)]
+pub struct FulfillmentCtxt<I: Interner, O: FulfillmentObligation<I>> {
+    obligations: ObligationStorage<I, O>,
+}
+
+impl<I: Interner, O: FulfillmentObligation<I>> Default for FulfillmentCtxt<I, O> {
+    fn default() -> Self {
+        Self { obligations: ObligationStorage::default() }
+    }
+}
+
+impl<I: Interner, O: FulfillmentObligation<I>> FulfillmentCtxt<I, O> {
+    pub fn new() -> Self {
+        Self { obligations: Default::default() }
+    }
+
+    pub fn register<D>(&mut self, delegate: &D, obligation: O)
+    where
+        D: SolverDelegate<Interner = I>,
+    {
+        if let Some(GoalEvaluation { certainty, stalled_on, .. }) =
+            compute_goal_fast_path(delegate, obligation.as_goal(), obligation.span())
+        {
+            // If we can take the fast path, do not add a successful goal to
+            // the pending obligations. For `Certainty::Maybe`, retain the
+            // precise `stalled_on` information for later re-evaluation.
+            match certainty {
+                Certainty::Yes => {}
+                Certainty::Maybe(_) => {
+                    self.obligations.register(obligation, stalled_on);
+                }
+            }
+        } else {
+            self.obligations.register(obligation, None);
+        }
+    }
+
+    fn on_fulfillment_overflow<D>(&mut self, delegate: &D)
+    where
+        D: SolverDelegate<Interner = I>,
+    {
+        delegate.probe(|| {
+            // IMPORTANT: we must not resolve any inference variables in the
+            // obligations, as this is all happening inside of a probe. The
+            // probe makes sure we collect every obligation involved in the
+            // overflow. Conceptually, we check which goals would change if we
+            // performed one more fulfillment iteration.
+            let overflowed = self
+                .obligations
+                .pending
+                .extract_if(.., |(obligation, stalled_on)| {
+                    let result = delegate.evaluate_root_goal(
+                        obligation.as_goal(),
+                        obligation.span(),
+                        stalled_on.take(),
+                    );
+
+                    matches!(result, Ok(GoalEvaluation { has_changed: HasChanged::Yes, .. }))
+                })
+                .map(|(obligation, _)| obligation)
+                .collect::<Vec<_>>();
+
+            self.obligations.overflowed.extend(overflowed);
+        });
+    }
+
+    pub fn try_evaluate_obligations<D, Inspect, OnSuccess>(
+        &mut self,
+        delegate: &D,
+        mut inspect: Inspect,
+        mut on_success: OnSuccess,
+    ) -> ThinVec<NextSolverError<O>>
+    where
+        D: SolverDelegate<Interner = I>,
+        Inspect: FnMut(&O, Goal<I, I::Predicate>, &Result<GoalEvaluation<I>, NoSolution>),
+        OnSuccess: FnMut(&O),
+    {
+        let mut errors = ThinVec::new();
+
+        loop {
+            let mut any_changed = false;
+            let mut overflowed = false;
+
+            self.obligations.pending.retain_mut(|(obligation, opt_stalled_on)| {
+                if overflowed {
+                    return false;
+                }
+
+                // Common case: still stalled; keep the obligation. This path is extremely hot in
+                // some cases; there can be thousands of pending obligations.
+                if let Some(stalled_on) = opt_stalled_on
+                    && let Some(certainty) = delegate.goal_remains_stalled(stalled_on)
+                    && matches!(certainty, Certainty::Maybe(_))
+                {
+                    return true;
+                }
+
+                let goal = obligation.as_goal();
+                let result =
+                    delegate.evaluate_root_goal(goal, obligation.span(), opt_stalled_on.take());
+
+                inspect(obligation, goal, &result);
+
+                let GoalEvaluation { goal, certainty, has_changed, stalled_on } = match result {
+                    Ok(result) => result,
+                    Err(NoSolution) => {
+                        errors.push(NextSolverError::TrueError(obligation.clone()));
+                        return false;
+                    }
+                };
+
+                // We resolved the goal in `evaluate_root_goal`; retain the eagerly resolved
+                // predicate to avoid repeating this work in the next iteration.
+                obligation.set_predicate(goal.predicate);
+
+                if has_changed == HasChanged::Yes {
+                    // Track the number of times this root goal resulted in inference progress.
+                    let depth = obligation.recursion_depth() + 1;
+                    obligation.set_recursion_depth(depth);
+
+                    if depth > delegate.cx().recursion_limit() {
+                        // We cannot break out of `retain_mut`, so use a flag and handle
+                        // fulfillment overflow after the iteration.
+                        overflowed = true;
+                        return false;
+                    }
+
+                    any_changed = true;
+                }
+
+                match certainty {
+                    Certainty::Yes => {
+                        on_success(obligation);
+                        false
+                    }
+                    Certainty::Maybe(_) => {
+                        *opt_stalled_on = stalled_on;
+                        true
+                    }
+                }
+            });
+
+            if overflowed {
+                self.on_fulfillment_overflow(delegate);
+                // Only return true errors accumulated while processing.
+                return errors;
+            }
+
+            if !any_changed {
+                break;
+            }
+        }
+
+        errors
+    }
+
+    pub fn has_pending_obligations(&self) -> bool {
+        self.obligations.has_pending_obligations()
+    }
+
+    pub fn pending_obligations(&self) -> ThinVec<O>
+    where
+        O: Clone,
+    {
+        self.obligations.clone_pending()
+    }
+
+    pub fn pending_obligations_filtered<F>(&self, filter: F) -> ThinVec<O>
+    where
+        O: Clone,
+        F: FnMut(&O, &Option<GoalStalledOn<I>>) -> bool,
+    {
+        self.obligations.clone_pending_filtered(filter)
+    }
+
+    pub fn drain_pending_obligations<F>(&mut self, filter: F) -> ThinVec<O>
+    where
+        F: FnMut(&O, &Option<GoalStalledOn<I>>) -> bool,
+    {
+        self.obligations.drain_pending(filter)
+    }
+
+    pub fn collect_remaining_errors<E>(
+        &mut self,
+        map: impl FnMut(NextSolverError<O>) -> E,
+    ) -> ThinVec<E> {
+        self.obligations.collect_remaining_errors(map)
+    }
+}
diff --git a/compiler/rustc_next_trait_solver/src/solve/mod.rs b/compiler/rustc_next_trait_solver/src/solve/mod.rs
index 3504882834268..f4a0df6f2c973 100644
--- a/compiler/rustc_next_trait_solver/src/solve/mod.rs
+++ b/compiler/rustc_next_trait_solver/src/solve/mod.rs
@@ -14,6 +14,7 @@
 mod assembly;
 mod effect_goals;
 mod eval_ctxt;
+pub mod fulfill;
 pub mod inspect;
 mod normalizes_to;
 mod project_goals;
diff --git a/compiler/rustc_trait_selection/src/solve.rs b/compiler/rustc_trait_selection/src/solve.rs
index f6c01b12ae4c0..ab5eada726961 100644
--- a/compiler/rustc_trait_selection/src/solve.rs
+++ b/compiler/rustc_trait_selection/src/solve.rs
@@ -1,18 +1,18 @@
 pub use rustc_next_trait_solver::solve::*;
 
 mod delegate;
-mod fulfill;
 pub mod inspect;
 mod normalize;
+mod rustc_fulfill;
 mod select;
 
 pub(crate) use delegate::SolverDelegate;
-pub use fulfill::{FulfillmentCtxt, NextSolverError};
 pub(crate) use normalize::deeply_normalize_for_diagnostics;
 pub use normalize::{
     deeply_normalize, deeply_normalize_with_skipped_universes,
     deeply_normalize_with_skipped_universes_and_ambiguous_coroutine_goals, normalize,
 };
+pub use rustc_fulfill::{FulfillmentCtxt, NextSolverError};
 use rustc_middle::query::Providers;
 use rustc_middle::ty::TyCtxt;
 pub use select::InferCtxtSelectExt;
diff --git a/compiler/rustc_trait_selection/src/solve/fulfill.rs b/compiler/rustc_trait_selection/src/solve/fulfill.rs
deleted file mode 100644
index da596fe3b44b0..0000000000000
--- a/compiler/rustc_trait_selection/src/solve/fulfill.rs
+++ /dev/null
@@ -1,463 +0,0 @@
-use std::marker::PhantomData;
-use std::mem;
-
-use rustc_infer::infer::InferCtxt;
-use rustc_infer::traits::query::NoSolution;
-use rustc_infer::traits::{
-    FromSolverError, PredicateObligation, PredicateObligations, TraitEngine, TraitErrors,
-};
-use rustc_middle::ty::{self, TyCtxt, TypeVisitableExt, TypingMode};
-use rustc_next_trait_solver::solve::fast_path::compute_goal_fast_path;
-use rustc_next_trait_solver::solve::{
-    GoalEvaluation, GoalStalledOn, HasChanged, MaybeInfo, SolverDelegateEvalExt as _,
-    StalledOnCoroutines,
-};
-use thin_vec::ThinVec;
-use tracing::instrument;
-
-use self::derive_errors::*;
-use super::Certainty;
-use super::delegate::SolverDelegate;
-use crate::traits::{FulfillmentError, ScrubbedTraitError};
-
-mod derive_errors;
-
-// FIXME: Do we need to use a `ThinVec` here?
-type PendingObligations<'tcx> =
-    ThinVec<(PredicateObligation<'tcx>, Option<GoalStalledOn<TyCtxt<'tcx>>>)>;
-
-/// A trait engine using the new trait solver.
-///
-/// This is mostly identical to how `evaluate_all` works inside of the
-/// solver, except that the requirements are slightly different.
-///
-/// Unlike `evaluate_all` it is possible to add new obligations later on
-/// and we also have to track diagnostics information by using `Obligation`
-/// instead of `Goal`.
-///
-/// It is also likely that we want to use slightly different datastructures
-/// here as this will have to deal with far more root goals than `evaluate_all`.
-pub struct FulfillmentCtxt<'tcx, E: 'tcx> {
-    obligations: ObligationStorage<'tcx>,
-
-    /// The snapshot in which this context was created. Using the context
-    /// outside of this snapshot leads to subtle bugs if the snapshot
-    /// gets rolled back. Because of this we explicitly check that we only
-    /// use the context in exactly this snapshot.
-    usable_in_snapshot: usize,
-    _errors: PhantomData<E>,
-}
-
-#[derive(Default, Debug)]
-struct ObligationStorage<'tcx> {
-    /// Obligations which resulted in an overflow in fulfillment itself.
-    ///
-    /// We cannot eagerly return these as error so we instead store them here
-    /// to avoid recomputing them each time `try_evaluate_obligations` is called.
-    /// This also allows us to return the correct `FulfillmentError` for them.
-    overflowed: Vec<PredicateObligation<'tcx>>,
-    pending: PendingObligations<'tcx>,
-}
-
-impl<'tcx> ObligationStorage<'tcx> {
-    fn register(
-        &mut self,
-        obligation: PredicateObligation<'tcx>,
-        stalled_on: Option<GoalStalledOn<TyCtxt<'tcx>>>,
-    ) {
-        self.pending.push((obligation, stalled_on));
-    }
-
-    fn has_pending_obligations(&self) -> bool {
-        !self.pending.is_empty() || !self.overflowed.is_empty()
-    }
-
-    fn clone_pending(&self) -> PredicateObligations<'tcx> {
-        let mut obligations: PredicateObligations<'tcx> =
-            self.pending.iter().map(|(o, _)| o.clone()).collect();
-        obligations.extend(self.overflowed.iter().cloned());
-        obligations
-    }
-
-    fn clone_pending_filtered<F>(&self, f: F) -> PredicateObligations<'tcx>
-    where
-        F: FnMut(&&(PredicateObligation<'tcx>, Option<GoalStalledOn<TyCtxt<'tcx>>>)) -> bool,
-    {
-        let mut obligations: PredicateObligations<'tcx> =
-            self.pending.iter().filter(f).map(|(o, _)| o.clone()).collect();
-        obligations.extend(self.overflowed.iter().cloned());
-        obligations
-    }
-
-    fn drain_pending(
-        &mut self,
-        cond: impl Fn(&PredicateObligation<'tcx>, &Option<GoalStalledOn<TyCtxt<'tcx>>>) -> bool,
-    ) -> PendingObligations<'tcx> {
-        let (unstalled, pending) =
-            mem::take(&mut self.pending).into_iter().partition(|(o, s)| cond(o, s));
-        self.pending = pending;
-        unstalled
-    }
-
-    fn on_fulfillment_overflow(&mut self, infcx: &InferCtxt<'tcx>) {
-        infcx.probe(|_| {
-            // IMPORTANT: we must not use solve any inference variables in the obligations
-            // as this is all happening inside of a probe. We use a probe to make sure
-            // we get all obligations involved in the overflow. We pretty much check: if
-            // we were to do another step of `try_evaluate_obligations`, which goals would
-            // change.
-            self.overflowed.extend(
-                self.pending
-                    .extract_if(.., |(o, stalled_on)| {
-                        let goal = o.as_goal();
-                        let result = <&SolverDelegate<'tcx>>::from(infcx).evaluate_root_goal(
-                            goal,
-                            o.cause.span,
-                            stalled_on.take(),
-                        );
-                        matches!(result, Ok(GoalEvaluation { has_changed: HasChanged::Yes, .. }))
-                    })
-                    .map(|(o, _)| o),
-            );
-        })
-    }
-}
-
-impl<'tcx, E: 'tcx> FulfillmentCtxt<'tcx, E> {
-    pub fn new(infcx: &InferCtxt<'tcx>) -> FulfillmentCtxt<'tcx, E> {
-        assert!(
-            infcx.next_trait_solver(),
-            "new trait solver fulfillment context created when \
-            infcx is set up for old trait solver"
-        );
-        FulfillmentCtxt {
-            obligations: Default::default(),
-            usable_in_snapshot: infcx.num_open_snapshots(),
-            _errors: PhantomData,
-        }
-    }
-
-    fn inspect_evaluated_obligation(
-        infcx: &InferCtxt<'tcx>,
-        obligation: &PredicateObligation<'tcx>,
-        result: &Result<GoalEvaluation<TyCtxt<'tcx>>, NoSolution>,
-    ) {
-        if let Some(inspector) = infcx.obligation_inspector.get() {
-            let result = match result {
-                Ok(GoalEvaluation { certainty, .. }) => Ok(*certainty),
-                Err(NoSolution) => Err(NoSolution),
-            };
-            (inspector)(infcx, &obligation, result);
-        }
-    }
-}
-
-impl<'tcx, E> TraitEngine<'tcx, E> for FulfillmentCtxt<'tcx, E>
-where
-    E: FromSolverError<'tcx, NextSolverError<'tcx>>,
-{
-    #[instrument(level = "trace", skip(self, infcx))]
-    fn register_predicate_obligation(
-        &mut self,
-        infcx: &InferCtxt<'tcx>,
-        obligation: PredicateObligation<'tcx>,
-    ) {
-        assert_eq!(self.usable_in_snapshot, infcx.num_open_snapshots());
-
-        let delegate = <&SolverDelegate<'tcx>>::from(infcx);
-        if let Some(GoalEvaluation { goal: _, certainty, has_changed: _, stalled_on }) =
-            compute_goal_fast_path(delegate, obligation.as_goal(), obligation.cause.span)
-        {
-            // If we can take the fast path, don't even bother adding the goal to obligations,
-            // or if `Certainty::Maybe`, add it with precise stalled_on information.
-            match certainty {
-                Certainty::Yes => {}
-                Certainty::Maybe(_) => {
-                    self.obligations.register(obligation, stalled_on);
-                }
-            }
-        } else {
-            self.obligations.register(obligation, None);
-        }
-    }
-
-    #[inline]
-    fn collect_remaining_errors(&mut self, infcx: &InferCtxt<'tcx>) -> TraitErrors<E> {
-        if self.obligations.pending.is_empty() && self.obligations.overflowed.is_empty() {
-            // Typically in more than 99.9% of cases this condition is true, therefore we outline
-            // the other case.
-            TraitErrors::NoErrors
-        } else {
-            TraitErrors::HasErrors(collect_remaining_errors_impl(self, infcx))
-        }
-    }
-
-    fn try_evaluate_obligations(&mut self, infcx: &InferCtxt<'tcx>) -> TraitErrors<E> {
-        assert_eq!(self.usable_in_snapshot, infcx.num_open_snapshots());
-        let mut errors = TraitErrors::NoErrors;
-        let delegate = <&SolverDelegate<'tcx>>::from(infcx);
-        loop {
-            let mut any_changed = false;
-            let mut overflowed = false;
-
-            self.obligations.pending.retain_mut(|(obligation, opt_stalled_on)| {
-                if overflowed {
-                    return false;
-                }
-
-                // Common case: still stalled; keep the obligation. This path is extremely hot in
-                // some cases; there can be thousands of pending obligations.
-                if let Some(stalled_on) = opt_stalled_on
-                    && let Some(certainty) = delegate.goal_remains_stalled(stalled_on)
-                    && matches!(certainty, Certainty::Maybe(_))
-                {
-                    return true;
-                }
-
-                let result = delegate.evaluate_root_goal(
-                    obligation.as_goal(),
-                    obligation.cause.span,
-                    opt_stalled_on.take(),
-                );
-                Self::inspect_evaluated_obligation(infcx, &obligation, &result);
-                let GoalEvaluation { goal, certainty, has_changed, stalled_on } = match result {
-                    Ok(result) => result,
-                    Err(NoSolution) => {
-                        errors.push(E::from_solver_error(
-                            infcx,
-                            NextSolverError::TrueError(obligation.clone()),
-                        ));
-                        return false;
-                    }
-                };
-
-                // We've resolved the goal in `evaluate_root_goal`, avoid redoing this work
-                // in the next iteration. This does not resolve the inference variables
-                // constrained by evaluating the goal.
-                obligation.predicate = goal.predicate;
-                if has_changed == HasChanged::Yes {
-                    // We increment the recursion depth here to track the number of times
-                    // this goal has resulted in inference progress. This doesn't precisely
-                    // model the way that we track recursion depth in the old solver due
-                    // to the fact that we only process root obligations, but it is a good
-                    // approximation and should only result in fulfillment overflow in
-                    // pathological cases.
-                    obligation.recursion_depth += 1;
-
-                    if !infcx.tcx.recursion_limit().value_within_limit(obligation.recursion_depth) {
-                        // At this point we want to stop evaluating goals. We can't break out of
-                        // `retain_mut`, so instead we set this flag which causes all other
-                        // elements to be skipped.
-                        overflowed = true;
-                        return false;
-                    } else {
-                        any_changed = true;
-                    }
-                }
-
-                match certainty {
-                    Certainty::Yes => {
-                        // Goals may depend on structural identity. Region uniquification at the
-                        // start of MIR borrowck may cause things to no longer be so, potentially
-                        // causing an ICE.
-                        //
-                        // While we uniquify root goals in HIR this does not handle cases where
-                        // regions are hidden inside of a type or const inference variable.
-                        //
-                        // FIXME(-Znext-solver): This does not handle inference variables hidden
-                        // inside of an opaque type, e.g. if there's `Opaque = (?x, ?x)` in the
-                        // storage, we can also rely on structural identity of `?x` even if we
-                        // later uniquify it in MIR borrowck.
-                        if infcx.in_hir_typeck
-                            && (obligation.has_non_region_infer() || obligation.has_free_regions())
-                        {
-                            infcx.push_hir_typeck_potentially_region_dependent_goal(
-                                obligation.clone(),
-                            );
-                        }
-                        false
-                    }
-                    Certainty::Maybe(_) => {
-                        // Update `opt_stalled_on` goal, for the next retain_mut, because we are
-                        // running until a fixpoint.
-                        *opt_stalled_on = stalled_on;
-                        true
-                    }
-                }
-            });
-            if overflowed {
-                self.obligations.on_fulfillment_overflow(infcx);
-                // Only return true errors that we have accumulated while processing.
-                return errors;
-            }
-
-            if !any_changed {
-                break;
-            }
-        }
-
-        errors
-    }
-
-    fn has_pending_obligations(&self) -> bool {
-        self.obligations.has_pending_obligations()
-    }
-
-    fn pending_obligations(&self) -> PredicateObligations<'tcx> {
-        self.obligations.clone_pending()
-    }
-
-    fn pending_obligations_potentially_referencing_sub_root(
-        &self,
-        infcx: &InferCtxt<'tcx>,
-        vid: ty::TyVid,
-    ) -> PredicateObligations<'tcx> {
-        // `-Zdisable-fast-paths`: same gate as the other new-solver fast paths.
-        if infcx.tcx.disable_trait_solver_fast_paths() {
-            return self.obligations.clone_pending();
-        }
-        self.obligations.clone_pending_filtered(|(_, stalled_on)| {
-            let Some(stalled_on) = stalled_on else { return true };
-            // Don't reuse the sub-unification roots cached on `stalled_on`:
-            // a later sub-unification merge can have changed which root
-            // each stalled var belongs to, so the cached info can be stale.
-            // Walk `stalled_vars` and recompute the current root instead.
-            //
-            // Conservative here: if a stalled var no longer resolves to an
-            // infer var, some unification happened, so the goal is no longer
-            // stalled. Include it to be re-evaluated downstream.
-            stalled_on.stalled_vars.iter().filter_map(|arg| arg.as_type()).any(|ty| {
-                match *infcx.shallow_resolve(ty).kind() {
-                    ty::Infer(ty::TyVar(tv)) => infcx.sub_unification_table_root_var(tv) == vid,
-                    _ => true,
-                }
-            })
-        })
-    }
-
-    fn pending_obligations_potentially_referencing_float_infer(
-        &self,
-        infcx: &InferCtxt<'tcx>,
-    ) -> PredicateObligations<'tcx> {
-        // `-Zdisable-fast-paths`: same gate as the other new-solver fast paths.
-        if infcx.tcx.disable_trait_solver_fast_paths() {
-            return self.obligations.clone_pending();
-        }
-
-        self.obligations.clone_pending_filtered(|(_, stalled_on)| {
-            let Some(stalled_on) = stalled_on else { return true };
-            // If the stalled vars don't have float infers, the nested goals won't
-            // have them either. We only create float infers for user written literals.
-            stalled_on
-                .stalled_vars
-                .iter()
-                .filter_map(|arg| arg.as_type())
-                .any(|ty| matches!(infcx.shallow_resolve(ty).kind(), ty::Infer(ty::FloatVar(_))))
-        })
-    }
-
-    fn drain_stalled_obligations_for_coroutines(
-        &mut self,
-        infcx: &InferCtxt<'tcx>,
-    ) -> PredicateObligations<'tcx> {
-        let stalled_coroutines = match infcx.typing_mode_raw().assert_not_erased() {
-            TypingMode::Typeck { defining_opaque_types_and_generators } => {
-                defining_opaque_types_and_generators
-            }
-            TypingMode::Coherence
-            | TypingMode::PostTypeckUntilBorrowck { defining_opaque_types: _ }
-            | TypingMode::PostBorrowck { defined_opaque_types: _ }
-            | TypingMode::Reflection
-            | TypingMode::PostAnalysis
-            | TypingMode::Codegen => return Default::default(),
-        };
-
-        if stalled_coroutines.is_empty() {
-            return Default::default();
-        }
-
-        self.obligations
-            .drain_pending(|_, stalled_on| {
-                stalled_on.as_ref().is_some_and(|s| match s.stalled_certainty {
-                    Certainty::Maybe(MaybeInfo {
-                        cause: _,
-                        opaque_types_jank: _,
-                        stalled_on_coroutines: StalledOnCoroutines::Yes,
-                    }) => true,
-                    Certainty::Maybe(_) | Certainty::Yes => false,
-                })
-            })
-            .into_iter()
-            .map(|(o, _)| o)
-            .collect()
-    }
-}
-
-#[cold]
-#[inline(never)]
-fn collect_remaining_errors_impl<'tcx, E>(
-    cx: &mut FulfillmentCtxt<'tcx, E>,
-    infcx: &InferCtxt<'tcx>,
-) -> ThinVec<E>
-where
-    E: FromSolverError<'tcx, NextSolverError<'tcx>>,
-{
-    cx.obligations
-        .pending
-        .drain(..)
-        .map(|(obligation, _)| NextSolverError::Ambiguity(obligation))
-        .chain(
-            cx.obligations
-                .overflowed
-                .drain(..)
-                .map(|obligation| NextSolverError::Overflow(obligation)),
-        )
-        .map(|e| E::from_solver_error(infcx, e))
-        .collect()
-}
-
-pub enum NextSolverError<'tcx> {
-    TrueError(PredicateObligation<'tcx>),
-    Ambiguity(PredicateObligation<'tcx>),
-    Overflow(PredicateObligation<'tcx>),
-}
-
-impl<'tcx> FromSolverError<'tcx, NextSolverError<'tcx>> for FulfillmentError<'tcx> {
-    fn from_solver_error(infcx: &InferCtxt<'tcx>, error: NextSolverError<'tcx>) -> Self {
-        match error {
-            NextSolverError::TrueError(obligation) => {
-                fulfillment_error_for_no_solution(infcx, obligation)
-            }
-            NextSolverError::Ambiguity(obligation) => {
-                fulfillment_error_for_stalled(infcx, obligation)
-            }
-            NextSolverError::Overflow(obligation) => {
-                fulfillment_error_for_overflow(infcx, obligation)
-            }
-        }
-    }
-}
-
-impl<'tcx> FromSolverError<'tcx, NextSolverError<'tcx>> for ScrubbedTraitError<'tcx> {
-    fn from_solver_error(_infcx: &InferCtxt<'tcx>, error: NextSolverError<'tcx>) -> Self {
-        match error {
-            NextSolverError::TrueError(_) => ScrubbedTraitError::TrueError,
-            NextSolverError::Ambiguity(_) | NextSolverError::Overflow(_) => {
-                ScrubbedTraitError::Ambiguity
-            }
-        }
-    }
-}
-
-// Some types are used a lot. Make sure they don't unintentionally get bigger.
-#[cfg(target_pointer_width = "64")]
-mod size_asserts {
-    use rustc_data_structures::static_assert_size;
-
-    use super::*;
-    // tidy-alphabetical-start
-    // Before #160005 this pair was greater than 128 bytes, which triggered the use of (slow)
-    // `memcpy` for moving elements of `PendingObligations`.
-    static_assert_size!((PredicateObligation<'_>, Option<GoalStalledOn<TyCtxt<'_>>>), 104);
-    // tidy-alphabetical-end
-}
diff --git a/compiler/rustc_trait_selection/src/solve/rustc_fulfill.rs b/compiler/rustc_trait_selection/src/solve/rustc_fulfill.rs
new file mode 100644
index 0000000000000..73efd327034b9
--- /dev/null
+++ b/compiler/rustc_trait_selection/src/solve/rustc_fulfill.rs
@@ -0,0 +1,289 @@
+use std::marker::PhantomData;
+
+use rustc_infer::infer::InferCtxt;
+use rustc_infer::traits::query::NoSolution;
+use rustc_infer::traits::{
+    FromSolverError, PredicateObligation, PredicateObligations, TraitEngine, TraitErrors,
+};
+use rustc_middle::ty::{self, TyCtxt, TypeVisitableExt, TypingMode};
+use rustc_next_trait_solver::solve::fulfill::{
+    FulfillmentCtxt as SolverFulfillmentCtxt, NextSolverError as SolverNextSolverError,
+};
+use rustc_next_trait_solver::solve::{GoalEvaluation, MaybeInfo, StalledOnCoroutines};
+use tracing::instrument;
+
+use self::derive_errors::*;
+use super::Certainty;
+use super::delegate::SolverDelegate;
+use crate::traits::{FulfillmentError, ScrubbedTraitError};
+
+mod derive_errors;
+
+/// A trait engine using the new trait solver.
+///
+/// The frontend wrapper keeps rustc-specific snapshot checks, diagnostics,
+/// and successful-obligation handling outside of the shared engine.
+pub struct FulfillmentCtxt<'tcx, E: 'tcx> {
+    core: SolverFulfillmentCtxt<TyCtxt<'tcx>, PredicateObligation<'tcx>>,
+
+    /// The snapshot in which this context was created. Using the context
+    /// outside of this snapshot leads to subtle bugs if the snapshot
+    /// gets rolled back. Because of this we explicitly check that we only
+    /// use the context in exactly this snapshot.
+    usable_in_snapshot: usize,
+    _errors: PhantomData<E>,
+}
+
+impl<'tcx, E: 'tcx> FulfillmentCtxt<'tcx, E> {
+    pub fn new(infcx: &InferCtxt<'tcx>) -> FulfillmentCtxt<'tcx, E> {
+        assert!(
+            infcx.next_trait_solver(),
+            "new trait solver fulfillment context created when \
+            infcx is set up for old trait solver"
+        );
+        FulfillmentCtxt {
+            core: Default::default(),
+            usable_in_snapshot: infcx.num_open_snapshots(),
+            _errors: PhantomData,
+        }
+    }
+
+    fn inspect_evaluated_obligation(
+        infcx: &InferCtxt<'tcx>,
+        obligation: &PredicateObligation<'tcx>,
+        result: &Result<GoalEvaluation<TyCtxt<'tcx>>, NoSolution>,
+    ) {
+        if let Some(inspector) = infcx.obligation_inspector.get() {
+            let result = match result {
+                Ok(GoalEvaluation { certainty, .. }) => Ok(*certainty),
+                Err(NoSolution) => Err(NoSolution),
+            };
+            (inspector)(infcx, &obligation, result);
+        }
+    }
+}
+
+impl<'tcx> From<SolverNextSolverError<PredicateObligation<'tcx>>> for NextSolverError<'tcx> {
+    fn from(error: SolverNextSolverError<PredicateObligation<'tcx>>) -> Self {
+        match error {
+            SolverNextSolverError::TrueError(obligation) => Self::TrueError(obligation),
+            SolverNextSolverError::Ambiguity(obligation) => Self::Ambiguity(obligation),
+            SolverNextSolverError::Overflow(obligation) => Self::Overflow(obligation),
+        }
+    }
+}
+
+impl<'tcx, E> TraitEngine<'tcx, E> for FulfillmentCtxt<'tcx, E>
+where
+    E: FromSolverError<'tcx, NextSolverError<'tcx>>,
+{
+    #[instrument(level = "trace", skip(self, infcx))]
+    fn register_predicate_obligation(
+        &mut self,
+        infcx: &InferCtxt<'tcx>,
+        obligation: PredicateObligation<'tcx>,
+    ) {
+        assert_eq!(self.usable_in_snapshot, infcx.num_open_snapshots());
+
+        let delegate = <&SolverDelegate<'tcx>>::from(infcx);
+        self.core.register(delegate, obligation);
+    }
+
+    #[inline]
+    fn collect_remaining_errors(&mut self, infcx: &InferCtxt<'tcx>) -> TraitErrors<E> {
+        if !self.core.has_pending_obligations() {
+            TraitErrors::NoErrors
+        } else {
+            TraitErrors::HasErrors(
+                self.core
+                    .collect_remaining_errors(|error| E::from_solver_error(infcx, error.into())),
+            )
+        }
+    }
+
+    fn try_evaluate_obligations(&mut self, infcx: &InferCtxt<'tcx>) -> TraitErrors<E> {
+        assert_eq!(self.usable_in_snapshot, infcx.num_open_snapshots());
+
+        let delegate = <&SolverDelegate<'tcx>>::from(infcx);
+
+        let errors = self.core.try_evaluate_obligations(
+            delegate,
+            |obligation, _, result| {
+                Self::inspect_evaluated_obligation(infcx, obligation, result);
+            },
+            |obligation| {
+                // Goals may depend on structural identity. Region uniquification at the
+                // start of MIR borrowck may cause things to no longer be so, potentially
+                // causing an ICE.
+                //
+                // While we uniquify root goals in HIR this does not handle cases where
+                // regions are hidden inside of a type or const inference variable.
+                //
+                // FIXME(-Znext-solver): This does not handle inference variables hidden
+                // inside of an opaque type, e.g. if there's `Opaque = (?x, ?x)` in the
+                // storage, we can also rely on structural identity of `?x` even if we
+                // later uniquify it in MIR borrowck.
+                if infcx.in_hir_typeck
+                    && (obligation.has_non_region_infer() || obligation.has_free_regions())
+                {
+                    infcx.push_hir_typeck_potentially_region_dependent_goal(obligation.clone());
+                }
+            },
+        );
+
+        TraitErrors::from_iter(
+            errors
+                .into_iter()
+                .map(NextSolverError::from)
+                .map(|error| E::from_solver_error(infcx, error)),
+        )
+    }
+
+    fn has_pending_obligations(&self) -> bool {
+        self.core.has_pending_obligations()
+    }
+
+    fn pending_obligations(&self) -> PredicateObligations<'tcx> {
+        self.core.pending_obligations()
+    }
+
+    fn pending_obligations_potentially_referencing_sub_root(
+        &self,
+        infcx: &InferCtxt<'tcx>,
+        vid: ty::TyVid,
+    ) -> PredicateObligations<'tcx> {
+        // `-Zdisable-fast-paths`: same gate as the other new-solver fast paths.
+        if infcx.tcx.disable_trait_solver_fast_paths() {
+            return self.pending_obligations();
+        }
+
+        self.core.pending_obligations_filtered(|_, stalled_on| {
+            let Some(stalled_on) = stalled_on else {
+                return true;
+            };
+
+            // Don't reuse the sub-unification roots cached on `stalled_on`:
+            // a later sub-unification merge can have changed which root
+            // each stalled var belongs to, so the cached info can be stale.
+            // Walk `stalled_vars` and recompute the current root instead.
+            //
+            // Conservative here: if a stalled var no longer resolves to an
+            // infer var, some unification happened, so the goal is no longer
+            // stalled. Include it to be re-evaluated downstream.
+            stalled_on.stalled_vars.iter().filter_map(|arg| arg.as_type()).any(|ty| {
+                match *infcx.shallow_resolve(ty).kind() {
+                    ty::Infer(ty::TyVar(tv)) => infcx.sub_unification_table_root_var(tv) == vid,
+                    _ => true,
+                }
+            })
+        })
+    }
+
+    fn pending_obligations_potentially_referencing_float_infer(
+        &self,
+        infcx: &InferCtxt<'tcx>,
+    ) -> PredicateObligations<'tcx> {
+        // `-Zdisable-fast-paths`: same gate as the other new-solver fast paths.
+        if infcx.tcx.disable_trait_solver_fast_paths() {
+            return self.pending_obligations();
+        }
+
+        self.core.pending_obligations_filtered(|_, stalled_on| {
+            let Some(stalled_on) = stalled_on else {
+                return true;
+            };
+
+            // If the stalled vars don't have float infers, the nested goals
+            // won't have them either. We only create float infers for
+            // user-written literals.
+            stalled_on
+                .stalled_vars
+                .iter()
+                .filter_map(|arg| arg.as_type())
+                .any(|ty| matches!(infcx.shallow_resolve(ty).kind(), ty::Infer(ty::FloatVar(_))))
+        })
+    }
+
+    fn drain_stalled_obligations_for_coroutines(
+        &mut self,
+        infcx: &InferCtxt<'tcx>,
+    ) -> PredicateObligations<'tcx> {
+        let stalled_coroutines = match infcx.typing_mode_raw().assert_not_erased() {
+            TypingMode::Typeck { defining_opaque_types_and_generators } => {
+                defining_opaque_types_and_generators
+            }
+
+            TypingMode::Coherence
+            | TypingMode::PostTypeckUntilBorrowck { defining_opaque_types: _ }
+            | TypingMode::PostBorrowck { defined_opaque_types: _ }
+            | TypingMode::Reflection
+            | TypingMode::PostAnalysis
+            | TypingMode::Codegen => return Default::default(),
+        };
+
+        if stalled_coroutines.is_empty() {
+            return Default::default();
+        }
+
+        self.core.drain_pending_obligations(|_, stalled_on| {
+            stalled_on.as_ref().is_some_and(|stalled_on| {
+                matches!(
+                    stalled_on.stalled_certainty,
+                    Certainty::Maybe(MaybeInfo {
+                        cause: _,
+                        opaque_types_jank: _,
+                        stalled_on_coroutines: StalledOnCoroutines::Yes,
+                    })
+                )
+            })
+        })
+    }
+}
+
+pub enum NextSolverError<'tcx> {
+    TrueError(PredicateObligation<'tcx>),
+    Ambiguity(PredicateObligation<'tcx>),
+    Overflow(PredicateObligation<'tcx>),
+}
+
+impl<'tcx> FromSolverError<'tcx, NextSolverError<'tcx>> for FulfillmentError<'tcx> {
+    fn from_solver_error(infcx: &InferCtxt<'tcx>, error: NextSolverError<'tcx>) -> Self {
+        match error {
+            NextSolverError::TrueError(obligation) => {
+                fulfillment_error_for_no_solution(infcx, obligation)
+            }
+            NextSolverError::Ambiguity(obligation) => {
+                fulfillment_error_for_stalled(infcx, obligation)
+            }
+            NextSolverError::Overflow(obligation) => {
+                fulfillment_error_for_overflow(infcx, obligation)
+            }
+        }
+    }
+}
+
+impl<'tcx> FromSolverError<'tcx, NextSolverError<'tcx>> for ScrubbedTraitError<'tcx> {
+    fn from_solver_error(_infcx: &InferCtxt<'tcx>, error: NextSolverError<'tcx>) -> Self {
+        match error {
+            NextSolverError::TrueError(_) => ScrubbedTraitError::TrueError,
+            NextSolverError::Ambiguity(_) | NextSolverError::Overflow(_) => {
+                ScrubbedTraitError::Ambiguity
+            }
+        }
+    }
+}
+
+// Some types are used a lot. Make sure they don't unintentionally get bigger.
+#[cfg(target_pointer_width = "64")]
+mod size_asserts {
+    use rustc_data_structures::static_assert_size;
+
+    use super::*;
+    use crate::solve::GoalStalledOn;
+
+    // tidy-alphabetical-start
+    // Before #160005 this pair was greater than 128 bytes, which triggered the use of (slow)
+    // `memcpy` for moving elements of `PendingObligations`.
+    static_assert_size!((PredicateObligation<'_>, Option<GoalStalledOn<TyCtxt<'_>>>), 104);
+    // tidy-alphabetical-end
+}
diff --git a/compiler/rustc_trait_selection/src/solve/fulfill/derive_errors.rs b/compiler/rustc_trait_selection/src/solve/rustc_fulfill/derive_errors.rs
similarity index 100%
rename from compiler/rustc_trait_selection/src/solve/fulfill/derive_errors.rs
rename to compiler/rustc_trait_selection/src/solve/rustc_fulfill/derive_errors.rs
diff --git a/compiler/rustc_type_ir/src/solve/fulfill.rs b/compiler/rustc_type_ir/src/solve/fulfill.rs
new file mode 100644
index 0000000000000..56ebfff452d45
--- /dev/null
+++ b/compiler/rustc_type_ir/src/solve/fulfill.rs
@@ -0,0 +1,19 @@
+use super::Goal;
+use crate::Interner;
+
+/// An obligation that can be processed by the shared fulfillment engine.
+///
+/// The shared engine only accesses the parts needed for fulfillment through this trait.
+pub trait FulfillmentObligation<I: Interner>: Clone {
+    fn as_goal(&self) -> Goal<I, I::Predicate>;
+
+    fn span(&self) -> I::Span;
+
+    fn recursion_depth(&self) -> usize;
+
+    fn set_recursion_depth(&mut self, depth: usize);
+
+    /// Stores the eagerly resolved predicate returned by the solver so that
+    /// later fulfillment iterations do not repeat that work.
+    fn set_predicate(&mut self, predicate: I::Predicate);
+}
diff --git a/compiler/rustc_type_ir/src/solve/mod.rs b/compiler/rustc_type_ir/src/solve/mod.rs
index a916dbd079efa..22e59e958d6cd 100644
--- a/compiler/rustc_type_ir/src/solve/mod.rs
+++ b/compiler/rustc_type_ir/src/solve/mod.rs
@@ -1,3 +1,4 @@
+pub mod fulfill;
 pub mod inspect;
 
 use std::convert::Infallible;