make call-scope context a required solve boundary

Summary:
This is part of the solver-boundary plan: constraint solving is usually per-call, but we also rely on ad-hoc subset boundaries where call-like behavior can occur with residual tracking disabled, so we must make boundary ownership explicit and always drain call-scoped state before leaving the boundary. This change makes callable finishing consume the active call context, adds explicit boundary-consumption state with debug-time drop invariants, and tracks fresh quantified vars created during call-scoped solving so boundary finishing still sees them even when residual reachability is lost.

NOTE: there's a lot of new code here and not much deltion, but a lot of the new logic is just the preparation / plumbing phase of a mini-stack that shifts responsibility for finishing around - there's offsetting deletion of a lot of legacy finishing code in D103337567 (which is the end goal - that commit fixes the `reduce` regression by storing all fresh vars in context and finishing them at the appropriate time).

Reviewed By: rchen152

Differential Revision: D103328383

fbshipit-source-id: d5bf4bbff28154cb64f7365a9e64c66c78fe90fd

Author: stroxler

pyrefly/lib/alt/callable.rs

@@ -1385,6 +1385,7 @@ impl<'a, Ans: LookupAnswer> AnswersSolver<'a, Ans> {
     ) {
         let mut call_context = CallContext::outside();
         call_context.set_argument_side(ArgumentSide::Got);
+        call_context.require_boundary_consumption();
 
         // Look up meta-shape early so we can conditionally collect bound args.
         // Only consult the registry when tensor_shapes is enabled to avoid
@@ -1422,6 +1423,7 @@ impl<'a, Ans: LookupAnswer> AnswersSolver<'a, Ans> {
         } else {
             (QuantifiedHandle::empty(), callable)
         };
+        call_context.register_fresh_quantified_vars(callable_qs.vars());
         let (self_qs, remaining_callable_qs) = if self_obj.is_some()
             && let Some(first_param) = callable.get_first_param()
             // TODO(https://github.com/facebook/pyrefly/issues/105): handle nested vars
@@ -1437,6 +1439,7 @@ impl<'a, Ans: LookupAnswer> AnswersSolver<'a, Ans> {
         };
         let ctor_qs = if let Some(targs) = ctor_targs.as_mut() {
             let qs = self.solver().freshen_class_targs(targs, self.uniques);
+            call_context.register_fresh_quantified_vars(qs.vars());
             let mp = targs.substitution_map();
             callable.params.visit_mut(&mut |t| t.subst_mut(&mp));
             if let Some(obj) = self_obj.as_mut() {
@@ -1564,10 +1567,11 @@ impl<'a, Ans: LookupAnswer> AnswersSolver<'a, Ans> {
         }
         let mut errors = self
             .solver()
-            .finish_quantified_with_type_order(
+            .finish_quantified_with_type_order_and_call_context(
                 remaining_callable_qs,
                 self.solver().infer_with_first_use,
                 self.type_order(),
+                &call_context,
             )
             .map_or_else(|e| e.to_vec(), |_| Vec::new());
         if let Err(e) = self.solver().finish_quantified_with_type_order(

pyrefly/lib/solver/solver.rs

@@ -17,6 +17,8 @@ use std::hash::Hash;
 use std::hash::Hasher;
 use std::mem;
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 
 use itertools::Either;
 use itertools::Itertools;
@@ -315,6 +317,10 @@ impl QuantifiedHandle {
         Self(Vec::new())
     }
 
+    pub(crate) fn vars(&self) -> &[Var] {
+        &self.0
+    }
+
     /// Split the handle into (vars in ty, vars not in ty)
     pub fn partition_by(self, ty: &Type) -> (Self, Self) {
         let vars_in_ty = ty.collect_maybe_placeholder_vars();
@@ -2570,6 +2576,118 @@ impl Solver {
         })
     }
 
+    /// Finish quantified vars at a call boundary by consuming tracked fresh
+    /// quantified vars from `CallContext`, then finishing the reachable
+    /// quantified closure from explicit roots plus tracked roots.
+    pub fn finish_quantified_with_type_order_and_call_context<Ans: LookupAnswer>(
+        &self,
+        vs: QuantifiedHandle,
+        infer_with_first_use: bool,
+        type_order: TypeOrder<Ans>,
+        call_context: &CallContext,
+    ) -> Result<(), Vec1<TypeVarSpecializationError>> {
+        let tracked_fresh_vars = call_context.take_deferred_quantified_vars();
+        let overload_witness_payloads = call_context.take_overload_witness_payloads();
+        call_context.mark_boundary_consumed_and_drained();
+        let payload_vars: SmallSet<Var> = overload_witness_payloads
+            .values()
+            .flat_map(|branch_captures| branch_captures.iter())
+            .flat_map(|capture| capture.values.keys().copied())
+            .collect();
+        let mut roots: SmallSet<Var> = vs.0.into_iter().collect();
+        roots.extend(tracked_fresh_vars.0);
+        // Forall instantiation during call analysis can unify call-scope vars
+        // with additional fresh vars that are only visible through Answer /
+        // ResidualAnswer payloads. Finish the full reachable closure so no
+        // reachable Variable::Quantified can leak to pinning.
+        //
+        // We also must include reachable vars that already became Answer but
+        // still have pending instantiation errors. Those errors are surfaced by
+        // finish_quantified, so excluding Answer vars here can silently drop
+        // call-site specialization diagnostics.
+        let roots = roots.into_iter().collect::<Vec<_>>();
+        let mut already_finished: SmallSet<Var> = SmallSet::new();
+        loop {
+            // Fixed-point: finishing can mutate solver state and expose new
+            // reachable vars that also require finishing.
+            let reachable_finish_vars = self.reachable_finish_vars_from_roots(&roots);
+            let mut next_round: Vec<Var> = reachable_finish_vars
+                .into_iter()
+                .filter(|var| !already_finished.contains(var))
+                .collect();
+            // Payload-driven overload pruning must consider solved vars even if
+            // they already collapsed to `Answer` before finishing and therefore
+            // are not selected by reachability-based finishing alone.
+            next_round.extend(
+                payload_vars
+                    .iter()
+                    .copied()
+                    .filter(|var| !already_finished.contains(var)),
+            );
+            next_round.sort_unstable();
+            next_round.dedup();
+            if next_round.is_empty() {
+                break;
+            }
+            already_finished.extend(next_round.iter().copied());
+            let mut subset = self.subset(type_order);
+            self.finish_quantified_with_subset_and_payloads(
+                QuantifiedHandle(next_round),
+                infer_with_first_use,
+                &mut |got, want| subset.is_subset_eq_probe_for_pruning(got, want),
+                Some(&overload_witness_payloads),
+            )?;
+        }
+        Ok(())
+    }
+
+    fn reachable_finish_vars_from_roots(&self, roots: &[Var]) -> SmallSet<Var> {
+        if roots.is_empty() {
+            return SmallSet::new();
+        }
+        let variables = self.variables.lock();
+        let instantiation_errors = self.instantiation_errors.read();
+        let mut visited: SmallSet<Var> = SmallSet::new();
+        let mut reachable_finish_vars: SmallSet<Var> = SmallSet::new();
+        let mut stack = roots.to_vec();
+        while let Some(var) = stack.pop() {
+            if !visited.insert(var) {
+                continue;
+            }
+            let variable = variables.get(var);
+            let needs_finish = match &*variable {
+                Variable::Quantified { .. } => true,
+                Variable::Answer(_) | Variable::ResidualAnswer { .. } => {
+                    instantiation_errors.contains_key(&var)
+                }
+                _ => false,
+            };
+            if needs_finish {
+                reachable_finish_vars.insert(var);
+            }
+            match &*variable {
+                Variable::Answer(ty) => {
+                    stack.extend(ty.collect_maybe_placeholder_vars());
+                }
+                Variable::ResidualAnswer { target_vars: _, ty } => {
+                    // `target_vars` are read-gates for residual visibility,
+                    // not ownership edges for finishing reachability.
+                    stack.extend(ty.collect_maybe_placeholder_vars());
+                }
+                Variable::Quantified { .. } => {}
+                Variable::Unwrap(bounds) => {
+                    for ty in bounds.lower.iter().chain(bounds.upper.iter()) {
+                        stack.extend(ty.collect_maybe_placeholder_vars());
+                    }
+                }
+                Variable::PartialQuantified(_)
+                | Variable::PartialContained(_)
+                | Variable::Recursive => {}
+            }
+        }
+        reachable_finish_vars
+    }
+
     /// Finish all quantified vars reachable from `ty` using the solver default
     /// inference mode.
     ///
@@ -3499,6 +3617,10 @@ pub struct CallContext {
     /// Invariant: payload entries are scoped to this call-context lineage and
     /// must not leak across `with_outside_call_context` boundaries.
     overload_witness_payloads: Arc<Mutex<OverloadWitnessPayloadByHash>>,
+    /// Whether this context must be consumed at a solve boundary.
+    require_boundary_consumption: Arc<AtomicBool>,
+    /// Whether deferred state from this context lineage was consumed/drained.
+    boundary_consumed_and_drained: Arc<AtomicBool>,
 }
 
 impl Default for CallContext {
@@ -3508,6 +3630,8 @@ impl Default for CallContext {
             argument_side: ArgumentSide::default(),
             deferred_quantified_vars: Arc::new(Mutex::new(SmallSet::new())),
             overload_witness_payloads: Arc::new(Mutex::new(SmallMap::new())),
+            require_boundary_consumption: Arc::new(AtomicBool::new(false)),
+            boundary_consumed_and_drained: Arc::new(AtomicBool::new(false)),
         }
     }
 }
@@ -3521,6 +3645,18 @@ impl CallContext {
         self.argument_side = argument_side;
     }
 
+    pub(crate) fn register_fresh_quantified_vars(&self, vars: &[Var]) {
+        let mut deferred_quantified_vars = self.deferred_quantified_vars.lock();
+        deferred_quantified_vars.extend(vars.iter().copied());
+    }
+
+    pub fn require_boundary_consumption(&self) {
+        self.require_boundary_consumption
+            .store(true, Ordering::Relaxed);
+        self.boundary_consumed_and_drained
+            .store(false, Ordering::Relaxed);
+    }
+
     pub fn residual_witness(&self) -> Option<&ResidualWitnessContext> {
         self.witness.as_ref()
     }
@@ -3556,6 +3692,53 @@ impl CallContext {
             SubsetCacheContext::Default
         }
     }
+
+    pub(crate) fn take_deferred_quantified_vars(&self) -> QuantifiedHandle {
+        let mut deferred_quantified_vars = self.deferred_quantified_vars.lock();
+        QuantifiedHandle(
+            mem::take(&mut *deferred_quantified_vars)
+                .into_iter()
+                .collect(),
+        )
+    }
+
+    pub(crate) fn take_overload_witness_payloads(&self) -> OverloadWitnessPayloadByHash {
+        let mut overload_witness_payloads = self.overload_witness_payloads.lock();
+        mem::take(&mut *overload_witness_payloads)
+    }
+
+    fn mark_boundary_consumed_and_drained(&self) {
+        self.boundary_consumed_and_drained
+            .store(true, Ordering::Relaxed);
+    }
+}
+
+impl Drop for CallContext {
+    fn drop(&mut self) {
+        #[cfg(debug_assertions)]
+        {
+            if std::thread::panicking()
+                || Arc::strong_count(&self.require_boundary_consumption) != 1
+            {
+                return;
+            }
+            if !self.require_boundary_consumption.load(Ordering::Relaxed) {
+                return;
+            }
+            assert!(
+                self.boundary_consumed_and_drained.load(Ordering::Relaxed),
+                "CallContext dropped without boundary consume/drain",
+            );
+            assert!(
+                self.deferred_quantified_vars.lock().is_empty(),
+                "CallContext dropped with deferred quantified vars still pending",
+            );
+            assert!(
+                self.overload_witness_payloads.lock().is_empty(),
+                "CallContext dropped with overload witness payloads still pending",
+            );
+        }
+    }
 }
 
 /// A helper to implement subset ergonomically.
@@ -3704,11 +3887,21 @@ impl<'a, Ans: LookupAnswer> Subset<'a, Ans> {
             &mut self.active_call_context.overload_witness_payloads,
             call_context.overload_witness_payloads.clone(),
         );
+        let old_require_boundary_consumption = mem::replace(
+            &mut self.active_call_context.require_boundary_consumption,
+            call_context.require_boundary_consumption.clone(),
+        );
+        let old_boundary_consumed_and_drained = mem::replace(
+            &mut self.active_call_context.boundary_consumed_and_drained,
+            call_context.boundary_consumed_and_drained.clone(),
+        );
         let res = f(self);
         self.active_call_context.witness = old_witness;
         self.active_call_context.argument_side = old_argument_side;
         self.active_call_context.deferred_quantified_vars = old_deferred_quantified_vars;
         self.active_call_context.overload_witness_payloads = old_overload_witness_payloads;
+        self.active_call_context.require_boundary_consumption = old_require_boundary_consumption;
+        self.active_call_context.boundary_consumed_and_drained = old_boundary_consumed_and_drained;
         res
     }
 
@@ -3746,18 +3939,16 @@ impl<'a, Ans: LookupAnswer> Subset<'a, Ans> {
             &mut self.active_call_context.argument_side,
             ArgumentSide::NotAnalyzingACall,
         );
-        let old_deferred_quantified_vars = mem::replace(
-            &mut self.active_call_context.deferred_quantified_vars,
-            Arc::new(Mutex::new(SmallSet::new())),
-        );
+        // Keep fresh-var tracking attached to the same boundary while
+        // temporarily disabling residual hooks. Fresh quantified vars created in
+        // this scope must still be finished when the outer boundary drains.
         let old_overload_witness_payloads = mem::replace(
             &mut self.active_call_context.overload_witness_payloads,
             Arc::new(Mutex::new(SmallMap::new())),
         );
         let res = f(self);
         self.active_call_context.witness = old_witness;
         self.active_call_context.argument_side = old_argument_side;
-        self.active_call_context.deferred_quantified_vars = old_deferred_quantified_vars;
         self.active_call_context.overload_witness_payloads = old_overload_witness_payloads;
         res
     }

pyrefly/lib/solver/subset.rs

@@ -2248,6 +2248,8 @@ impl<'a, Ans: LookupAnswer> Subset<'a, Ans> {
                 let forall_type = Type::Forall(forall.clone());
                 // Finalizing the quantified vars returns instantiation errors
                 let (vs, got) = self.type_order.instantiate_fresh_forall((**forall).clone());
+                self.active_call_context
+                    .register_fresh_quantified_vars(vs.vars());
                 let argument_side = self.active_argument_side();
                 let witness = self.make_forall_witness(&forall_type, &vs, want);
                 let (result, mut maybe_witness) =

pyrefly/lib/test/callable_residuals.rs

@@ -602,7 +602,6 @@ reveal_type(result)  # E: revealed type: () -> tuple[int, int]
 );
 
 testcase!(
-    bug = "Constrained type vars can collapse to Answer before finishing, so overload pruning can miss them",
     test_overload_pruning_ignored_for_constrained_tvar_solved_early,
     r#"
 from typing import Callable, overload, reveal_type
@@ -615,7 +614,7 @@ def f(x: float) -> float: ...  # E: Overload return type `float` is not assignab
 def f(x: bytes) -> bytes: ...  # E: Overload return type `bytes` is not assignable to implementation return type `None`
 def f(x): ...
 
-result = project(f, 1)
+result = project(f, 1)  # E: Overload type was not compatible with solved type variables: unknown = int
 reveal_type(result)  # E: revealed type: (int) -> int
 "#,
 );