fix(resolver): exempt top-level pinned versions from transitive cooldown

src/fromager/__main__.py

@@ -283,11 +283,7 @@ def main(
         network_isolation=network_isolation,
         max_jobs=jobs,
         settings_dir=settings_dir,
-        cooldown=(
-            candidate.Cooldown(min_age=datetime.timedelta(days=min_release_age))
-            if min_release_age > 0
-            else None
-        ),
+        cooldown=candidate.Cooldown(min_age=datetime.timedelta(days=min_release_age)),
     )
     wkctx.setup()
     ctx.obj = wkctx

src/fromager/candidate.py

@@ -15,18 +15,34 @@
 logger = logging.getLogger(__name__)
 
 
-@dataclasses.dataclass
+@dataclasses.dataclass(frozen=True)
 class Cooldown:
     """Policy for rejecting recently-published package versions.
 
+    Frozen so that cooldown policy cannot be accidentally weakened after
+    construction — all parameters are set once and shared read-only.
+
+    A cooldown with ``min_age`` of zero (or negative) is effectively disabled —
+    every package age exceeds the threshold.  Use :meth:`disabled` as a
+    convenient factory instead of ``None``.
+
     bootstrap_time is fixed at construction so all resolutions in a single run
     share the same cutoff.
+
+    exempt_versions bypasses the age check for specific versions that were
+    already approved via a top-level exact pin.
     """
 
     min_age: datetime.timedelta
     bootstrap_time: datetime.datetime = dataclasses.field(
         default_factory=lambda: datetime.datetime.now(datetime.UTC)
     )
+    exempt_versions: frozenset[Version] = dataclasses.field(default_factory=frozenset)
+
+    @classmethod
+    def disabled(cls) -> "Cooldown":
+        """Return a cooldown that never filters any candidate."""
+        return cls(min_age=datetime.timedelta(0))
 
 
 @dataclasses.dataclass(frozen=True, order=True, slots=True, repr=False, kw_only=True)

src/fromager/context.py

@@ -20,9 +20,10 @@
     external_commands,
     packagesettings,
 )
+from .candidate import Cooldown
 
 if typing.TYPE_CHECKING:
-    from . import build_environment, candidate
+    from . import build_environment
 
 logger = logging.getLogger(__name__)
 
@@ -47,7 +48,7 @@ def __init__(
         max_jobs: int | None = None,
         settings_dir: pathlib.Path | None = None,
         wheel_server_url: str = "",
-        cooldown: candidate.Cooldown | None = None,
+        cooldown: Cooldown | None = None,
         max_release_age: datetime.timedelta | None = None,
     ):
         if active_settings is None:
@@ -95,7 +96,9 @@ def __init__(
 
         self._parallel_builds = False
 
-        self.cooldown: candidate.Cooldown | None = cooldown
+        self.cooldown: Cooldown = (
+            cooldown if cooldown is not None else Cooldown.disabled()
+        )
         self._max_release_age: datetime.timedelta | None = max_release_age
 
     @property

src/fromager/finders.py

@@ -61,7 +61,6 @@ def __init__(
             ignore_platform=False,
             use_resolver_cache=use_resolver_cache,
             override_download_url=None,
-            cooldown=None,
             supports_upload_time=False,
         )
 

src/fromager/resolver.py

@@ -147,42 +147,78 @@ def _has_equality_pin(req: Requirement) -> bool:
     return len(specs) == 1 and specs[0].operator == "==" and "*" not in specs[0].version
 
 
+def _get_toplevel_pinned_versions(
+    ctx: context.WorkContext, req: Requirement
+) -> frozenset[Version]:
+    """Return versions of *req* that have a top-level exact ``==`` pin in the graph."""
+    top_level_edges = ctx.dependency_graph.get_root_node().get_outgoing_edges(
+        req.name, RequirementType.TOP_LEVEL
+    )
+    return frozenset(
+        edge.destination_node.version
+        for edge in top_level_edges
+        if _has_equality_pin(edge.req)
+    )
+
+
+def _resolve_cooldown_params(
+    ctx: context.WorkContext,
+    req: Requirement,
+) -> tuple[datetime.timedelta, datetime.datetime]:
+    """Resolve min_age and bootstrap_time for a package's cooldown.
+
+    Always returns a ``(min_age, bootstrap_time)`` pair.  When cooldown is
+    disabled (per-package override of 0, or no global/per-package config),
+    ``min_age`` will be zero — every package age exceeds that threshold.
+    """
+    min_age_override = ctx.package_build_info(req).resolver_min_release_age
+
+    if min_age_override == 0:
+        min_age = datetime.timedelta(0)
+    elif min_age_override is not None:
+        min_age = datetime.timedelta(days=min_age_override)
+    else:
+        min_age = ctx.cooldown.min_age
+
+    return min_age, ctx.cooldown.bootstrap_time
+
+
 def resolve_package_cooldown(
     ctx: context.WorkContext,
     req: Requirement,
     req_type: RequirementType | None = None,
-) -> Cooldown | None:
+) -> Cooldown:
     """Compute the effective cooldown for a single package.
 
-    Args:
-        ctx: The current work context (provides the global cooldown).
-        req: The package requirement being resolved.
-        req_type: The requirement type (top-level, install, etc.).
+    Always returns a ``Cooldown`` instance.  A ``min_age`` of zero means
+    cooldown is effectively disabled — every package age exceeds zero.
+
+    Returns a *disabled* cooldown (min_age=0) when:
+
+    * The requirement is a top-level exact ``==`` pin — the user explicitly
+      approved that version.
+    * A per-package ``min_release_age=0`` override disables cooldown.
+    * No global cooldown is configured and no per-package override enables one.
 
-    Returns:
-        The cooldown to pass to the provider, or ``None`` if disabled.
+    Otherwise returns an *active* cooldown with:
+
+    * *min_age* from the per-package override (if set) or the global cooldown.
+    * *bootstrap_time* inherited from the global cooldown (for a consistent
+      cutoff across the entire run).
+    * *exempt_versions* populated from top-level exact-pinned entries in the
+      dependency graph, so transitive resolutions of the same package honour
+      the user's explicit pin.
     """
     if req_type == RequirementType.TOP_LEVEL and _has_equality_pin(req):
-        if ctx.cooldown is not None:
+        if ctx.cooldown.min_age > datetime.timedelta(0):
             logger.info("cooldown bypassed as the top-level requirement uses == pin")
-        return None
+        return Cooldown.disabled()
 
-    per_package_days = ctx.package_build_info(req).resolver_min_release_age
-    global_cooldown = ctx.cooldown
-    if per_package_days is None:
-        return global_cooldown
-    if per_package_days == 0:
-        return None
-    # Per-package positive override: inherit bootstrap_time from global so all
-    # resolutions in a single run share the same fixed cutoff point.
-    bootstrap_time = (
-        global_cooldown.bootstrap_time
-        if global_cooldown is not None
-        else datetime.datetime.now(datetime.UTC)
-    )
+    min_age, bootstrap_time = _resolve_cooldown_params(ctx, req)
     return Cooldown(
-        min_age=datetime.timedelta(days=per_package_days),
+        min_age=min_age,
         bootstrap_time=bootstrap_time,
+        exempt_versions=_get_toplevel_pinned_versions(ctx, req),
     )
 
 
@@ -196,12 +232,7 @@ def _compute_max_age_cutoff(
     """
     if ctx.max_release_age is None:
         return None
-    bootstrap_time = (
-        ctx.cooldown.bootstrap_time
-        if ctx.cooldown is not None
-        else datetime.datetime.now(datetime.UTC)
-    )
-    return bootstrap_time - ctx.max_release_age
+    return ctx.cooldown.bootstrap_time - ctx.max_release_age
 
 
 def extract_filename_from_url(url: str) -> str:
@@ -576,8 +607,9 @@ def __init__(
         self.req_type = req_type
         self.use_cache_candidates = use_resolver_cache
 
-        # cooldown specific settings
-        self.cooldown = cooldown
+        self.cooldown: Cooldown = (
+            cooldown if cooldown is not None else Cooldown.disabled()
+        )
         # Does this provider supply upload timestamps for candidates?
         # Defaults to False (safe/unknown). Subclasses that reliably populate
         # upload_time on every candidate should set this to True in their __init__.
@@ -699,11 +731,12 @@ def is_satisfied_by(self, requirement: Requirement, candidate: Candidate) -> boo
     def is_blocked_by_cooldown(self, candidate: Candidate) -> bool:
         """Return True if the candidate is rejected by the release-age cooldown."""
 
-        # a cooldown is not specified...
-        if self.cooldown is None:
+        if self.cooldown.min_age <= datetime.timedelta(0):
+            return False
+
+        if candidate.version in self.cooldown.exempt_versions:
             return False
 
-        # the target candidate doesn't provide a valid upload timestamp
         if candidate.upload_time is None:
             if not self.supports_upload_time:
                 # this provider does not yet support timestamp retrieval (e.g. GitHub).
@@ -947,7 +980,7 @@ def _get_no_match_error_message(
 
         # If a cooldown is active, check whether it's responsible for the
         # failure so we can give a more actionable error message.
-        if self.cooldown is not None:
+        if self.cooldown.min_age > datetime.timedelta(0):
             cutoff = self.cooldown.bootstrap_time - self.cooldown.min_age
             all_candidates = list(self._find_cached_candidates(identifier))
             missing_time = [c for c in all_candidates if c.upload_time is None]