diff --git a/packages/gnss-product-management/src/gnss_product_management/client/gnss_client.py b/packages/gnss-product-management/src/gnss_product_management/client/gnss_client.py
index 3ed9b85..4521063 100644
--- a/packages/gnss-product-management/src/gnss_product_management/client/gnss_client.py
+++ b/packages/gnss-product-management/src/gnss_product_management/client/gnss_client.py
@@ -281,6 +281,9 @@ def resolve_dependencies(
         date: datetime.datetime,
         *,
         sink_id: str,
+        force_download: bool = False,
+        centers: list[str] | None = None,
+        bundle_centers: list[str] | None = None,
     ) -> tuple[DependencyResolution, AnyPath | None]:
         """Resolve all dependencies in a spec for the given date.
 
@@ -297,6 +300,8 @@ def resolve_dependencies(
             date: Target date (timezone-aware datetime, midnight UTC).
             sink_id: Local resource alias for storing resolved files
                 (e.g. ``"local"``).
+            centers: Optional remote resource IDs to search.  When omitted,
+                all configured centers are eligible.
 
         Returns:
             A ``(DependencyResolution, lockfile_path)`` tuple.
@@ -312,4 +317,11 @@ def resolve_dependencies(
             workspace=self._workspace,
             transport=self._transport,
         )
-        return pipeline.run(dep_spec, date, sink_id=sink_id)
+        return pipeline.run(
+            dep_spec,
+            date,
+            sink_id=sink_id,
+            force_download=force_download,
+            centers=centers,
+            bundle_centers=bundle_centers,
+        )
diff --git a/packages/gnss-product-management/src/gnss_product_management/client/product_query.py b/packages/gnss-product-management/src/gnss_product_management/client/product_query.py
index 83f0352..9210dcd 100644
--- a/packages/gnss-product-management/src/gnss_product_management/client/product_query.py
+++ b/packages/gnss-product-management/src/gnss_product_management/client/product_query.py
@@ -24,6 +24,12 @@
 logger = logging.getLogger(__name__)
 
 
+def _remote_uri(protocol: str, hostname: str, directory: str, filename: str) -> str:
+    """Build a remote URI without duplicating schemes or path separators."""
+    base = hostname if "://" in hostname else f"{protocol}://{hostname}"
+    return f"{base.rstrip('/')}/{directory.strip('/')}/{filename}"
+
+
 class ProductQuery:
     """Fluent builder for constructing and executing a GNSS product search.
 
@@ -342,6 +348,13 @@ def search(self) -> list[FoundResource]:
                 continue
             seen[key] = True
             params = {p.name: p.value for p in rq.product.parameters if p.value is not None}
+            # Search targets may retain wildcard regexes for unconstrained
+            # filename axes (for example TTT/PPP on Wuhan ORBIT).  Replace
+            # those patterns with metadata parsed from the actual match so
+            # callers can group candidates by their real product family.
+            classified = self._search_planner._product_registry.classify(filename)
+            if classified and classified.get("product") == rq.product.name:
+                params.update(classified.get("parameters", {}))
             protocol = (rq.server.protocol or "").upper()
             is_local = protocol in ("FILE", "LOCAL")
             if is_local:
@@ -352,8 +365,11 @@ def search(self) -> list[FoundResource]:
                 )
             else:
                 proto = (rq.server.protocol or "ftp").lower()
-                uri = (
-                    f"{proto}://{hostname}/{rq.directory.value or rq.directory.pattern}/{filename}"  # type: ignore[union-attr]
+                uri = _remote_uri(
+                    proto,
+                    hostname,
+                    rq.directory.value or rq.directory.pattern,  # type: ignore[union-attr]
+                    filename,
                 )
             r = FoundResource(
                 product=rq.product.name,
diff --git a/packages/gnss-product-management/src/gnss_product_management/factories/connection_pool.py b/packages/gnss-product-management/src/gnss_product_management/factories/connection_pool.py
index 85ee2f3..0d4672e 100644
--- a/packages/gnss-product-management/src/gnss_product_management/factories/connection_pool.py
+++ b/packages/gnss-product-management/src/gnss_product_management/factories/connection_pool.py
@@ -2,11 +2,16 @@
 
 import logging
 import os
+import random
+import re
+import shutil
 import threading
+import time
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Optional
-from urllib.parse import urlparse
+from urllib.parse import quote, urlparse
+from urllib.request import Request, urlopen
 
 import fsspec
 import fsspec.utils
@@ -23,16 +28,24 @@ class ConnectionPool:
         max_connections: Maximum number of concurrent connections.
     """
 
-    def __init__(self, hostname: str, max_connections: int = 4):
+    def __init__(
+        self,
+        hostname: str,
+        max_connections: int = 4,
+        listing_url: str | None = None,
+    ):
         """Initialise a connection pool for *hostname*.
 
         Args:
             hostname: Server address or local path.
             max_connections: Maximum number of concurrent connections.
+            listing_url: Optional URL template for a separate HTML directory
+                listing. ``{directory}`` is replaced with the quoted path.
         """
         self.hostname = hostname
         self.protocol = fsspec.utils.get_protocol(hostname) or "file"
         self.max_connections = max_connections
+        self.listing_url = listing_url
         self._pool: list[fsspec.AbstractFileSystem] = []
         self._semaphore: threading.Semaphore | None = None
         self._pool_lock = threading.Lock()
@@ -193,15 +206,22 @@ def __init__(self, max_connections: int = 4):
         self._listing_cache: dict[str, list[str]] = {}
         self._listing_cache_lock = threading.Lock()
 
-    def add_connection(self, hostname: str):
+    def add_connection(self, hostname: str, *, listing_url: str | None = None):
         """Ensure a connection pool exists for *hostname*.
 
         Args:
             hostname: Server address to pool.
+            listing_url: Optional separate HTML directory-listing URL template.
         """
         with self._factory_lock:
             if hostname not in self._pools:
-                self._pools[hostname] = ConnectionPool(hostname, self.max_connections)
+                self._pools[hostname] = ConnectionPool(
+                    hostname,
+                    self.max_connections,
+                    listing_url=listing_url,
+                )
+            elif self._pools[hostname].listing_url != listing_url:
+                raise ValueError(f"Conflicting listing configuration for: {hostname}")
 
     @contextmanager
     def get_connection(self, hostname: str):
@@ -249,6 +269,14 @@ def list_directory(self, hostname: str, directory: str) -> list[str]:
         full_path = pool.full_path(directory)
 
         def _ls(conn: "fsspec.AbstractFileSystem") -> list[str]:
+            if pool.listing_url:
+                quoted_directory = quote(directory.strip("/"), safe="/")
+                listing_url = pool.listing_url.format(directory=quoted_directory)
+                html = conn.cat_file(listing_url)
+                if isinstance(html, bytes):
+                    html = html.decode("utf-8", errors="replace")
+                download_prefix = re.escape(f"{hostname.rstrip('/')}/{directory.strip('/')}/")
+                return sorted(set(re.findall(rf'href="{download_prefix}([^"/]+)"', html)))
             raw = conn.ls(full_path, detail=False)
             return [Path(p).name for p in raw]
 
@@ -335,6 +363,30 @@ def download_file(self, hostname: str, remote_path: str, target_dir: str) -> Pat
         filename = Path(remote_path).name
         local_path = Path(target_dir) / filename
 
+        # GithubFileSystem's Contents API path can return only the first 4 MiB
+        # of larger blobs.  Use GitHub's immutable raw-content endpoint for
+        # the actual transfer while retaining GithubFileSystem for listings.
+        if pool.protocol == "github":
+            match = re.fullmatch(r"github://([^:]+):([^@]+)@(.+)", hostname)
+            if match:
+                owner, repository, revision = match.groups()
+                raw_url = (
+                    f"https://raw.githubusercontent.com/{owner}/{repository}/"
+                    f"{revision}/{remote_path.lstrip('/')}"
+                )
+                try:
+                    request = Request(raw_url, headers={"Accept-Encoding": "identity"})
+                    with (
+                        urlopen(request, timeout=60) as source,
+                        local_path.open("wb") as destination,
+                    ):
+                        shutil.copyfileobj(source, destination, length=1024 * 1024)
+                    return local_path if local_path.stat().st_size > 0 else None
+                except Exception as exc:
+                    logger.warning("Raw GitHub download failed for %s: %s", raw_url, exc)
+                    local_path.unlink(missing_ok=True)
+                    return None
+
         def _get(conn: "fsspec.AbstractFileSystem") -> Path | None:
             conn.get(full_path, str(local_path))
             if local_path.exists() and local_path.stat().st_size > 0:
@@ -348,6 +400,8 @@ def _get(conn: "fsspec.AbstractFileSystem") -> Path | None:
                 return _get(conn)
             except (BrokenPipeError, ConnectionError, EOFError, OSError) as e:
                 logger.debug("Stale connection for %s, reconnecting: %s", hostname, e)
+                if pool.protocol != "file":
+                    time.sleep(random.uniform(1.0, 3.0))
                 fresh = pool.replace_connection(conn)
                 if fresh is None:
                     return None
diff --git a/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/download.py b/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/download.py
index 3a94b47..929ae84 100644
--- a/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/download.py
+++ b/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/download.py
@@ -61,6 +61,7 @@ def run(
         date: datetime.datetime,
         *,
         sink_id: str = "local_config",
+        force: bool = False,
     ) -> Path | None | list[Path | None]:
         """Download found resources to the workspace.
 
@@ -79,7 +80,7 @@ def run(
 
         paths: list[Path | None] = []
         for r in resources:
-            path = self._download_one(r, date, sink_id)
+            path = self._download_one(r, date, sink_id, force=force)
             paths.append(path)
 
         if single:
@@ -91,6 +92,8 @@ def _download_one(
         resource: FoundResource,
         date: datetime.datetime,
         sink_id: str,
+        *,
+        force: bool = False,
     ) -> Path | None:
         """Download a single resource and write its sidecar lockfile.
 
@@ -102,7 +105,7 @@ def _download_one(
         Returns:
             Path to the resolved file, or ``None`` on failure.
         """
-        if resource.is_local:
+        if resource.is_local and not force:
             local_path = resource.path
             if local_path and local_path.exists():
                 logger.debug("Already local: %s", local_path)
@@ -121,9 +124,10 @@ def _download_one(
             local_resource_id=sink_id,
             local_factory=self._planner._workspace,
             date=date,
+            force=force,
         )
         if path is not None:
-            if get_lock_product(path) is None:
+            if force or get_lock_product(path) is None:
                 lock = build_lock_product(sink=path, url=resource.uri, name=resource.product)
                 write_lock_product(lock)
             logger.info("Downloaded %s → %s", resource.product, path)
diff --git a/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/resolve.py b/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/resolve.py
index db301f6..4eca0fa 100644
--- a/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/resolve.py
+++ b/packages/gnss-product-management/src/gnss_product_management/factories/pipelines/resolve.py
@@ -15,11 +15,10 @@
 
 import datetime
 import logging
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from functools import partial
 
 from gnss_product_management.environments import ProductRegistry, WorkSpace
-from gnss_product_management.factories.models import FoundResource
 from gnss_product_management.factories.pipelines.download import DownloadPipeline
 from gnss_product_management.factories.pipelines.lockfile_writer import LockfileWriter
 from gnss_product_management.factories.remote_transport import WormHole
@@ -33,6 +32,7 @@
 )
 from gnss_product_management.specifications.dependencies.dependencies import (
     Dependency,
+    DependencyBundle,
     DependencyResolution,
     DependencySpec,
     ResolvedDependency,
@@ -77,6 +77,7 @@ def __init__(
         self._env = env
         self._workspace = workspace
         transport = transport or WormHole(max_connections=max_connections, product_registry=env)
+        self._transport = transport
         planner = SearchPlanner(product_registry=env, workspace=workspace)
         self._query = ProductQuery(wormhole=transport, search_planner=planner)
         self._downloader = DownloadPipeline(
@@ -93,7 +94,9 @@ def run(
         *,
         sink_id: str = "local_config",
         centers: list[str] | None = None,
+        bundle_centers: list[str] | None = None,
         download: bool = True,
+        force_download: bool = False,
     ) -> tuple[DependencyResolution, AnyPath | None]:
         """Resolve all dependencies in *spec* for *date*.
 
@@ -109,6 +112,9 @@ def run(
             A tuple of (:class:`DependencyResolution`, lockfile path or
             ``None`` if nothing was resolved).
         """
+        # A failed URL should be suppressed only within one resolution run.
+        # Product publication state may change before this client is reused.
+        self._transport.reset_failed_downloads()
         version = get_package_version()
         lockfile_dir = self._workspace.lockfile_dir(sink_id)
         manager = LockfileManager(lockfile_dir)
@@ -126,8 +132,11 @@ def run(
             date=date,
             version=version,
         )
-        if existing is not None:
-            resolution = self._resolution_from_lockfile(existing, spec)
+        cached_resolution = (
+            self._resolution_from_lockfile(existing, spec) if existing is not None else None
+        )
+        if cached_resolution is not None and not force_download:
+            resolution = cached_resolution
             if resolution.all_required_fulfilled:
                 logger.info(
                     "Lockfile already exists for %s on %s — skipping resolution: %s",
@@ -144,6 +153,17 @@ def run(
             )
 
         # --- Full resolution -------------------------------------------------
+        cached_static = {
+            result.spec: result
+            for result in (cached_resolution.resolved if cached_resolution else [])
+            if result.status != "missing"
+        }
+        dependencies_to_resolve = [
+            dep
+            for dep in spec.dependencies
+            if dep.refresh_on_force or dep.spec not in cached_static
+        ]
+
         resolve_one = partial(
             self._resolve_one,
             date=date,
@@ -151,16 +171,54 @@ def run(
             preferences=spec.preferences,
             centers=centers,
             download=download,
+            force_download=force_download,
         )
+        dep_by_spec = {dep.spec: dep for dep in dependencies_to_resolve}
+        bundled_specs = {
+            member for bundle in spec.bundles for member in bundle.members if member in dep_by_spec
+        }
+        independent = [dep for dep in dependencies_to_resolve if dep.spec not in bundled_specs]
+
         with ThreadPoolExecutor(max_workers=15) as executor:
-            resolved = list(executor.map(resolve_one, spec.dependencies))
+            futures = [executor.submit(resolve_one, dep) for dep in independent]
+            futures.extend(
+                executor.submit(
+                    self._resolve_bundle,
+                    bundle,
+                    dep_by_spec,
+                    date=date,
+                    sink_id=sink_id,
+                    preferences=spec.preferences,
+                    centers=bundle_centers if bundle_centers is not None else centers,
+                    download=download,
+                    force_download=force_download,
+                )
+                for bundle in spec.bundles
+                if any(member in dep_by_spec for member in bundle.members)
+            )
+            newly_resolved = []
+            for future in as_completed(futures):
+                result = future.result()
+                newly_resolved.extend(result if isinstance(result, list) else [result])
+
+        by_spec = {result.spec: result for result in newly_resolved}
+        by_spec.update(
+            {
+                dep.spec: cached_static[dep.spec]
+                for dep in spec.dependencies
+                if not dep.refresh_on_force and dep.spec in cached_static
+            }
+        )
+        resolved = [by_spec[dep.spec] for dep in spec.dependencies]
 
         resolution = DependencyResolution(spec_name=spec.name, resolved=resolved)
 
         lf_path: AnyPath | None = None
-        if resolution.fulfilled:
+        if resolution.all_required_fulfilled:
             writer = LockfileWriter(lockfile_dir, package=spec.package)
             lf_path = writer.write(resolution, date)
+        elif resolution.fulfilled:
+            logger.warning("Not writing aggregate lockfile: required dependencies are missing")
 
         logger.info(resolution.summary())
         return resolution, lf_path
@@ -176,6 +234,7 @@ def _resolve_one(
         preferences: list[SearchPreference],
         centers: list[str] | None,
         download: bool,
+        force_download: bool,
     ) -> ResolvedDependency:
         """Resolve a single dependency.
 
@@ -190,7 +249,36 @@ def _resolve_one(
         Returns:
             A :class:`ResolvedDependency` with the resolution result.
         """
-        logger.debug("Attempting to resolve dependency %s on %s", dep.spec, date.date())
+        candidates = self._search_candidates(
+            dep,
+            date=date,
+            preferences=preferences,
+            centers=centers,
+        )
+        if force_download and dep.refresh_on_force:
+            found = next((item for item in candidates if not item.is_local), None)
+        else:
+            found = candidates[0] if candidates else None
+
+        return self._materialize(
+            dep,
+            found,
+            date=date,
+            sink_id=sink_id,
+            download=download,
+            force_download=force_download,
+        )
+
+    def _search_candidates(
+        self,
+        dep: Dependency,
+        *,
+        date: datetime.datetime,
+        preferences: list[SearchPreference],
+        centers: list[str] | None,
+    ) -> list:
+        """Search for all candidates for one dependency."""
+        logger.debug("Searching for dependency %s on %s", dep.spec, date.date())
         try:
             q = self._query.for_product(dep.spec).on(date)
             if dep.constraints:
@@ -200,12 +288,22 @@ def _resolve_one(
                     q = q.prefer(**{pref.parameter: pref.sorting})
             if centers:
                 q = q.sources(*centers)
-            candidates = q.search()
-            found: FoundResource | None = candidates[0] if candidates else None
+            return q.search()
         except Exception as exc:
             logger.debug("No candidates for %s: %s", dep.spec, exc)
-            return ResolvedDependency(spec=dep.spec, required=dep.required, status="missing")
+            return []
 
+    def _materialize(
+        self,
+        dep: Dependency,
+        found,
+        *,
+        date: datetime.datetime,
+        sink_id: str,
+        download: bool,
+        force_download: bool,
+    ) -> ResolvedDependency:
+        """Turn a selected local or remote candidate into a resolution result."""
         if found is None:
             logger.warning("No search results for dependency %s", dep.spec)
             return ResolvedDependency(spec=dep.spec, required=dep.required, status="missing")
@@ -227,7 +325,12 @@ def _resolve_one(
                 remote_url=found.uri,
             )
 
-        path = self._downloader.run(found, date, sink_id=sink_id)
+        path = self._downloader.run(
+            found,
+            date,
+            sink_id=sink_id,
+            force=force_download and dep.refresh_on_force,
+        )
         if path is None:
             logger.warning("Download failed for dependency %s", dep.spec)
             return ResolvedDependency(spec=dep.spec, required=dep.required, status="missing")
@@ -241,6 +344,122 @@ def _resolve_one(
             remote_url=found.uri,
         )
 
+    def _resolve_bundle(
+        self,
+        bundle: DependencyBundle,
+        dep_by_spec: dict[str, Dependency],
+        *,
+        date: datetime.datetime,
+        sink_id: str,
+        preferences: list[SearchPreference],
+        centers: list[str] | None,
+        download: bool,
+        force_download: bool,
+    ) -> list[ResolvedDependency]:
+        """Resolve a coherent dependency family, falling back as a whole."""
+        members = [dep_by_spec[name] for name in bundle.members if name in dep_by_spec]
+        required = [dep for dep in members if dep.required]
+
+        def family_rank(key: tuple[str, ...]) -> tuple:
+            values = dict(zip(bundle.coherence, key, strict=True))
+            ranks = []
+            for preference in preferences:
+                value = values.get(preference.parameter, "")
+                try:
+                    ranks.append(preference.sorting.index(value))
+                except ValueError:
+                    ranks.append(len(preference.sorting))
+            return (*ranks, key)
+
+        if centers is not None:
+            center_stages: list[list[str] | None] = [centers]
+        else:
+            preferred_centers = next(
+                (pref.sorting for pref in preferences if pref.parameter == "AAA"), []
+            )
+            center_stages = (
+                [[preferred_centers[0]], preferred_centers[1:]]
+                if len(preferred_centers) > 1
+                else [None]
+            )
+
+        for stage_centers in center_stages:
+            logger.info("Searching bundle %s center stage %s", bundle.name, stage_centers or "all")
+            with ThreadPoolExecutor(max_workers=max(1, len(members))) as executor:
+                searched = list(
+                    executor.map(
+                        lambda dep: self._search_candidates(
+                            dep,
+                            date=date,
+                            preferences=preferences,
+                            centers=stage_centers,
+                        ),
+                        members,
+                    )
+                )
+
+            candidates_by_member: dict[str, dict[tuple[str, ...], list]] = {}
+            for dep, candidates in zip(members, searched, strict=True):
+                if force_download and dep.refresh_on_force:
+                    candidates = [candidate for candidate in candidates if not candidate.is_local]
+                grouped: dict[tuple[str, ...], list] = {}
+                for candidate in candidates:
+                    key = tuple(candidate.parameters.get(field, "") for field in bundle.coherence)
+                    grouped.setdefault(key, []).append(candidate)
+                candidates_by_member[dep.spec] = grouped
+
+            family_keys = set().union(*(set(groups) for groups in candidates_by_member.values()))
+            for key in sorted(family_keys, key=family_rank):
+                present = [dep.spec for dep in members if key in candidates_by_member[dep.spec]]
+                missing = [
+                    dep.spec for dep in required if key not in candidates_by_member[dep.spec]
+                ]
+                log = logger.warning if missing else logger.info
+                log(
+                    "Bundle preflight %s family %s: present=%s; missing_required=%s",
+                    bundle.name,
+                    key,
+                    present,
+                    missing,
+                )
+
+            viable = set(candidates_by_member[required[0].spec]) if required else set()
+            for dep in required[1:]:
+                viable &= set(candidates_by_member[dep.spec])
+
+            for key in sorted(viable, key=family_rank):
+                selected = {
+                    dep.spec: candidates_by_member[dep.spec].get(key, [None])[0] for dep in members
+                }
+                with ThreadPoolExecutor(max_workers=max(1, len(members))) as executor:
+                    results = list(
+                        executor.map(
+                            lambda dep: self._materialize(
+                                dep,
+                                selected[dep.spec],
+                                date=date,
+                                sink_id=sink_id,
+                                download=download,
+                                force_download=force_download,
+                            ),
+                            members,
+                        )
+                    )
+                if all(result.status != "missing" for result in results if result.required):
+                    logger.info("Selected %s bundle family %s", bundle.name, key)
+                    return results
+                logger.warning(
+                    "Rejected %s bundle family %s after a required product failed",
+                    bundle.name,
+                    key,
+                )
+
+        logger.warning("No complete downloadable family found for bundle %s", bundle.name)
+        return [
+            ResolvedDependency(spec=dep.spec, required=dep.required, status="missing")
+            for dep in members
+        ]
+
     @staticmethod
     def _lockfile_entry_is_valid(lp) -> bool:
         """Check a lockfile entry's sink file: existence, then sidecar hash.
diff --git a/packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py b/packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py
index 7548eac..4405c65 100644
--- a/packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py
+++ b/packages/gnss-product-management/src/gnss_product_management/factories/remote_transport.py
@@ -2,7 +2,10 @@
 
 import datetime
 import logging
+import random
 import re
+import threading
+import time
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
@@ -64,9 +67,16 @@ def __init__(
         """
         self._connection_pool_factory = ConnectionPoolFactory(max_connections=max_connections)
         self._product_registry = product_registry
+        self._failed_downloads: set[tuple[str, str]] = set()
+        self._failed_downloads_lock = threading.Lock()
 
     # -- Public API ------------------------------------------------
 
+    def reset_failed_downloads(self) -> None:
+        """Forget failures recorded during the previous resolution run."""
+        with self._failed_downloads_lock:
+            self._failed_downloads.clear()
+
     def search(self, targets: list[SearchTarget]) -> list[SearchTarget]:
         """Search every target's server/directory for matching files.
 
@@ -85,8 +95,12 @@ def search(self, targets: list[SearchTarget]) -> list[SearchTarget]:
         groups, rejected = self._group_targets(targets)
 
         # Ensure connection pools exist for every hostname we'll contact.
-        for hostname, _ in groups:
-            self._connection_pool_factory.add_connection(hostname)
+        for (hostname, _), group_targets in groups.items():
+            server = group_targets[0][0].server
+            self._connection_pool_factory.add_connection(
+                hostname,
+                listing_url=server.listing_url,
+            )
 
         # List each unique directory in parallel.
         dir_keys = list(groups.keys())
@@ -299,6 +313,8 @@ def download_one(
         local_resource_id: str,
         local_factory: WorkSpace,
         date: datetime.datetime,
+        *,
+        force: bool = False,
     ) -> AnyPath | None:
         """Synchronously download matched files for one search target.
 
@@ -329,11 +345,13 @@ def download_one(
         destination_dir.mkdir(parents=True, exist_ok=True)
         destination_path = destination_dir / query.product.filename.value  # type: ignore[union-attr]
 
-        # Prefer an already-decompressed version on disk
+        # Prefer an already-decompressed version on disk unless the caller
+        # explicitly requested a fresh copy from the remote source.
         if destination_path.suffix == ".gz":
             decompressed_path = destination_path.with_suffix("")
             if (
-                decompressed_path.exists()
+                not force
+                and decompressed_path.exists()
                 and decompressed_path.stat().st_size > 0
                 and self._validate_cached_or_evict(decompressed_path)
             ):
@@ -348,7 +366,8 @@ def download_one(
         # describes the file as served, so it only applies to the
         # non-decompressed destination path.
         if (
-            destination_path.exists()
+            not force
+            and destination_path.exists()
             and destination_path.stat().st_size > 0
             and self._validate_cached_or_evict(destination_path, query.checksum)
         ):
@@ -358,15 +377,27 @@ def download_one(
         remote_file_path = str(
             Path(query.directory.value) / query.product.filename.value  # type: ignore[union-attr]
         )
+        failure_key = (hostname, remote_file_path)
+        with self._failed_downloads_lock:
+            if failure_key in self._failed_downloads:
+                logger.info("Skipping product that already failed this run: %s", remote_file_path)
+                return None
+
+        if force:
+            logger.info("Refreshing mutable cached product: %s", destination_path.name)
 
         # Skip download if the remote file is zero bytes (stale/incomplete upload).
         remote_size = self._connection_pool_factory.get_file_size(hostname, remote_file_path)
         if remote_size is not None and remote_size == 0:
             logger.warning("Skipping zero-byte remote file: %s/%s", hostname, remote_file_path)
+            with self._failed_downloads_lock:
+                self._failed_downloads.add(failure_key)
             return None
 
         result: Path | None = None
         for attempt in range(2):
+            if attempt and fsspec.utils.get_protocol(hostname) != "file":
+                time.sleep(random.uniform(1.0, 3.0))
             try:
                 result = self._connection_pool_factory.download_file(
                     hostname=hostname,
@@ -381,8 +412,12 @@ def download_one(
                     query.product.filename.value,
                     e,
                 )
+                with self._failed_downloads_lock:
+                    self._failed_downloads.add(failure_key)
                 return None
             if result is None:
+                with self._failed_downloads_lock:
+                    self._failed_downloads.add(failure_key)
                 return None
             # Truncated/corrupt transfers must never be left on disk where
             # a later run would treat them as satisfied dependencies.
@@ -408,8 +443,13 @@ def download_one(
             result = None
 
         if result is None:
+            with self._failed_downloads_lock:
+                self._failed_downloads.add(failure_key)
             return None
 
+        with self._failed_downloads_lock:
+            self._failed_downloads.discard(failure_key)
+
         # Decompress gzip files after download
         if result.suffix == ".gz":
             decompressed = decompress_gzip(result)
diff --git a/packages/gnss-product-management/src/gnss_product_management/specifications/dependencies/dependencies.py b/packages/gnss-product-management/src/gnss_product_management/specifications/dependencies/dependencies.py
index 4d0c3a9..849f921 100644
--- a/packages/gnss-product-management/src/gnss_product_management/specifications/dependencies/dependencies.py
+++ b/packages/gnss-product-management/src/gnss_product_management/specifications/dependencies/dependencies.py
@@ -28,6 +28,24 @@ class Dependency(BaseModel):
     required: bool = True
     description: str = ""
     constraints: dict[str, str] = Field(default_factory=dict)
+    refresh_on_force: bool = Field(
+        default=True,
+        description=(
+            "Re-query and re-download this dependency when force_download is enabled. "
+            "Set false for immutable files bundled with an application."
+        ),
+    )
+
+
+class DependencyBundle(BaseModel):
+    """Dependencies that must come from one coherent product family."""
+
+    name: str
+    members: list[str]
+    coherence: list[str] = Field(
+        default_factory=lambda: ["AAA", "TTT", "PPP"],
+        description="Product parameters that identify a coherent family.",
+    )
 
 
 class DependencySpec(BaseModel):
@@ -37,6 +55,7 @@ class DependencySpec(BaseModel):
     description: str = ""
     preferences: list[SearchPreference] = Field(default_factory=list)
     dependencies: list[Dependency] = Field(default_factory=list)
+    bundles: list[DependencyBundle] = Field(default_factory=list)
     package: str
     task: str
 
@@ -83,6 +102,7 @@ class DependencyResolution:
 
     spec_name: str
     resolved: list[ResolvedDependency] = field(default_factory=list)
+    diagnostics: list[str] = field(default_factory=list)
 
     @property
     def fulfilled(self) -> list[ResolvedDependency]:
diff --git a/packages/gnss-product-management/src/gnss_product_management/specifications/remote/resource.py b/packages/gnss-product-management/src/gnss_product_management/specifications/remote/resource.py
index 985127c..d670ab8 100644
--- a/packages/gnss-product-management/src/gnss_product_management/specifications/remote/resource.py
+++ b/packages/gnss-product-management/src/gnss_product_management/specifications/remote/resource.py
@@ -21,6 +21,9 @@ class Server(BaseModel):
         hostname: Server hostname or URL.
         protocol: Protocol (``'ftp'``, ``'http'``, ``'https'``, etc.).
         auth_required: Whether authentication is needed.
+        listing_url: Optional URL template for servers whose directory listing
+            is exposed separately from their download root.  ``{directory}``
+            is replaced with the URL-quoted directory being searched.
         description: Human-readable server description.
     """
 
@@ -28,6 +31,7 @@ class Server(BaseModel):
     hostname: str
     protocol: str | None = None
     auth_required: bool | None = False
+    listing_url: str | None = None
     description: str | None = None
 
 
diff --git a/packages/gnss-product-management/test/conftest.py b/packages/gnss-product-management/test/conftest.py
index ff503b0..6118e96 100644
--- a/packages/gnss-product-management/test/conftest.py
+++ b/packages/gnss-product-management/test/conftest.py
@@ -88,6 +88,11 @@ def cddis_env() -> ProductRegistry:
     return _build_env("cddis_config.yaml")
 
 
+@pytest.fixture(scope="session")
+def bkg_env() -> ProductRegistry:
+    return _build_env("bkg_config.yaml")
+
+
 @pytest.fixture(scope="session")
 def esa_env() -> ProductRegistry:
     return _build_env("esa_config.yaml")
@@ -143,6 +148,11 @@ def cddis_qf(cddis_env, workspace) -> SearchPlanner:
     return SearchPlanner(product_registry=cddis_env, workspace=workspace)
 
 
+@pytest.fixture(scope="session")
+def bkg_qf(bkg_env, workspace) -> SearchPlanner:
+    return SearchPlanner(product_registry=bkg_env, workspace=workspace)
+
+
 @pytest.fixture(scope="session")
 def igs_qf(igs_env, workspace) -> SearchPlanner:
     return SearchPlanner(product_registry=igs_env, workspace=workspace)
diff --git a/packages/gnss-product-management/test/test_code_https.py b/packages/gnss-product-management/test/test_code_https.py
new file mode 100644
index 0000000..aa7b47b
--- /dev/null
+++ b/packages/gnss-product-management/test/test_code_https.py
@@ -0,0 +1,90 @@
+"""Regression tests for CODE's S3-backed HTTPS product archive."""
+
+from unittest.mock import MagicMock
+
+import pytest
+from gnss_product_management.client.product_query import _remote_uri
+from gnss_product_management.factories.connection_pool import ConnectionPoolFactory
+
+
+def test_configured_https_listing_extracts_download_filenames(monkeypatch) -> None:
+    html = b"""
+    <a href="https://archive.example/download/CODE/COD0OPSRAP_20262390000_01D_05M_ORB.SP3">
+      COD0OPSRAP_20262390000_01D_05M_ORB.SP3
+    </a>
+    """
+    factory = ConnectionPoolFactory(max_connections=1)
+    hostname = "https://archive.example/download"
+    factory.add_connection(
+        hostname,
+        listing_url="https://listing.example/browse?path={directory}",
+    )
+    pool = factory._pools[hostname]
+    connection = MagicMock()
+    connection.cat_file.return_value = html
+    monkeypatch.setattr(pool, "_connect", lambda: connection)
+
+    assert factory.list_directory(hostname, "CODE/") == ["COD0OPSRAP_20262390000_01D_05M_ORB.SP3"]
+    connection.cat_file.assert_called_once_with("https://listing.example/browse?path=CODE")
+
+
+def test_code_spec_configures_aiub_listing_url(cod_qf, test_date) -> None:
+    targets = cod_qf.get(date=test_date, product={"name": "ORBIT"})
+    remote = [target for target in targets if target.server.protocol == "https"]
+
+    assert remote
+    assert all(
+        target.server.listing_url
+        == ("https://code.aiub.unibe.ch/s3_script/aiub_s3_bucket_listing.php?path={directory}")
+        for target in remote
+    )
+
+
+@pytest.mark.parametrize("product", ["ORBIT", "CLOCK", "ERP", "BIA"])
+@pytest.mark.parametrize(
+    ("timeliness", "expected_directory"),
+    [("FIN", "CODE/2025/"), ("RAP", "CODE/")],
+)
+def test_code_precise_product_directory_matches_family(
+    cod_qf, test_date, product, timeliness, expected_directory
+) -> None:
+    targets = cod_qf.get(
+        date=test_date,
+        product={"name": product},
+        parameters={"TTT": timeliness},
+    )
+    remote = [target for target in targets if target.server.protocol == "https"]
+
+    assert remote
+    assert {target.directory.value or target.directory.pattern for target in remote} == {
+        expected_directory
+    }
+
+
+@pytest.mark.parametrize(
+    ("timeliness", "expected_directory"),
+    [("FIN", "CODE/2025/"), ("RAP", "CODE/"), ("PRD", "CODE/")],
+)
+def test_code_ionosphere_directory_matches_family(
+    cod_qf, test_date, timeliness, expected_directory
+) -> None:
+    targets = cod_qf.get(
+        date=test_date,
+        product={"name": "IONEX"},
+        parameters={"TTT": timeliness},
+    )
+    remote = [target for target in targets if target.server.protocol == "https"]
+
+    assert remote
+    assert {target.directory.value or target.directory.pattern for target in remote} == {
+        expected_directory
+    }
+
+
+def test_remote_uri_keeps_existing_https_scheme() -> None:
+    assert _remote_uri(
+        "https",
+        "https://www.aiub.unibe.ch/download",
+        "CODE/",
+        "COD0OPSRAP_20262390000_01D_05M_ORB.SP3",
+    ) == ("https://www.aiub.unibe.ch/download/CODE/COD0OPSRAP_20262390000_01D_05M_ORB.SP3")
diff --git a/packages/gnss-product-management/test/test_dependency_resolution.py b/packages/gnss-product-management/test/test_dependency_resolution.py
index f87a4c6..589851f 100644
--- a/packages/gnss-product-management/test/test_dependency_resolution.py
+++ b/packages/gnss-product-management/test/test_dependency_resolution.py
@@ -9,14 +9,20 @@
 
 from __future__ import annotations
 
+import datetime
 from pathlib import Path
+from unittest.mock import MagicMock
 
 import pytest
 from gnss_product_management.environments import WorkSpace
+from gnss_product_management.factories.models import FoundResource
 from gnss_product_management.factories.pipelines.resolve import ResolvePipeline
 from gnss_product_management.specifications.dependencies.dependencies import (
+    Dependency,
+    DependencyBundle,
     DependencySpec,
     ResolvedDependency,
+    SearchPreference,
 )
 
 # ── Paths ──────────────────────────────────────────────────────────
@@ -100,6 +106,164 @@ def test_pipeline_builds(self, pipeline) -> None:
         assert pipeline is not None
 
 
+class TestSelectiveForceDownload:
+    @staticmethod
+    def _pipeline_with_candidates(candidates):
+        pipeline = ResolvePipeline.__new__(ResolvePipeline)
+        query = MagicMock()
+        query.for_product.return_value.on.return_value = query
+        query.search.return_value = candidates
+        pipeline._query = query
+        pipeline._downloader = MagicMock()
+        pipeline._downloader.run.return_value = Path("/tmp/downloaded.product")
+        return pipeline
+
+    def test_force_keeps_non_refreshable_bundled_product(self, tmp_path) -> None:
+        table = tmp_path / "igs20.atx"
+        table.write_text("static")
+        local = FoundResource(product="ATTATX", source="local", uri=str(table))
+        remote = FoundResource(
+            product="ATTATX", source="remote", uri="https://example.test/igs20.atx"
+        )
+        pipeline = self._pipeline_with_candidates([local, remote])
+
+        result = pipeline._resolve_one(
+            Dependency(spec="ATTATX", refresh_on_force=False),
+            date=datetime.datetime(2026, 8, 19, tzinfo=datetime.UTC),
+            sink_id="local_config",
+            preferences=[],
+            centers=None,
+            download=True,
+            force_download=True,
+        )
+
+        assert result.status == "local"
+        pipeline._downloader.run.assert_not_called()
+
+    def test_force_refreshes_mutable_remote_product(self, tmp_path) -> None:
+        clock = tmp_path / "clock.CLK"
+        clock.write_text("cached")
+        local = FoundResource(product="CLOCK", source="local", uri=str(clock))
+        remote = FoundResource(
+            product="CLOCK", source="remote", uri="https://example.test/clock.CLK"
+        )
+        pipeline = self._pipeline_with_candidates([local, remote])
+
+        result = pipeline._resolve_one(
+            Dependency(spec="CLOCK"),
+            date=datetime.datetime(2026, 8, 19, tzinfo=datetime.UTC),
+            sink_id="local_config",
+            preferences=[],
+            centers=None,
+            download=True,
+            force_download=True,
+        )
+
+        assert result.status == "downloaded"
+        selected = pipeline._downloader.run.call_args.args[0]
+        assert selected.uri == remote.uri
+        assert pipeline._downloader.run.call_args.kwargs["force"] is True
+
+
+class TestCoherentBundles:
+    date = datetime.datetime(2026, 8, 19, tzinfo=datetime.UTC)
+    preferences = [
+        SearchPreference(parameter="AAA", sorting=["WUM"]),
+        SearchPreference(parameter="TTT", sorting=["RTS", "RAP"]),
+    ]
+    bundle = DependencyBundle(
+        name="precise-products",
+        members=["ORBIT", "CLOCK", "ATTOBX"],
+        coherence=["AAA", "TTT", "PPP"],
+    )
+
+    @staticmethod
+    def _candidate(product: str, timeliness: str) -> FoundResource:
+        return FoundResource(
+            product=product,
+            source="remote",
+            uri=f"https://example.test/{timeliness}/{product}",
+            parameters={"AAA": "WUM", "TTT": timeliness, "PPP": "MGX"},
+        )
+
+    @staticmethod
+    def _pipeline(search_results: dict[str, list[FoundResource]]) -> ResolvePipeline:
+        pipeline = ResolvePipeline.__new__(ResolvePipeline)
+        pipeline._search_candidates = lambda dep, **kwargs: search_results[dep.spec]
+        pipeline._downloader = MagicMock()
+        pipeline._downloader.run.side_effect = lambda found, *args, **kwargs: Path(
+            f"/tmp/{found.quality}-{found.product}"
+        )
+        return pipeline
+
+    def test_incomplete_families_are_not_mixed(self) -> None:
+        pipeline = self._pipeline(
+            {
+                "ORBIT": [self._candidate("ORBIT", "RTS")],
+                "CLOCK": [self._candidate("CLOCK", "RAP")],
+                "ATTOBX": [],
+            }
+        )
+        dependencies = {
+            "ORBIT": Dependency(spec="ORBIT"),
+            "CLOCK": Dependency(spec="CLOCK"),
+            "ATTOBX": Dependency(spec="ATTOBX", required=False),
+        }
+
+        results = pipeline._resolve_bundle(
+            self.bundle,
+            dependencies,
+            date=self.date,
+            sink_id="local_config",
+            preferences=self.preferences,
+            centers=None,
+            download=True,
+            force_download=True,
+        )
+
+        assert all(result.status == "missing" for result in results)
+        pipeline._downloader.run.assert_not_called()
+
+    def test_required_failure_falls_back_as_a_whole(self) -> None:
+        search_results = {
+            product: [
+                self._candidate(product, "RTS"),
+                self._candidate(product, "RAP"),
+            ]
+            for product in ("ORBIT", "CLOCK", "ATTOBX")
+        }
+        pipeline = self._pipeline(search_results)
+
+        def download(found, *args, **kwargs):
+            if found.product == "CLOCK" and found.quality == "RTS":
+                return None
+            return Path(f"/tmp/{found.quality}-{found.product}")
+
+        pipeline._downloader.run.side_effect = download
+        dependencies = {
+            "ORBIT": Dependency(spec="ORBIT"),
+            "CLOCK": Dependency(spec="CLOCK"),
+            "ATTOBX": Dependency(spec="ATTOBX", required=False),
+        }
+
+        results = pipeline._resolve_bundle(
+            self.bundle,
+            dependencies,
+            date=self.date,
+            sink_id="local_config",
+            preferences=self.preferences,
+            centers=None,
+            download=True,
+            force_download=True,
+        )
+
+        assert all(result.status == "downloaded" for result in results)
+        assert all("/RAP/" in result.remote_url for result in results)
+        downloaded_urls = [call.args[0].uri for call in pipeline._downloader.run.call_args_list]
+        assert "https://example.test/RTS/ORBIT" in downloaded_urls
+        assert "https://example.test/RAP/ORBIT" in downloaded_urls
+
+
 # ===================================================================
 # Integration: Resolve with remote search (network required)
 # ===================================================================
diff --git a/packages/gnss-product-management/test/test_download_integrity.py b/packages/gnss-product-management/test/test_download_integrity.py
index 63a0cde..ec9b6d4 100644
--- a/packages/gnss-product-management/test/test_download_integrity.py
+++ b/packages/gnss-product-management/test/test_download_integrity.py
@@ -96,13 +96,20 @@ def env(tmp_path: Path):
     }
 
 
-def _download(env, filename: str = "TEST.SP3", checksum: str | None = None) -> Path | None:
+def _download(
+    env,
+    filename: str = "TEST.SP3",
+    checksum: str | None = None,
+    *,
+    force: bool = False,
+) -> Path | None:
     query = _make_query(env["remote_root"], filename, checksum)
     return env["wormhole"].download_one(
         query=query,
         local_resource_id="local_config",
         local_factory=env["workspace"],
         date=TEST_DATE,
+        force=force,
     )
 
 
@@ -157,6 +164,44 @@ def _flaky(self, hostname, remote_path, target_dir):
         assert result.read_bytes() == REMOTE_CONTENT
         assert len(calls) == 2
 
+    def test_failed_url_is_not_retried_again_in_same_run(self, env, monkeypatch) -> None:
+        calls: list[int] = []
+
+        def _always_truncated(self, hostname, remote_path, target_dir):
+            calls.append(1)
+            local = Path(target_dir) / Path(remote_path).name
+            local.write_bytes(REMOTE_CONTENT[:10])
+            return local
+
+        monkeypatch.setattr(ConnectionPoolFactory, "download_file", _always_truncated)
+
+        assert _download(env) is None
+        assert len(calls) == 2
+        assert _download(env) is None
+        assert len(calls) == 2
+
+    def test_failed_url_can_be_retried_in_next_run(self, env, monkeypatch) -> None:
+        calls: list[int] = []
+
+        def _download_after_publication(self, hostname, remote_path, target_dir):
+            calls.append(1)
+            local = Path(target_dir) / Path(remote_path).name
+            content = REMOTE_CONTENT[:10] if len(calls) <= 2 else REMOTE_CONTENT
+            local.write_bytes(content)
+            return local
+
+        monkeypatch.setattr(ConnectionPoolFactory, "download_file", _download_after_publication)
+
+        assert _download(env) is None
+        assert len(calls) == 2
+
+        env["wormhole"].reset_failed_downloads()
+
+        result = _download(env)
+        assert result is not None
+        assert result.read_bytes() == REMOTE_CONTENT
+        assert len(calls) == 3
+
 
 # ── Cached files ──────────────────────────────────────────────────
 
@@ -191,6 +236,16 @@ def test_cache_without_sidecar_is_trusted(self, env, monkeypatch) -> None:
         assert _download(env) == cached
         assert calls == []
 
+    def test_force_redownload_replaces_valid_cache(self, env) -> None:
+        cached = env["sink_dir"] / "TEST.SP3"
+        cached.parent.mkdir(parents=True)
+        cached.write_bytes(b"valid but stale product")
+
+        result = _download(env, force=True)
+
+        assert result == cached
+        assert result.read_bytes() == REMOTE_CONTENT
+
     def test_corrupt_cache_is_evicted_and_redownloaded(self, env) -> None:
         """A cached file whose hash no longer matches its sidecar must be
         evicted (with its stale sidecar) and fetched fresh."""
diff --git a/packages/gnss-product-management/test/test_filepath_generation.py b/packages/gnss-product-management/test/test_filepath_generation.py
index 7249b71..0c6b45e 100644
--- a/packages/gnss-product-management/test/test_filepath_generation.py
+++ b/packages/gnss-product-management/test/test_filepath_generation.py
@@ -304,12 +304,16 @@ def test_ionex_filename_has_gim_inx(self, cod_qf, test_date) -> None:
             fn = q.product.filename.pattern
             assert "GIM" in fn and "INX" in fn
 
-    def test_ionex_directory_contains_year(self, cod_qf, test_date) -> None:
-        queries = cod_qf.get(date=test_date, product={"name": "IONEX"})
+    def test_final_ionex_directory_uses_code_year(self, cod_qf, test_date) -> None:
+        queries = cod_qf.get(
+            date=test_date,
+            product={"name": "IONEX"},
+            parameters={"TTT": "FIN"},
+        )
         remote = [q for q in queries if q.server.protocol != "file"]
         for q in remote:
-            d = q.directory.pattern
-            assert "2025" in d
+            d = q.directory.value or q.directory.pattern
+            assert d == "CODE/2025/"
 
     def test_brdc_filename_contains_brdc(self, wuhan_qf, test_date) -> None:
         queries = wuhan_qf.get(date=test_date, product={"name": "RNX3_BRDC"})
@@ -473,12 +477,16 @@ def test_invalid_product_raises(self, wuhan_qf, test_date) -> None:
             wuhan_qf.get(date=test_date, product={"name": "NONEXISTENT"})
 
     def test_cod_orbit_directory_is_code_yyyy(self, cod_qf, test_date) -> None:
-        queries = cod_qf.get(date=test_date, product={"name": "ORBIT"})
+        queries = cod_qf.get(
+            date=test_date,
+            product={"name": "ORBIT"},
+            parameters={"TTT": "FIN"},
+        )
         remote = [q for q in queries if q.server.protocol != "file"]
         assert len(remote) > 0
         for q in remote:
-            d = q.directory.pattern
-            assert d.startswith("CODE/")
+            d = q.directory.value or q.directory.pattern
+            assert d == "CODE/2025/"
 
     def test_cddis_orbit_directory_is_gpsweek(self, cddis_qf, test_date) -> None:
         gpsweek = str((test_date.date() - datetime.date(1980, 1, 6)).days // 7)
diff --git a/packages/gnss-product-management/test/test_ionosphere_resources.py b/packages/gnss-product-management/test/test_ionosphere_resources.py
index 9d38b1e..4eb3af1 100644
--- a/packages/gnss-product-management/test/test_ionosphere_resources.py
+++ b/packages/gnss-product-management/test/test_ionosphere_resources.py
@@ -44,7 +44,7 @@ def test_ionex_queries_returned(self, cod_qf, test_date) -> None:
     def test_ionex_server_protocol(self, cod_qf, test_date) -> None:
         queries = _get_remote_queries(cod_qf, test_date, "IONEX")
         for q in queries:
-            assert q.server.protocol.lower() == "ftp"
+            assert q.server.protocol.lower() == "https"
 
     def test_ionex_directory_not_empty(self, cod_qf, test_date) -> None:
         queries = _get_remote_queries(cod_qf, test_date, "IONEX")
diff --git a/packages/gnss-product-management/test/test_navigation_resources.py b/packages/gnss-product-management/test/test_navigation_resources.py
index 2a8cba6..1766c37 100644
--- a/packages/gnss-product-management/test/test_navigation_resources.py
+++ b/packages/gnss-product-management/test/test_navigation_resources.py
@@ -2,7 +2,7 @@
 Tests: Broadcast navigation products via SearchPlanner.
 
 Products: RNX3_BRDC
-Centers : Wuhan (FTP), CDDIS (FTPS)
+Centers : Wuhan (FTP), CDDIS (FTPS), BKG (HTTPS)
 """
 
 from __future__ import annotations
@@ -79,6 +79,22 @@ def test_brdc_filename_contains_brdc(self, cddis_qf, test_date) -> None:
         assert any("BRDC" in p for p in patterns)
 
 
+class TestBKGNavigationExpansion:
+    def test_wrd_current_day_candidate_is_generated(self, bkg_qf, test_date) -> None:
+        queries = _get_remote_queries(
+            bkg_qf,
+            test_date,
+            "RNX3_BRDC",
+            parameters={"CCC": "WRD", "D": "M"},
+        )
+
+        assert len(queries) == 1
+        query = queries[0]
+        assert query.server.protocol.lower() == "https"
+        assert "WRD_R_20250150000_01D_MN.rnx" in query.product.filename.pattern
+        assert "/IGS/BRDC/" in query.directory.pattern
+
+
 # ---------------------------------------------------------------------------
 # Integration: Wuhan navigation probe
 # ---------------------------------------------------------------------------
diff --git a/packages/gnss-product-management/test/test_orbit_resources.py b/packages/gnss-product-management/test/test_orbit_resources.py
index be0ffb9..52828a5 100644
--- a/packages/gnss-product-management/test/test_orbit_resources.py
+++ b/packages/gnss-product-management/test/test_orbit_resources.py
@@ -96,10 +96,10 @@ def test_orbit_queries_returned(self, cod_qf, test_date) -> None:
         queries = _get_remote_queries(cod_qf, test_date, "ORBIT")
         assert len(queries) > 0
 
-    def test_orbit_server_protocol_is_ftp(self, cod_qf, test_date) -> None:
+    def test_orbit_server_protocol_is_https(self, cod_qf, test_date) -> None:
         queries = _get_remote_queries(cod_qf, test_date, "ORBIT")
         for q in queries:
-            assert q.server.protocol.lower() == "ftp"
+            assert q.server.protocol.lower() == "https"
 
     def test_orbit_directory_contains_code(self, cod_qf, test_date) -> None:
         queries = _get_remote_queries(cod_qf, test_date, "ORBIT")
diff --git a/packages/gnss-product-management/test/test_resource_finding.py b/packages/gnss-product-management/test/test_resource_finding.py
index 408c7de..364b139 100644
--- a/packages/gnss-product-management/test/test_resource_finding.py
+++ b/packages/gnss-product-management/test/test_resource_finding.py
@@ -94,12 +94,12 @@ def test_orbit_filenames_match_date(self, wuhan_qf, fetcher, test_date) -> None:
 
 
 # ---------------------------------------------------------------------------
-# CODE (FTP) — ftp.aiub.unibe.ch
+# CODE (HTTPS) — www.aiub.unibe.ch/download
 # ---------------------------------------------------------------------------
 
 
 class TestCODResourceFinding:
-    """Search for products on CODE FTP."""
+    """Search for products through the CODE HTTPS archive."""
 
     def test_orbit_found(self, cod_qf, fetcher, test_date) -> None:
         results = _search_remote(cod_qf, fetcher, test_date, "ORBIT")
@@ -125,14 +125,14 @@ def test_ionex_found(self, cod_qf, fetcher, test_date) -> None:
             for r in found
         )
 
-    def test_orbit_directory_is_code_year(self, cod_qf, fetcher, test_date) -> None:
-        """CODE orbit files should be under CODE/{YYYY}/."""
-        results = _search_remote(cod_qf, fetcher, test_date, "ORBIT")
+    def test_final_orbit_directory_is_code_year(self, cod_qf, fetcher, test_date) -> None:
+        """AIUB archives final products in year-specific directories."""
+        results = _search_remote(cod_qf, fetcher, test_date, "ORBIT", {"TTT": "FIN"})
         found = _assert_found(results, "ORBIT")
         for r in found:
             d = r.directory
-            d_str = d.pattern if isinstance(d, PathTemplate) else str(d)
-            assert "CODE/2025" in d_str
+            d_str = (d.value or d.pattern) if isinstance(d, PathTemplate) else str(d)
+            assert d_str == "CODE/2025/"
 
     def test_orbit_filenames_contain_cod(self, cod_qf, fetcher, test_date) -> None:
         results = _search_remote(cod_qf, fetcher, test_date, "ORBIT")
diff --git a/packages/gpm-specs/src/gpm_specs/configs/centers/bkg_config.yaml b/packages/gpm-specs/src/gpm_specs/configs/centers/bkg_config.yaml
index fbb76fb..4924b68 100644
--- a/packages/gpm-specs/src/gpm_specs/configs/centers/bkg_config.yaml
+++ b/packages/gpm-specs/src/gpm_specs/configs/centers/bkg_config.yaml
@@ -114,6 +114,19 @@ products:
     directory: {pattern: "root_ftp/IGS/products/{GPSWEEK}/"}
 
   # ── Broadcast navigation ─────────────────────────────────────────
+  # BKG's Broadcast Working Group publishes this rolling mixed-GNSS
+  # file during the current UTC day, before BRDC00IGS is available.
+  # ex: BRDC00WRD_R_20262470000_01D_MN.rnx.gz
+  - id: bkg_nav_wrd
+    product_name: RNX3_BRDC
+    server_id: bkg_https
+    available: true
+    description: Rolling current-day mixed-GNSS broadcast navigation (RINEX 3)
+    parameters:
+      - {name: CCC, value: WRD}
+      - {name: D, value: M}
+    directory: {pattern: "root_ftp/IGS/BRDC/{YYYY}/{DDD}/"}
+
   # ex: BRDC00IGS_R_20250150000_01D_MN.rnx.gz
   - id: bkg_nav
     product_name: RNX3_BRDC
diff --git a/packages/gpm-specs/src/gpm_specs/configs/centers/cod_config.yaml b/packages/gpm-specs/src/gpm_specs/configs/centers/cod_config.yaml
index efef862..b0238d2 100644
--- a/packages/gpm-specs/src/gpm_specs/configs/centers/cod_config.yaml
+++ b/packages/gpm-specs/src/gpm_specs/configs/centers/cod_config.yaml
@@ -1,93 +1,153 @@
 id: COD
 name: Center for Orbit Determination in Europe (AIUB)
-website: http://www.aiub.unibe.ch/
+website: https://www.aiub.unibe.ch/
 
 servers:
-  - id: code_ftp
-    name: Primary FTP
-    hostname: "ftp://ftp.aiub.unibe.ch"
-    protocol: ftp
+  - id: code_https
+    name: AIUB CODE Products HTTPS
+    hostname: "https://www.aiub.unibe.ch/download"
+    protocol: https
     auth_required: false
-    description: CODE FTP server at University of Bern
+    listing_url: "https://code.aiub.unibe.ch/s3_script/aiub_s3_bucket_listing.php?path={directory}"
+    description: >
+      CODE product downloads at the University of Bern. Directory listings
+      are provided by AIUB's S3-backed product browser; FTP is retired.
 
 products:
   # ── Orbits ──────────────────────────────────────────────────────
-  - id: code_orbit
+  - id: code_orbit_final
     product_name: ORBIT
-    server_id: code_ftp
+    server_id: code_https
     available: true
-    description: CODE precise orbits
+    description: Archived CODE final precise orbits
     parameters:
       - {name: AAA, value: COD}
       - {name: TTT, value: FIN}
-      - {name: TTT, value: RAP}
       - {name: PPP, value: OPS}
       - {name: PPP, value: MGX}
       - {name: SMP, value: 05M}
       - {name: SMP, value: 15M}
     directory: {pattern: "CODE/{YYYY}/"}
 
+  - id: code_orbit_rapid
+    product_name: ORBIT
+    server_id: code_https
+    available: true
+    description: Current CODE rapid precise orbits
+    parameters:
+      - {name: AAA, value: COD}
+      - {name: TTT, value: RAP}
+      - {name: PPP, value: OPS}
+      - {name: PPP, value: MGX}
+      - {name: SMP, value: 05M}
+      - {name: SMP, value: 15M}
+    directory: {pattern: "CODE/"}
+
   # ── Clocks ──────────────────────────────────────────────────────
-  - id: code_clock
+  - id: code_clock_final
     product_name: CLOCK
-    server_id: code_ftp
+    server_id: code_https
     available: true
-    description: CODE precise clocks
+    description: Archived CODE final precise clocks
     parameters:
       - {name: AAA, value: COD}
       - {name: TTT, value: FIN}
-      - {name: TTT, value: RAP}
       - {name: PPP, value: OPS}
       - {name: SMP, value: 30S}
       - {name: SMP, value: 05M}
     directory: {pattern: "CODE/{YYYY}/"}
 
+  - id: code_clock_rapid
+    product_name: CLOCK
+    server_id: code_https
+    available: true
+    description: Current CODE rapid precise clocks
+    parameters:
+      - {name: AAA, value: COD}
+      - {name: TTT, value: RAP}
+      - {name: PPP, value: OPS}
+      - {name: SMP, value: 30S}
+      - {name: SMP, value: 05M}
+    directory: {pattern: "CODE/"}
+
   # ── ERP ─────────────────────────────────────────────────────────
-  - id: code_erp
+  - id: code_erp_final
     product_name: ERP
-    server_id: code_ftp
+    server_id: code_https
     available: true
-    description: Earth rotation parameters
+    description: Archived CODE final Earth rotation parameters
     parameters:
       - {name: AAA, value: COD}
       - {name: TTT, value: FIN}
-      - {name: TTT, value: RAP}
       - {name: PPP, value: OPS}
     directory: {pattern: "CODE/{YYYY}/"}
 
+  - id: code_erp_rapid
+    product_name: ERP
+    server_id: code_https
+    available: true
+    description: Current CODE rapid Earth rotation parameters
+    parameters:
+      - {name: AAA, value: COD}
+      - {name: TTT, value: RAP}
+      - {name: PPP, value: OPS}
+    directory: {pattern: "CODE/"}
+
   # ── Biases ──────────────────────────────────────────────────────
-  - id: code_bias
+  - id: code_bias_final
     product_name: BIA
-    server_id: code_ftp
+    server_id: code_https
     available: true
-    description: Differential code biases (DCB/OSB)
+    description: Archived CODE final differential code biases (DCB/OSB)
     parameters:
       - {name: AAA, value: COD}
       - {name: TTT, value: FIN}
-      - {name: TTT, value: RAP}
       - {name: PPP, value: OPS}
     directory: {pattern: "CODE/{YYYY}/"}
 
+  - id: code_bias_rapid
+    product_name: BIA
+    server_id: code_https
+    available: true
+    description: Current CODE rapid differential code biases (DCB/OSB)
+    parameters:
+      - {name: AAA, value: COD}
+      - {name: TTT, value: RAP}
+      - {name: PPP, value: OPS}
+    directory: {pattern: "CODE/"}
+
   # ── Ionosphere ─────────────────────────────────────────────────
-  - id: code_gim
+  - id: code_gim_final
     product_name: IONEX
-    server_id: code_ftp
+    server_id: code_https
     available: true
-    description: Global Ionosphere Maps (CODE is primary producer)
+    description: Archived CODE final Global Ionosphere Maps
     parameters:
       - {name: AAA, value: COD}
       - {name: TTT, value: FIN}
+      - {name: PPP, value: OPS}
+      - {name: SMP, value: 01H}
+      - {name: SMP, value: 02H}
+    directory: {pattern: "CODE/{YYYY}/"}
+
+  - id: code_gim_operational
+    product_name: IONEX
+    server_id: code_https
+    available: true
+    description: Current CODE rapid and predicted Global Ionosphere Maps
+    parameters:
+      - {name: AAA, value: COD}
       - {name: TTT, value: RAP}
       - {name: TTT, value: PRD}
       - {name: PPP, value: OPS}
       - {name: SMP, value: 01H}
       - {name: SMP, value: 02H}
-    directory: {pattern: "CODE/{YYYY}/"}
+    directory: {pattern: "CODE/"}
 
   # ── SINEX weekly solutions ──────────────────────────────────────
   - id: code_sinex
     product_name: SINEX
-    server_id: code_ftp
+    server_id: code_https
     available: true
     description: CODE weekly station coordinate SINEX solutions
     parameters:
@@ -99,7 +159,7 @@ products:
   # ── Troposphere SINEX ──────────────────────────────────────────
   - id: code_trop
     product_name: TROP
-    server_id: code_ftp
+    server_id: code_https
     available: true
     description: CODE troposphere SINEX (zenith total delay + horizontal gradients)
     parameters:
diff --git a/packages/gpm-specs/src/gpm_specs/configs/centers/wuhan_config.yaml b/packages/gpm-specs/src/gpm_specs/configs/centers/wuhan_config.yaml
index 9e8fbc0..d2fc91d 100644
--- a/packages/gpm-specs/src/gpm_specs/configs/centers/wuhan_config.yaml
+++ b/packages/gpm-specs/src/gpm_specs/configs/centers/wuhan_config.yaml
@@ -16,7 +16,12 @@ products:
     product_name: ORBIT
     server_id: wuhan_ftp
     available: true
-    description: Precise satellite orbits
+    description: >
+      Precise satellite orbits. No TTT filter, so this already matches
+      RTS (Real-Time Service) candidates alongside FIN/RAP; TTT sorting
+      in pride_pppar.yaml doesn't list RTS, so those candidates just
+      sort last. See wuhan_clock for why CLOCK/ERP/BIA need an explicit
+      TTT: RTS entry to get the same behavior.
     parameters:
       - {name: AAA, value: WUM}
       - {name: AAA, value: WMC}
@@ -92,16 +97,25 @@ products:
     product_name: CLOCK
     server_id: wuhan_ftp
     available: true
-    description: Precise satellite and station clocks
+    description: >
+      Precise satellite and station clocks. Includes RTS (Real-Time
+      Service), WUM's own near-real-time product line — no FIN/RAP is
+      published yet for the most recent 1-3 days, and this is the same
+      fallback PRIDE-PPPAR's pdp3.sh uses natively for recent days
+      (USERTS=YES). TTT sorting in pride_pppar.yaml only lists
+      FIN/RAP/ULT, so RTS candidates naturally sort last without
+      needing a preference entry.
     parameters:
       - {name: AAA, value: WUM}
       - {name: AAA, value: WMC}
       - {name: TTT, value: FIN}
       - {name: TTT, value: RAP}
+      - {name: TTT, value: RTS}
       - {name: PPP, value: MGX}
       - {name: PPP, value: DEM}
       - {name: SMP, value: 30S}
       - {name: SMP, value: 05M}
+      - {name: SMP, value: 05S}
     directory: {pattern: "pub/whu/phasebias/{YYYY}/clock/"}
 
   # ── ERP ─────────────────────────────────────────────────────────
@@ -109,11 +123,13 @@ products:
     product_name: ERP
     server_id: wuhan_ftp
     available: true
-    description: Earth rotation parameters
+    description: >
+      Earth rotation parameters. Includes RTS (see wuhan_clock note).
     parameters:
       - {name: AAA, value: WUM}
       - {name: TTT, value: FIN}
       - {name: TTT, value: RAP}
+      - {name: TTT, value: RTS}
       - {name: PPP, value: MGX}
     directory: {pattern: "pub/whu/phasebias/{YYYY}/orbit/"}
 
@@ -122,12 +138,15 @@ products:
     product_name: BIA
     server_id: wuhan_ftp
     available: true
-    description: Observable-specific signal biases (OSB)
+    description: >
+      Observable-specific signal biases (OSB). Includes RTS (see
+      wuhan_clock note).
     parameters:
       - {name: AAA, value: WUM}
       - {name: AAA, value: WMC}
       - {name: TTT, value: FIN}
       - {name: TTT, value: RAP}
+      - {name: TTT, value: RTS}
       - {name: PPP, value: MGX}
       - {name: PPP, value: DEM}
     directory: {pattern: "pub/whu/phasebias/{YYYY}/bias/"}
diff --git a/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar.yaml b/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar.yaml
index ede85b7..c5e8556 100644
--- a/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar.yaml
+++ b/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar.yaml
@@ -17,14 +17,26 @@ package: PRIDE
 task: PPP
 
 preferences:
+  - parameter: CCC
+    sorting: [WRD, IGS, IGN, DLR]
+    description: >
+      Prefer BKG's rolling WRD mixed-navigation file when it is available;
+      otherwise use the normal daily navigation sources.
   - parameter: AAA
     sorting: [WUM, COD, GFZ, ESA]
     description: >
       Prefer Wuhan, then CODE, GFZ, ESA analysis centres.
   - parameter: TTT
-    sorting: [FIN, RAP, ULT]
+    sorting: [FIN, RTS, RAP, ULT]
     description: >
-      Prefer final solutions, then rapid, then ultra-rapid.
+      Prefer final solutions, then Wuhan real-time streaming products,
+      then rapid and ultra-rapid products. RTS biases include the Galileo
+      phase signals needed for E17 processing before RAP products do.
+
+bundles:
+  - name: precise-products
+    members: [ORBIT, CLOCK, ERP, BIA, ATTOBX]
+    coherence: [AAA, TTT, PPP]
 
 dependencies:
   - spec: ORBIT
@@ -44,23 +56,33 @@ dependencies:
     description: Observable-specific signal biases
 
   - spec: ATTOBX
-    required: true
+    # pdp3.sh tolerates a missing attitude product outright (sets
+    # Quaternions = NONE and skips multipath/attitude modeling) — no AC
+    # publishes ATTOBX faster than FIN/RAP (~1 day lag), so requiring it
+    # blocks same-day/near-real-time processing pdp3 could otherwise do.
+    required: false
     description: Satellite attitude quaternions (OBX)
 
   - spec: ATTATX
     required: true
+    refresh_on_force: false
     description: Antenna phase center corrections (ANTEX)
 
   - spec: RNX3_BRDC
-    required: true
+    # BKG's rolling WRD file normally supplies same-day multi-GNSS navigation.
+    # Keep this optional: if WRD and the daily sources are not yet available,
+    # pdp3.sh falls back to Wuhan's hourly GPS+GLONASS navigation builder.
+    required: false
     description: Broadcast navigation RINEX 3
 
   - spec: LEAP_SEC
     required: true
+    refresh_on_force: false
     description: Leap second table
 
   - spec: SAT_PARAMS
     required: true
+    refresh_on_force: false
     description: Satellite metadata parameters table
 
   # ---------------------------------------------------------------------------
@@ -68,28 +90,35 @@ dependencies:
   # ---------------------------------------------------------------------------
   - spec: FILE_NAME_TABLE
     required: true
+    refresh_on_force: false
     description: Satellite name/identifier lookup table
 
   - spec: OCEAN_TIDE_FES2004
     required: true
+    refresh_on_force: false
     description: FES2004 ocean tide model coefficients
 
   - spec: OCEAN_LOAD_GREEN
     required: true
+    refresh_on_force: false
     description: Green's function coefficients for ocean tidal loading
 
   - spec: GPT3_GRID
     required: true
+    refresh_on_force: false
     description: GPT3 global pressure and temperature model grid
 
   - spec: OCEAN_LOAD_PROGRAM
     required: true
+    refresh_on_force: false
     description: Pre-computed ocean loading parameters
 
   - spec: OROGRAPHY_1X1
     required: true
+    refresh_on_force: false
     description: Ellipsoidal terrain height grid (1x1)
 
   - spec: OROGRAPHY
     required: true
+    refresh_on_force: false
     description: Ellipsoidal terrain height grid (VMF1)
diff --git a/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar_final.yaml b/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar_final.yaml
index 2512949..3b50c6e 100644
--- a/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar_final.yaml
+++ b/packages/pride-ppp/src/pride_ppp/configs/dependencies/pride_pppar_final.yaml
@@ -24,6 +24,11 @@ preferences:
     description: >
       Only accept final solutions.
 
+bundles:
+  - name: precise-products
+    members: [ORBIT, CLOCK, ERP, BIA, ATTOBX]
+    coherence: [AAA, TTT, PPP]
+
 dependencies:
   - spec: ORBIT
     constraints:
@@ -57,6 +62,7 @@ dependencies:
 
   - spec: ATTATX
     required: true
+    refresh_on_force: false
     description: Antenna phase center corrections (ANTEX)
 
   - spec: RNX3_BRDC
@@ -65,10 +71,12 @@ dependencies:
 
   - spec: LEAP_SEC
     required: true
+    refresh_on_force: false
     description: Leap second table
 
   - spec: SAT_PARAMS
     required: true
+    refresh_on_force: false
     description: Satellite metadata parameters table
 
   # ---------------------------------------------------------------------------
@@ -76,28 +84,35 @@ dependencies:
   # ---------------------------------------------------------------------------
   - spec: FILE_NAME_TABLE
     required: true
+    refresh_on_force: false
     description: Satellite name/identifier lookup table
 
   - spec: OCEAN_TIDE_FES2004
     required: true
+    refresh_on_force: false
     description: FES2004 ocean tide model coefficients
 
   - spec: OCEAN_LOAD_GREEN
     required: true
+    refresh_on_force: false
     description: Green's function coefficients for ocean tidal loading
 
   - spec: GPT3_GRID
     required: true
+    refresh_on_force: false
     description: GPT3 global pressure and temperature model grid
 
   - spec: OCEAN_LOAD_PROGRAM
     required: true
+    refresh_on_force: false
     description: Pre-computed ocean loading parameters
 
   - spec: OROGRAPHY_1X1
     required: true
+    refresh_on_force: false
     description: Ellipsoidal terrain height grid (1x1)
 
   - spec: OROGRAPHY
     required: true
+    refresh_on_force: false
     description: Ellipsoidal terrain height grid (VMF1)
diff --git a/packages/pride-ppp/src/pride_ppp/configs/products/pride_product_spec.yaml b/packages/pride-ppp/src/pride_ppp/configs/products/pride_product_spec.yaml
index 865bf96..d70c94a 100644
--- a/packages/pride-ppp/src/pride_ppp/configs/products/pride_product_spec.yaml
+++ b/packages/pride-ppp/src/pride_ppp/configs/products/pride_product_spec.yaml
@@ -62,27 +62,3 @@ products:
         version: "1"
         file_templates:
           - "oceanload"
-
-  OROGRAPHY_1X1:
-    description: >
-      Ellipsoidal terrain height grid (1x1) for VMF interpolation.
-
-    formats:
-
-      - format: VIENNA_MAPPING_FUNCTIONS
-        version: "1"
-        constraints:
-          RESOLUTION: "1x1"
-        variant: orography
-
-  OROGRAPHY:
-    description: >
-      Ellipsoidal terrain height grid (VMF1) bundled with PRIDE-PPPAR.
-      Required by PrepareTables regardless of the selected mapping function.
-
-    formats:
-
-      - format: TABLE
-        version: "1"
-        file_templates:
-          - "orography_ell"
diff --git a/packages/pride-ppp/src/pride_ppp/defaults/__init__.py b/packages/pride-ppp/src/pride_ppp/defaults/__init__.py
index 3dccc59..fbe244a 100644
--- a/packages/pride-ppp/src/pride_ppp/defaults/__init__.py
+++ b/packages/pride-ppp/src/pride_ppp/defaults/__init__.py
@@ -15,7 +15,7 @@
 
 _CONFIGS_DIR = Path(__file__).resolve().parent.parent / "configs"
 
-# Dependency spec: FIN → RAP → ULT cascade (default processing mode).
+# Dependency spec: FIN → RTS → RAP → ULT cascade (default processing mode).
 PRIDE_PPPAR_SPEC = _CONFIGS_DIR / "dependencies" / "pride_pppar.yaml"
 
 # Dependency spec: FINAL-only products (TTT restricted to [FIN]).
diff --git a/packages/pride-ppp/src/pride_ppp/factories/output.py b/packages/pride-ppp/src/pride_ppp/factories/output.py
index bf66995..e5e608f 100644
--- a/packages/pride-ppp/src/pride_ppp/factories/output.py
+++ b/packages/pride-ppp/src/pride_ppp/factories/output.py
@@ -7,7 +7,7 @@
 
 import logging
 import os
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
 import pandas as pd
@@ -59,22 +59,18 @@ def get_wrms_from_res(res_path):
                 sumOfSquares = 0
                 sumOfWeights = 0
 
-                seconds_str = line_data[6]
-                if "." in seconds_str:
-                    SS, fractional = seconds_str.split(".")
-                    SS = int(SS)
-                    fractional = fractional.ljust(6, "0")[:6]
-                else:
-                    SS = int(seconds_str)
-                    fractional = "000000"
-
-                isodate = (
-                    f"{line_data[1]}-{line_data[2].zfill(2)}-{line_data[3].zfill(2)}"
-                    f"T{line_data[4].zfill(2)}:{line_data[5].zfill(2)}:{str(SS).zfill(2)}"
-                    f".{fractional}+00:00"
-                )
-
-                timestamp = datetime.fromisoformat(isodate)
+                # PRIDE occasionally formats a rounded epoch with seconds
+                # equal to 60.0000000.  Constructing an ISO timestamp with
+                # second=60 is invalid; adding the seconds as a timedelta
+                # correctly carries it into the following minute/day.
+                timestamp = datetime(
+                    int(line_data[1]),
+                    int(line_data[2]),
+                    int(line_data[3]),
+                    int(line_data[4]),
+                    int(line_data[5]),
+                    tzinfo=timezone.utc,
+                ) + timedelta(seconds=float(line_data[6]))
                 timestamps.append(timestamp)
 
                 line = res_file.readline()
@@ -100,7 +96,9 @@ def get_wrms_from_res(res_path):
                     if line == "":
                         break
                     line_data = line.split()
-                wrms = (sumOfSquares / sumOfWeights) ** 0.5 * 1000 if sumOfWeights else float("nan")  # in mm
+                wrms = (
+                    (sumOfSquares / sumOfWeights) ** 0.5 * 1000 if sumOfWeights else float("nan")
+                )  # in mm
                 data.append(wrms)
             else:
                 line = res_file.readline()
diff --git a/packages/pride-ppp/src/pride_ppp/factories/processor.py b/packages/pride-ppp/src/pride_ppp/factories/processor.py
index 1ae8c74..32fcd60 100644
--- a/packages/pride-ppp/src/pride_ppp/factories/processor.py
+++ b/packages/pride-ppp/src/pride_ppp/factories/processor.py
@@ -51,6 +51,7 @@
 from ..specifications.cli import PrideCLIConfig
 from ..specifications.config import PRIDEPPPFileConfig, SatelliteProducts
 from .output import get_wrms_from_res, kin_to_kin_position_df
+from .product_validation import product_epoch_bounds, validate_pride_products
 from .rinex import rinex_get_time_range
 
 logger = logging.getLogger(__name__)
@@ -66,7 +67,7 @@ class ProcessingMode(enum.Enum):
 
     Selects which dependency-spec YAML governs which products are accepted.
 
-    * ``DEFAULT`` — cascades through FIN → RAP → ULT.  Uses the best
+    * ``DEFAULT`` — cascades through FIN → RTS → RAP → ULT. Uses the best
       available product at run time.  Suitable for near-real-time processing
       or when the observation date is within the last two weeks.
     * ``FINAL``   — accepts only IGS final (FIN) products (available ≥13 days
@@ -271,11 +272,24 @@ def _resolution_to_satellite_products(
 
     return (
         SatelliteProducts(
-            satellite_orbit=product_fields.get("satellite_orbit"),
-            satellite_clock=product_fields.get("satellite_clock"),
-            code_phase_bias=product_fields.get("code_phase_bias"),
-            quaternions=product_fields.get("quaternions"),
-            erp=product_fields.get("erp"),
+            # .get(key, "Default") — not .get(key): a bare None here is passed
+            # to the pydantic model explicitly, which bypasses the field's own
+            # "Default" default and gets f-string'd into the config file as the
+            # literal text "None". pdp3.sh only recognizes "Default" as its
+            # let-me-resolve-it-myself sentinel, so "None" makes it try to
+            # download a product file literally named "None" and fail outright
+            # — worse than the missing-required abort this is meant to replace
+            # for optional specs.  ATTOBX is different: RTS bundles do not
+            # provide satellite attitude, and leaving Quaternions as Default
+            # makes pdp3 attempt slow fallback downloads.  Its missing-file
+            # branch also uses a non-portable ``sed -i`` invocation that
+            # corrupts the generated config on macOS.  NONE is PRIDE's
+            # supported spelling for processing without an attitude product.
+            satellite_orbit=product_fields.get("satellite_orbit", "Default"),
+            satellite_clock=product_fields.get("satellite_clock", "Default"),
+            code_phase_bias=product_fields.get("code_phase_bias", "Default"),
+            quaternions=product_fields.get("quaternions", "NONE"),
+            erp=product_fields.get("erp", "Default"),
             product_directory=str(product_dir) if product_dir else "Default",
         ),
         product_dir,
@@ -305,6 +319,102 @@ def _resolution_to_table_dir(resolution: DependencyResolution) -> Path | None:
     return None
 
 
+def _stage_broadcast_navigation(
+    resolution: DependencyResolution,
+    rinex: Path,
+    date: datetime.date,
+) -> Path | None:
+    """Place resolved mixed navigation where ``pdp3`` expects to find it.
+
+    PRIDE derives the legacy mixed-navigation name from the observation file
+    and searches in the observation file's directory. Product resolution,
+    however, stores the RINEX 3 navigation product in the PRIDE workspace.
+    Bridge those layouts before invoking ``pdp3``.
+    """
+    source: Path | None = None
+    for rd in resolution.fulfilled:
+        if rd.spec == "RNX3_BRDC" and rd.local_path is not None:
+            source = Path(as_path(rd.local_path))
+            break
+
+    if source is None:
+        return None
+
+    destination = rinex.parent / f"brdm{date.timetuple().tm_yday:03d}0.{date.year % 100:02d}p"
+    if source.resolve() == destination.resolve():
+        return destination
+
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(source, destination)
+    logger.info("Staged broadcast navigation %s from %s", destination, source)
+    return destination
+
+
+def _prepare_rinex_for_product_coverage(
+    rinex: Path,
+    resolution: DependencyResolution,
+    date: datetime.date,
+    work_dir: Path,
+) -> Path:
+    """Clip same-day observations to safe orbit/clock coverage."""
+    if date != datetime.datetime.now(datetime.timezone.utc).date():
+        return rinex
+    by_spec = {
+        rd.spec: Path(as_path(rd.local_path))
+        for rd in resolution.fulfilled
+        if rd.local_path is not None and rd.spec in {"ORBIT", "CLOCK"}
+    }
+    bounds = [
+        product_epoch_bounds(path, product) for product, path in by_spec.items() if path.exists()
+    ]
+    bounds = [bound for bound in bounds if bound is not None]
+    if len(bounds) != 2:
+        return rinex
+    obs_start, obs_end = rinex_get_time_range(rinex)
+    coverage_end = min(bound[1] for bound in bounds)
+    safe_end = coverage_end - datetime.timedelta(minutes=5)
+    if obs_end <= safe_end:
+        return rinex
+    if safe_end <= obs_start:
+        raise ValueError(f"No observations inside product coverage ending {coverage_end}")
+
+    clipped = work_dir / "clipped_rinex" / rinex.name
+    clipped.parent.mkdir(parents=True, exist_ok=True)
+    keep_epoch = False
+    in_header = True
+    with rinex.open(errors="replace") as source, clipped.open("w") as destination:
+        for line in source:
+            if in_header:
+                destination.write(line)
+                if "END OF HEADER" in line:
+                    in_header = False
+                continue
+            if line.startswith(">"):
+                fields = line.split()
+                epoch = datetime.datetime(
+                    int(fields[1]),
+                    int(fields[2]),
+                    int(fields[3]),
+                    int(fields[4]),
+                    int(fields[5]),
+                    int(float(fields[6])),
+                )
+                keep_epoch = epoch <= safe_end
+                if not keep_epoch:
+                    break
+            if keep_epoch:
+                destination.write(line)
+    logger.warning(
+        "Same-day observations end at %s but common orbit/clock coverage ends at %s; "
+        "using clipped RINEX ending by %s: %s",
+        obs_end,
+        coverage_end,
+        safe_end,
+        clipped,
+    )
+    return clipped
+
+
 def _write_config(
     satellite_products: SatelliteProducts,
     table_dir: Path | None,
@@ -380,6 +490,7 @@ def __init__(
         pride_install_dir: Path | None = None,
         cli_config: PrideCLIConfig | None = PrideCLIConfig(),
         mode: ProcessingMode | Literal["FINAL", "DEFAULT"] = ProcessingMode.DEFAULT,
+        override_products_download: bool = False,
     ) -> None:
         """Initialise the processor and all its owned subsystems.
 
@@ -403,11 +514,13 @@ def __init__(
             mode: Product timeliness mode.  Selects which dependency-spec
                 YAML governs product resolution:
 
-                * ``ProcessingMode.DEFAULT`` — FIN → RAP → ULT cascade.
+                * ``ProcessingMode.DEFAULT`` — FIN → RTS → RAP → ULT cascade.
                 * ``ProcessingMode.FINAL``   — only FINAL products.
 
                 Also accepts the string literals ``"DEFAULT"`` or
                 ``"FINAL"`` for convenience.
+            override_products_download: Ignore product lockfiles and local
+                cached products, downloading fresh remote copies instead.
         """
         if isinstance(mode, str):
             mode = ProcessingMode(mode.upper())
@@ -416,6 +529,7 @@ def __init__(
         self._pride_install_dir = Path(pride_install_dir) if pride_install_dir else None
         self._cli_config = cli_config if cli_config is not None else PrideCLIConfig()
         self._mode = mode
+        self._override_products_download = override_products_download
 
         # Load the DependencySpec that matches the requested processing mode.
         # The dep-spec controls which TTT (timeliness) values the resolver
@@ -508,6 +622,7 @@ def _resolve(
         self,
         date: datetime.datetime,
         local_sink_id: str = "pride",
+        centers: list[str] | None = None,
     ) -> DependencyResolution:
         """Resolve all dependencies for a single UTC date.
 
@@ -520,18 +635,45 @@ def _resolve(
             date: Target date (midnight UTC) for product resolution.
             local_sink_id: WorkSpace alias that receives downloaded files.
                 Defaults to ``"pride"`` which maps to ``self._pride_dir``.
+            centers: Optional analysis-center resource IDs to search.  Primarily
+                useful for targeted diagnostics and controlled processing.
 
         Returns:
             A :class:`DependencyResolution` containing fulfilled and missing
             product entries.
         """
-        resolution, _ = self._client.resolve_dependencies(
-            self._dep_spec,
-            date,
-            sink_id=local_sink_id,
-        )
+        kwargs = {
+            "sink_id": local_sink_id,
+            "force_download": self._override_products_download,
+        }
+        if centers is not None:
+            kwargs["bundle_centers"] = centers
+        resolution, _ = self._client.resolve_dependencies(self._dep_spec, date, **kwargs)
         return resolution
 
+    def _add_product_diagnostics(
+        self, rinex_paths: Sequence[Path], resolution: DependencyResolution
+    ) -> None:
+        """Attach non-fatal product coverage diagnostics to a resolution."""
+        try:
+            cli_config = getattr(self, "_cli_config", None) or PrideCLIConfig()
+            messages = validate_pride_products(
+                rinex_paths,
+                resolution,
+                cli_config.frequency,
+            )
+        except (OSError, ValueError) as exc:
+            messages = [f"Product compatibility diagnostics unavailable: {exc}"]
+        resolution.diagnostics.extend(messages)
+        for message in messages:
+            if any(
+                marker in message
+                for marker in ("partial", "not present", "could not", "unavailable")
+            ):
+                logger.warning("Product compatibility: %s", message)
+            else:
+                logger.info("Product compatibility: %s", message)
+
     # ------------------------------------------------------------------ #
     # Directory helpers
     # ------------------------------------------------------------------ #
@@ -546,7 +688,15 @@ def _working_dir(self, date: datetime.date) -> Path:
     # ------------------------------------------------------------------ #
     # Subprocess execution
     # ------------------------------------------------------------------ #
-    def _build_pdp_command(self, rinex: Path, site: str, config_path: Path) -> list[str]:
+    def _build_pdp_command(
+        self,
+        rinex: Path,
+        site: str,
+        config_path: Path,
+        *,
+        resolution: DependencyResolution,
+        date: datetime.date,
+    ) -> list[str]:
         """Assemble the full ``pdp3`` command-line invocation.
 
         Clones the processor's CLI config, overriding
@@ -557,16 +707,43 @@ def _build_pdp_command(self, rinex: Path, site: str, config_path: Path) -> list[
             rinex: Path to the observation file passed to pdp3.
             site: 4-char site identifier (e.g. ``"NCC1"``).
             config_path: The ``config_file`` written by ``_write_config``.
+            resolution: Products resolved for the observation date.
+            date: Observation date used to detect the same-day nav fallback.
 
         Returns:
             A list of strings suitable for ``subprocess.run()``.
         """
-        cli = PrideCLIConfig(
-            **{
-                **self._cli_config.model_dump(),
-                "pride_configfile_path": config_path,
-            }
+        values = {
+            **self._cli_config.model_dump(),
+            "pride_configfile_path": config_path,
+        }
+        has_mixed_navigation = any(
+            rd.spec == "RNX3_BRDC" and rd.local_path is not None for rd in resolution.fulfilled
         )
+        today_utc = datetime.datetime.now(datetime.timezone.utc).date()
+        if date == today_utc:
+            values["mapping_function"] = "GMF"
+            logger.info(
+                "Using GMF for same-day PRIDE processing on %s; VMF grids "
+                "covering the complete interpolation window may not yet be published",
+                date,
+            )
+        if date == today_utc and not has_mixed_navigation:
+            frequencies = [f for f in self._cli_config.frequency if f.startswith(("G", "R"))]
+            if not frequencies:
+                raise ValueError(
+                    "Same-day GPS/GLONASS navigation fallback requires at least one "
+                    "GPS or GLONASS frequency combination"
+                )
+            values.update(system="GR", frequency=frequencies)
+            logger.warning(
+                "No mixed-GNSS broadcast navigation available for %s; limiting "
+                "pdp3 to GPS/GLONASS frequencies %s for its hourly nav fallback",
+                date,
+                frequencies,
+            )
+
+        cli = PrideCLIConfig(**values)
         return cli.generate_pdp_command(site=site, local_file_path=str(rinex))
 
     @staticmethod
@@ -582,9 +759,10 @@ def _run_pdp3(
         ``pride_dir/{year}/{doy}/`` directory so these artefacts are
         available for inspection after the run.
 
-        After execution the method searches recursively for ``kin_*`` and
-        ``res_*`` output files matching *site*, appends ``.kin`` / ``.res``
-        extensions, and moves them to *output_dir*.
+        After execution the method searches recursively for the site outputs.
+        A valid solution preserves ``kin``, ``res``, the runtime config,
+        editing log, ambiguity constraints, residual statistics, and captured
+        console output in *output_dir* using a common ``YYYYDOY_site`` suffix.
 
         Args:
             command: Full pdp3 argument list from ``_build_pdp_command``.
@@ -593,10 +771,28 @@ def _run_pdp3(
 
         Returns:
             ``(kin_path, res_path, returncode, stderr)`` where paths are
-            ``None`` when the corresponding output was not produced.
+            ``None`` when the corresponding output was not produced or
+            didn't pass validation (see below).
 
         Raises:
             FileNotFoundError: If the ``pdp3`` binary is not on ``PATH``.
+
+        Note:
+            ``pdp3`` can exit 0 and still leave a stale ``kin_*`` file
+            behind from an earlier, non-fatal stage (e.g. the initial
+            single-point-positioning seed file written before the real
+            ambiguity-resolution step) even when a later internal stage
+            fails outright. Trusting a bare filename-glob match would
+            silently accept that low-quality leftover as a successful
+            result, so two independent checks guard against it: (1)
+            ``pdp3.sh`` prints its own ``error:``-tagged lines to stdout on
+            a hard stage failure — distinct from its ``warning:`` lines for
+            recoverable conditions — which is scanned for regardless of
+            the process's own exit code; (2) any ``kin_*`` match found is
+            parsed with the same validator used to check cached output
+            (:func:`kin_to_kin_position_df`) before being trusted, since a
+            leftover seed file has a different (and much sparser) column
+            layout that fails to parse.
         """
         if not shutil.which("pdp3"):
             raise FileNotFoundError("pdp3 binary not found in PATH")
@@ -608,12 +804,14 @@ def _run_pdp3(
                 cwd=tmpdir,
                 capture_output=True,
                 text=True,
+                encoding="utf-8",
+                errors="replace",
             )
 
             # Replay stdout/stderr through the logger for observability
-            if result.stdout:
-                for line in result.stdout.strip().splitlines():
-                    logger.info(line)
+            stdout_lines = result.stdout.strip().splitlines() if result.stdout else []
+            for line in stdout_lines:
+                logger.info(line)
             if result.stderr:
                 for line in result.stderr.strip().splitlines():
                     logger.warning(line)
@@ -627,6 +825,15 @@ def _run_pdp3(
                     stderr_tail or "(no stderr)",
                 )
 
+            pdp3_error_lines = [line for line in stdout_lines if "error:" in line.lower()]
+            if pdp3_error_lines:
+                logger.error(
+                    "pdp3 reported failure for site %s despite returncode %d: %s",
+                    site,
+                    result.returncode,
+                    " | ".join(pdp3_error_lines),
+                )
+
             # pdp3 writes outputs as e.g. "kin_2025254_ncc1" (no extension).
             # Search recursively in the working dir to find them.
             kin_files = list(Path(tmpdir).rglob(f"kin_*_{site.lower()}"))
@@ -638,26 +845,67 @@ def _run_pdp3(
             output_dir.mkdir(parents=True, exist_ok=True)
 
             # Move outputs to the final output directory with proper extensions
-            if kin_files:
-                src = kin_files[0]
-                dst = output_dir / (src.name + ".kin")
-                shutil.move(str(src), str(dst))
-                kin_out = dst
-                logger.info("Generated kin file %s", dst)
-            else:
+            if not kin_files:
                 logger.error(
                     "pdp3 produced no kin output for site %s (returncode %d)",
                     site,
                     result.returncode,
                 )
+            elif pdp3_error_lines:
+                logger.error(
+                    "Discarding kin output for site %s: pdp3 reported an internal "
+                    "failure, so %s is likely an incomplete/seed-only leftover, "
+                    "not a real result.",
+                    site,
+                    kin_files[0].name,
+                )
+            elif kin_to_kin_position_df(kin_files[0]) is None:
+                logger.error(
+                    "Discarding kin output for site %s: %s exists but failed to "
+                    "parse as a valid kinematic position file.",
+                    site,
+                    kin_files[0].name,
+                )
+            else:
+                src = kin_files[0]
+                dst = output_dir / (src.name + ".kin")
+                shutil.move(str(src), str(dst))
+                kin_out = dst
+                logger.info("Generated kin file %s", dst)
 
-            if res_files:
+            if kin_out is not None and res_files:
                 src = res_files[0]
                 dst = output_dir / (src.name + ".res")
                 shutil.move(str(src), str(dst))
                 res_out = dst
                 logger.info("Generated res file %s", dst)
 
+            if kin_out is not None:
+                # The validated KIN name is authoritative for the session ID,
+                # e.g. kin_2026247_ncc1 -> 2026247_ncc1.  Keep the actual
+                # runtime-mutated config rather than the pre-pdp3 template.
+                session_id = kin_out.stem.removeprefix("kin_")
+                ancillary = {
+                    "config": list(Path(tmpdir).rglob("config.*")),
+                    "log": list(Path(tmpdir).rglob(f"log_{session_id}")),
+                    "cst": list(Path(tmpdir).rglob(f"cst_{session_id}")),
+                    "stt": list(Path(tmpdir).rglob(f"stt_{session_id}")),
+                }
+                for kind, sources in ancillary.items():
+                    if not sources:
+                        logger.warning("pdp3 produced no %s output for %s", kind, session_id)
+                        continue
+                    destination = output_dir / f"{kind}_{session_id}.{kind}"
+                    shutil.move(str(sources[0]), str(destination))
+                    logger.info("Generated %s file %s", kind, destination)
+
+                run_log = output_dir / f"run_{session_id}.log"
+                sections = ["=== pdp3 stdout ===\n", result.stdout or ""]
+                if result.stderr:
+                    sections.extend(["\n=== pdp3 stderr ===\n", result.stderr])
+                run_log.write_text("".join(sections), encoding="utf-8")
+                logger.info("Generated run log %s", run_log)
+
         return kin_out, res_out, result.returncode, result.stderr
 
     def _build_kin_res_paths(
@@ -788,6 +1036,7 @@ def process(
         # --- 2. Resolve products ----------------------------------------------
         logger.info("Resolving products for %s (site=%s)", start_date, site)
         resolution = self._resolve(target_dt)
+        self._add_product_diagnostics([rinex], resolution)
         logger.info(resolution.summary())
 
         # --- 3. Check cache ---------------------------------------------------
@@ -820,10 +1069,18 @@ def process(
         # be inspected after the run and reused by subsequent pdp3 calls for
         # the same date.
         work_dir = self._working_dir(start_date)
+        pdp_rinex = _prepare_rinex_for_product_coverage(rinex, resolution, start_date, work_dir)
+        _stage_broadcast_navigation(resolution, pdp_rinex, start_date)
         sat_products, _ = _resolution_to_satellite_products(resolution)
         table_dir = _resolution_to_table_dir(resolution)
         config_path = _write_config(sat_products, table_dir, work_dir / "config_file")
-        command = self._build_pdp_command(rinex=rinex, site=site, config_path=config_path)
+        command = self._build_pdp_command(
+            rinex=pdp_rinex,
+            site=site,
+            config_path=config_path,
+            resolution=resolution,
+            date=start_date,
+        )
 
         # --- 5. Run pdp3 ------------------------------------------------------
         kin_path, res_path, returncode, stderr = self._run_pdp3(
@@ -911,6 +1168,7 @@ def process_batch(
         jobs_sorted = sorted(jobs, key=lambda j: j[2])
         resolutions: dict[datetime.date, DependencyResolution] = {}
         for date_key, group in groupby(jobs_sorted, key=lambda j: j[2]):
+            date_jobs = list(group)
             target_dt = datetime.datetime(
                 date_key.year,
                 date_key.month,
@@ -919,6 +1177,9 @@ def process_batch(
             )
             logger.info("Resolving products for %s", date_key)
             resolutions[date_key] = self._resolve(target_dt)
+            self._add_product_diagnostics(
+                [rinex for rinex, _site, _date in date_jobs], resolutions[date_key]
+            )
             logger.info(resolutions[date_key].summary())
 
         # --- Step 3: Write per-date config files in year/doy dirs ---------------
@@ -981,10 +1242,15 @@ def process_batch(
                 )
                 continue
 
+            pdp_rinex = _prepare_rinex_for_product_coverage(rinex, resolutions[d], d, work_dirs[d])
+            _stage_broadcast_navigation(resolutions[d], pdp_rinex, d)
+
             command = self._build_pdp_command(
-                rinex=rinex,
+                rinex=pdp_rinex,
                 site=site,
                 config_path=config_paths[d],
+                resolution=resolutions[d],
+                date=d,
             )
             pending.append((i, command, site, d))
 
diff --git a/packages/pride-ppp/src/pride_ppp/factories/product_validation.py b/packages/pride-ppp/src/pride_ppp/factories/product_validation.py
new file mode 100644
index 0000000..2e0c9c4
--- /dev/null
+++ b/packages/pride-ppp/src/pride_ppp/factories/product_validation.py
@@ -0,0 +1,168 @@
+"""Advisory checks for PRIDE precise-product compatibility."""
+
+from __future__ import annotations
+
+import datetime
+import re
+from collections.abc import Iterable
+from pathlib import Path
+
+from gnss_product_management.specifications.dependencies.dependencies import DependencyResolution
+
+from .rinex import rinex_get_time_range
+
+_OBS = re.compile(r"^[CLDS][1-9][A-Z]$")
+_SAT = re.compile(r"^[GRECJ]\d{2,3}$")
+
+
+def rinex_phase_observables(paths: Iterable[Path]) -> dict[str, set[str]]:
+    """Return phase observables advertised by RINEX 3/4 headers."""
+    observables: dict[str, set[str]] = {}
+    for path in paths:
+        current_system = ""
+        with Path(path).open(errors="replace") as stream:
+            for line in stream:
+                if "END OF HEADER" in line:
+                    break
+                if "SYS / # / OBS TYPES" not in line:
+                    continue
+                is_first_line = bool(line[:1].strip())
+                if is_first_line:
+                    current_system = line[0]
+                if not current_system:
+                    continue
+                tokens = line[:60].split()
+                for token in tokens[2:] if is_first_line else tokens:
+                    if _OBS.match(token) and token.startswith("L"):
+                        observables.setdefault(current_system, set()).add(token)
+    return observables
+
+
+def bias_phase_observables(path: Path) -> dict[str, set[str]]:
+    """Return satellite phase observables present in a SINEX BIA file."""
+    observables: dict[str, set[str]] = {}
+    with path.open(errors="replace") as stream:
+        for line in stream:
+            if not line.lstrip().startswith("OSB"):
+                continue
+            tokens = line.split()
+            satellite = next((token for token in tokens if _SAT.match(token)), None)
+            observable = next(
+                (token for token in tokens if _OBS.match(token) and token.startswith("L")), None
+            )
+            if satellite and observable:
+                observables.setdefault(satellite[0], set()).add(observable)
+    return observables
+
+
+def rinex_phase_bands(paths: Iterable[Path]) -> dict[str, set[str]]:
+    """Return phase-frequency bands advertised by RINEX 3/4 headers."""
+    return {
+        system: {observable[1] for observable in observables}
+        for system, observables in rinex_phase_observables(paths).items()
+    }
+
+
+def bias_phase_bands(path: Path) -> dict[str, set[str]]:
+    """Return satellite phase-frequency bands present in a SINEX BIA file."""
+    return {
+        system: {observable[1] for observable in observables}
+        for system, observables in bias_phase_observables(path).items()
+    }
+
+
+def product_epoch_bounds(
+    path: Path, product: str
+) -> tuple[datetime.datetime, datetime.datetime] | None:
+    """Read coarse epoch bounds from SP3 or RINEX-clock content."""
+    epochs: list[datetime.datetime] = []
+    with path.open(errors="replace") as stream:
+        for line in stream:
+            fields = line.split()
+            try:
+                if product == "ORBIT" and line.startswith("*"):
+                    values = fields[1:7]
+                elif product == "CLOCK" and fields[:1] in (["AS"], ["AR"]):
+                    values = fields[2:8]
+                else:
+                    continue
+                epochs.append(
+                    datetime.datetime(
+                        int(values[0]),
+                        int(values[1]),
+                        int(values[2]),
+                        int(values[3]),
+                        int(values[4]),
+                        int(float(values[5])),
+                    )
+                )
+            except (ValueError, IndexError):
+                continue
+    return (min(epochs), max(epochs)) if epochs else None
+
+
+def validate_pride_products(
+    rinex_paths: Iterable[Path],
+    resolution: DependencyResolution,
+    frequency_combinations: Iterable[str],
+) -> list[str]:
+    """Describe product coverage for the configured, observed constellations.
+
+    These checks are deliberately advisory.  Product coherence and successful
+    downloads determine bundle acceptance; imperfect coverage is reported so
+    callers can judge PPP-AR capability without losing a usable float solution.
+    """
+    rinex_paths = [Path(path) for path in rinex_paths]
+    messages: list[str] = []
+    observed = rinex_phase_observables(rinex_paths)
+    configured = {item[0]: set(item[1:]) for item in frequency_combinations if len(item) >= 2}
+    by_spec = {item.spec: Path(item.local_path) for item in resolution.fulfilled if item.local_path}
+
+    bias_path = by_spec.get("BIA")
+    if not bias_path or not bias_path.exists():
+        messages.append("BIA unavailable; phase-bias capability cannot be evaluated")
+        bias = None
+    else:
+        bias = bias_phase_observables(bias_path)
+    for system, required in configured.items():
+        present_observables = observed.get(system, set())
+        if not present_observables:
+            continue
+        if bias is None:
+            continue
+        present = {observable[1] for observable in present_observables}
+        relevant = required & present
+        supported_observables = bias.get(system, set()) & present_observables
+        supported = {observable[1] for observable in supported_observables}
+        missing = relevant - supported
+        if not relevant:
+            messages.append(
+                f"{system}: configured bands {sorted(required)} are not present in the RINEX"
+            )
+        elif missing:
+            messages.append(
+                f"{system}: BIA phase coverage is partial; missing bands {sorted(missing)} "
+                f"for observed signals {sorted(present_observables)} "
+                f"(matched {sorted(supported_observables)})"
+            )
+        else:
+            messages.append(f"{system}: BIA phase coverage supports bands {sorted(relevant)}")
+
+    starts_ends = [rinex_get_time_range(path) for path in rinex_paths]
+    obs_start = min(item[0] for item in starts_ends)
+    obs_end = max(item[1] for item in starts_ends)
+    for product in ("ORBIT", "CLOCK"):
+        path = by_spec.get(product)
+        if not path or not path.exists():
+            continue
+        bounds = product_epoch_bounds(path, product)
+        if bounds is None:
+            messages.append(f"{product}: epoch coverage could not be read")
+        elif bounds[0] > obs_start or bounds[1] < obs_end:
+            messages.append(
+                f"{product}: partial epoch coverage {bounds[0].isoformat()} to "
+                f"{bounds[1].isoformat()} for observations ending {obs_end.isoformat()}"
+            )
+        else:
+            messages.append(f"{product}: covers the observation interval")
+    return messages
diff --git a/packages/pride-ppp/src/pride_ppp/specifications/cli.py b/packages/pride-ppp/src/pride_ppp/specifications/cli.py
index 1493685..a50bcea 100644
--- a/packages/pride-ppp/src/pride_ppp/specifications/cli.py
+++ b/packages/pride-ppp/src/pride_ppp/specifications/cli.py
@@ -6,9 +6,13 @@
 
 from enum import Enum
 from pathlib import Path
+from typing import Literal
 
 from pydantic import BaseModel, Field
 
+PRIDE_NATIVE_FREQUENCIES = ("G12", "R12", "E15", "C26", "J12")
+DEFAULT_FREQUENCIES = ("G12", "R12", "E17", "C27", "J12")
+
 
 class Constellations(str, Enum):
     """GNSS constellation identifiers for pdp3 ``--system`` flag."""
@@ -80,12 +84,13 @@ class PrideCLIConfig(BaseModel):
 
     sample_frequency: float = 1
     system: str = "GREC23J"
-    frequency: list = ["G12", "R12", "E15", "C26", "J12"]
+    frequency: list[str] = Field(default_factory=lambda: list(DEFAULT_FREQUENCIES))
     loose_edit: bool = True
     cutoff_elevation: int = 7
     interval: float | None = None
     high_ion: bool | None = None
     tides: str = "SOP"
+    mapping_function: Literal["NIE", "GMF", "VM1", "VM3"] | None = None
 
     pride_configfile_path: Path | None = Field(
         None,
@@ -148,8 +153,12 @@ def generate_pdp_command(self, site: str, local_file_path: str) -> list[str]:
         if self.system != "GREC23J":
             command.extend(["--system", self.system])
 
-        if self.frequency != ["G12", "R12", "E15", "C26", "J12"]:
-            command.extend(["--frequency", " ".join(self.frequency)])
+        # Our E1/E5b and B1/B2 defaults intentionally differ from pdp3's
+        # native E1/E5a and B1/B3 defaults, so they must be passed explicitly.
+        if tuple(self.frequency) != PRIDE_NATIVE_FREQUENCIES:
+            # pdp3.sh consumes each three-character frequency combination as
+            # a separate argv item until it reaches the next option.
+            command.extend(["--frequency", *self.frequency])
 
         if self.loose_edit:
             command.append("--loose-edit")
@@ -166,6 +175,9 @@ def generate_pdp_command(self, site: str, local_file_path: str) -> list[str]:
         if self.tides != "SOP":
             command.extend(["--tide-off", self.tides])
 
+        if self.mapping_function:
+            command.extend(["--mapping-func", self.mapping_function])
+
         command.extend(["--site", site])
 
         if self.pride_configfile_path:
diff --git a/packages/pride-ppp/src/pride_ppp/specifications/config.py b/packages/pride-ppp/src/pride_ppp/specifications/config.py
index 4ef985f..3bbc83d 100644
--- a/packages/pride-ppp/src/pride_ppp/specifications/config.py
+++ b/packages/pride-ppp/src/pride_ppp/specifications/config.py
@@ -8,7 +8,7 @@
 from datetime import datetime
 from pathlib import Path
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field
 
 # ---------------------------------------------------------------------------
 # Default satellite table (all active GNSS PRNs, variance = 1)
@@ -211,7 +211,8 @@ class SatelliteProducts(BaseModel):
     erp : str, optional
         Earth rotation parameters filename (must end in ``.ERP``).
     quaternions : str, optional
-        Satellite attitude quaternions filename (must end in ``.OBX``).
+        Satellite attitude quaternions filename (must end in ``.OBX``), or
+        ``NONE`` to disable attitude corrections.
     code_phase_bias : str, optional
         Observable-specific signal bias filename (must end in ``.BIA``).
     leo_quaternions : str, optional
@@ -222,29 +223,38 @@ class SatelliteProducts(BaseModel):
         default="Default",
         description="Directory for satellite products",
     )
+    # Patterns accept the literal "Default" as well as a real filename: pdp3.sh
+    # compares this field's raw config-file text with `!= Default` to decide
+    # whether to resolve the product itself, so the sentinel has to survive
+    # into the file unchanged. A previous version of this pattern only
+    # accepted a real filename and relied on a validator to rewrite "Default"
+    # to e.g. "Default.OBX" to satisfy it — but that rewritten value then got
+    # written to the config file as-is, which pdp3.sh's strict `!= Default`
+    # check doesn't recognize, so it would try to fetch a file literally named
+    # "Default.OBX" instead of self-resolving.
     satellite_orbit: str | None = Field(
         default="Default",
-        pattern=r".*\.SP3",
+        pattern=r"^Default$|.*\.SP3",
         description="File name of SP3 file",
     )
     satellite_clock: str | None = Field(
         default="Default",
-        pattern=r".*\.CLK",
+        pattern=r"^Default$|.*\.CLK",
         description="File name of CLK file",
     )
     erp: str | None = Field(
         default="Default",
-        pattern=r".*\.ERP",
+        pattern=r"^Default$|.*\.ERP",
         description="File name of ERP file",
     )
     quaternions: str | None = Field(
         default="Default",
-        pattern=r".*\.OBX",
+        pattern=r"^(Default|NONE)$|.*\.OBX",
         description="File name of quaternions file",
     )
     code_phase_bias: str | None = Field(
         default="Default",
-        pattern=r".*\.BIA",
+        pattern=r"^Default$|.*\.BIA",
         description="File name of code/phase bias file",
     )
     leo_quaternions: str | None = Field(
@@ -252,32 +262,6 @@ class SatelliteProducts(BaseModel):
         description="File name of LEO quaternions file",
     )
 
-    @field_validator(
-        "satellite_orbit",
-        "satellite_clock",
-        "erp",
-        "quaternions",
-        "code_phase_bias",
-        mode="before",
-    )
-    def override_patternmatch(cls, value: str, field) -> str:
-        """Set default file extension when value is ``'Default'``."""
-        if value != "Default":
-            return value
-        match field.field_name:
-            case "satellite_orbit":
-                return "Default.SP3"
-            case "satellite_clock":
-                return "Default.CLK"
-            case "erp":
-                return "Default.ERP"
-            case "quaternions":
-                return "Default.OBX"
-            case "code_phase_bias":
-                return "Default.BIA"
-            case _:
-                return value
-
 
 class DataProcessingStrategies(BaseModel):
     """Data processing strategy defaults for the pdp3 config file.
diff --git a/packages/pride-ppp/tests/test_cli.py b/packages/pride-ppp/tests/test_cli.py
new file mode 100644
index 0000000..d1e2060
--- /dev/null
+++ b/packages/pride-ppp/tests/test_cli.py
@@ -0,0 +1,17 @@
+from pride_ppp.specifications.cli import PrideCLIConfig
+
+
+def test_frequency_combinations_are_separate_arguments() -> None:
+    config = PrideCLIConfig(frequency=["G12", "R12", "E17", "C27", "J12"])
+
+    command = config.generate_pdp_command("NTH1", "/tmp/nth1.rnx")
+
+    index = command.index("--frequency")
+    assert command[index + 1 : index + 6] == ["G12", "R12", "E17", "C27", "J12"]
+
+
+def test_mapping_function_is_passed_to_pdp3() -> None:
+    command = PrideCLIConfig(mapping_function="VM1").generate_pdp_command("NTH1", "/tmp/nth1.rnx")
+
+    index = command.index("--mapping-func")
+    assert command[index + 1] == "VM1"
diff --git a/packages/pride-ppp/tests/test_live_product_bundles.py b/packages/pride-ppp/tests/test_live_product_bundles.py
new file mode 100644
index 0000000..cc5a389
--- /dev/null
+++ b/packages/pride-ppp/tests/test_live_product_bundles.py
@@ -0,0 +1,103 @@
+"""Opt-in live tests for coherent near-real-time PRIDE product bundles."""
+
+from __future__ import annotations
+
+import datetime
+import re
+from pathlib import Path
+
+import pytest
+from pride_ppp.factories.processor import PrideProcessor
+
+_PRECISE_REQUIRED = ("ORBIT", "CLOCK", "ERP", "BIA")
+_FAMILY_RE = re.compile(r"([A-Z]{3})0([A-Z0-9]{3})(FIN|RTS|RAP|ULT)_")
+
+
+def _diagnostic_rinex(path: Path, day: datetime.date) -> Path:
+    """Write a header-only RINEX spanning the target UTC day."""
+    first = f"  {day.year:4d}    {day.month:2d}    {day.day:2d}     0     0    0.0000000     GPS"
+    last = f"  {day.year:4d}    {day.month:2d}    {day.day:2d}    23    59   30.0000000     GPS"
+    path.write_text(
+        "G    4 C1C L1C C2W L2W                              SYS / # / OBS TYPES\n"
+        "R    4 C1C L1C C2C L2C                              SYS / # / OBS TYPES\n"
+        "E    4 C1C L1C C7Q L7Q                              SYS / # / OBS TYPES\n"
+        "C    4 C2I L2I C7I L7I                              SYS / # / OBS TYPES\n"
+        "J    4 C1C L1C C2L L2L                              SYS / # / OBS TYPES\n"
+        f"{first:<60}TIME OF FIRST OBS\n"
+        f"{last:<60}TIME OF LAST OBS\n"
+        f"{'':60}END OF HEADER\n"
+    )
+    return path
+
+
+@pytest.fixture(scope="module")
+def live_processor(tmp_path_factory: pytest.TempPathFactory) -> PrideProcessor:
+    root = tmp_path_factory.mktemp("live-pride-products")
+    product_dir = root / "products"
+    output_dir = root / "output"
+    product_dir.mkdir()
+    output_dir.mkdir()
+    return PrideProcessor(
+        pride_dir=product_dir,
+        output_dir=output_dir,
+        override_products_download=True,
+    )
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("days_ago", [0, 1], ids=["current_utc_day", "previous_utc_day"])
+def test_live_coherent_product_bundle(
+    live_processor: PrideProcessor,
+    tmp_path: Path,
+    days_ago: int,
+) -> None:
+    """Download and inspect one coherent precise-product family."""
+    day = datetime.datetime.now(datetime.UTC).date() - datetime.timedelta(days=days_ago)
+    target = datetime.datetime.combine(day, datetime.time(), tzinfo=datetime.UTC)
+
+    resolution = live_processor._resolve(target)
+    rinex = _diagnostic_rinex(tmp_path / f"diagnostic-{day}.rnx", day)
+    live_processor._add_product_diagnostics([rinex], resolution)
+
+    by_spec = {result.spec: result for result in resolution.resolved}
+    missing = [name for name in _PRECISE_REQUIRED if by_spec[name].status == "missing"]
+    if missing and days_ago == 0:
+        pytest.skip(f"No complete same-day product family published yet: missing {missing}")
+    assert not missing, f"No complete precise-product bundle for {day}: missing {missing}"
+
+    families = set()
+    for name in _PRECISE_REQUIRED:
+        result = by_spec[name]
+        match = _FAMILY_RE.search(Path(result.remote_url or result.local_path or "").name)
+        assert match, f"Could not infer family from {name}: {result.remote_url}"
+        families.add(match.groups())
+    assert len(families) == 1, f"Mixed precise-product families selected: {families}"
+
+    print(f"\n{day}: selected family {next(iter(families))}")
+    for name in _PRECISE_REQUIRED:
+        result = by_spec[name]
+        print(f"  {name:<5} {result.status:<10} {result.remote_url or result.local_path}")
+    for message in resolution.diagnostics:
+        print(f"  diagnostic: {message}")
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("days_ago", [0, 1], ids=["current_utc_day", "previous_utc_day"])
+def test_live_wum_product_preflight(
+    live_processor: PrideProcessor,
+    days_ago: int,
+) -> None:
+    """Search and download using only Wuhan product sources."""
+    day = datetime.datetime.now(datetime.UTC).date() - datetime.timedelta(days=days_ago)
+    target = datetime.datetime.combine(day, datetime.time(), tzinfo=datetime.UTC)
+
+    resolution = live_processor._resolve(target, centers=["WUM"])
+    by_spec = {result.spec: result for result in resolution.resolved}
+
+    print(f"\nWUM-only result for {day}")
+    for name in _PRECISE_REQUIRED:
+        result = by_spec[name]
+        print(f"  {name:<5} {result.status:<10} {result.remote_url or result.local_path or '-'}")
+
+    missing = [name for name in _PRECISE_REQUIRED if by_spec[name].status == "missing"]
+    assert not missing, f"No complete WUM product bundle for {day}: missing {missing}"
diff --git a/packages/pride-ppp/tests/test_output_parsing.py b/packages/pride-ppp/tests/test_output_parsing.py
index d9d0e92..8ae619a 100644
--- a/packages/pride-ppp/tests/test_output_parsing.py
+++ b/packages/pride-ppp/tests/test_output_parsing.py
@@ -153,3 +153,16 @@ def test_wrms_values_in_mm_range(self, res_file: Path):
         """WRMS in mm — typical GNSS phase residuals are sub-centimetre."""
         df = get_wrms_from_res(res_file)
         assert (df["wrms"] < 1000).all(), "WRMS suspiciously large (> 1000 mm)"
+
+    def test_seconds_equal_sixty_roll_into_next_minute(self, tmp_path: Path):
+        res = tmp_path / "res_rollover.res"
+        res.write_text(
+            "Residuals COMMENT\n"
+            "TIM 2026 8 24 19 9 60.0000000 61276 69000.00\n"
+            "G01 0.0010 0.0000 1.0 1.0 0 45.0 90.0 L1C L2W C1C C2W\n"
+        )
+
+        df = get_wrms_from_res(res)
+
+        assert df.loc[0, "date"] == pd.Timestamp("2026-08-24T19:10:00Z")
+        assert df.loc[0, "wrms"] == pytest.approx(1.0)
diff --git a/packages/pride-ppp/tests/test_processor.py b/packages/pride-ppp/tests/test_processor.py
index ce24b33..58acfd6 100644
--- a/packages/pride-ppp/tests/test_processor.py
+++ b/packages/pride-ppp/tests/test_processor.py
@@ -12,26 +12,47 @@
 import os
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from unittest.mock import MagicMock
 
 import pytest
 
 try:
     from gnss_product_management.specifications.dependencies.dependencies import (
         DependencyResolution,
+        DependencySpec,
         ResolvedDependency,
     )
 except ImportError as e:
     pytest.skip(f"gnss-product-management not installed: {e}", allow_module_level=True)
 
+from pride_ppp.defaults import PRIDE_PPPAR_SPEC
 from pride_ppp.factories import processor as processor_module
 from pride_ppp.factories.processor import (
     MissingProductsError,
     PrideProcessor,
+    _prepare_rinex_for_product_coverage,
     _resolution_to_satellite_products,
     _resolution_to_table_dir,
+    _stage_broadcast_navigation,
 )
 
 
+def test_default_product_timeliness_prefers_rts_over_rap() -> None:
+    """RTS should win over RAP so near-real-time Galileo E17 has phase biases."""
+    spec = DependencySpec.from_yaml(PRIDE_PPPAR_SPEC)
+    timeliness = next(p for p in spec.preferences if p.parameter == "TTT")
+
+    assert timeliness.sorting == ["FIN", "RTS", "RAP", "ULT"]
+
+
+def test_default_navigation_allows_pdp3_hourly_fallback() -> None:
+    """Missing same-day mixed nav must not block PRIDE's GPS/GLO fallback."""
+    spec = DependencySpec.from_yaml(PRIDE_PPPAR_SPEC)
+    navigation = next(dep for dep in spec.dependencies if dep.spec == "RNX3_BRDC")
+
+    assert navigation.required is False
+
+
 @pytest.fixture
 def temp_product_files() -> dict[str, str]:
     """Create temporary product files and return their paths as strings."""
@@ -96,6 +117,7 @@ def test_resolution_to_satellite_products_with_string_local_path(
     assert satellite_products.satellite_orbit == Path(temp_product_files["orbit"]).name
     assert satellite_products.satellite_clock == Path(temp_product_files["clock"]).name
     assert satellite_products.code_phase_bias == Path(temp_product_files["bias"]).name
+    assert satellite_products.quaternions == "NONE"
     assert product_dir is not None
     assert product_dir == Path(temp_product_files["orbit"]).parent
 
@@ -116,10 +138,33 @@ def test_resolution_to_satellite_products_with_none_local_path() -> None:
     # Should not raise; returns empty products
     satellite_products, product_dir = _resolution_to_satellite_products(resolution)
 
-    assert satellite_products.satellite_orbit is None
+    assert satellite_products.satellite_orbit == "Default"
+    assert satellite_products.quaternions == "NONE"
     assert product_dir is None
 
 
+def test_resolution_to_satellite_products_keeps_resolved_attitude(
+    temp_product_files: dict[str, str],
+) -> None:
+    attitude = Path(temp_product_files["orbit"]).with_name("attitude.OBX")
+    attitude.write_text("attitude\n")
+    resolution = DependencyResolution(
+        spec_name="test",
+        resolved=[
+            ResolvedDependency(
+                spec="ATTOBX",
+                required=False,
+                status="local",
+                local_path=str(attitude),
+            )
+        ],
+    )
+
+    satellite_products, _ = _resolution_to_satellite_products(resolution)
+
+    assert satellite_products.quaternions == "attitude.OBX"
+
+
 def test_resolution_to_table_dir_with_string_local_path(
     temp_product_files: dict[str, str],
 ) -> None:
@@ -181,6 +226,161 @@ def test_resolution_to_table_dir_with_none_local_path() -> None:
     assert table_dir is None
 
 
+def test_stage_broadcast_navigation_uses_legacy_pdp3_name(tmp_path: Path) -> None:
+    source = tmp_path / "products" / "BRDC00IGS_R_20262450000_01D_MN.rnx"
+    source.parent.mkdir()
+    source.write_text("mixed navigation\n")
+    rinex = tmp_path / "observations" / "NCC100USA_R_20262450000_01D_02S_MO.rnx"
+    rinex.parent.mkdir()
+    rinex.write_text("observations\n")
+    resolution = DependencyResolution(
+        spec_name="test",
+        resolved=[
+            ResolvedDependency(
+                spec="RNX3_BRDC",
+                required=True,
+                status="downloaded",
+                local_path=str(source),
+            )
+        ],
+    )
+
+    staged = _stage_broadcast_navigation(resolution, rinex, datetime.date(2026, 9, 2))
+
+    assert staged == rinex.parent / "brdm2450.26p"
+    assert staged.read_text() == "mixed navigation\n"
+
+
+def test_current_day_without_mixed_nav_limits_pdp3_to_gps_glonass(tmp_path: Path) -> None:
+    proc = object.__new__(PrideProcessor)
+    proc._cli_config = processor_module.PrideCLIConfig()
+    today = datetime.datetime.now(datetime.timezone.utc).date()
+    resolution = DependencyResolution(
+        spec_name="test",
+        resolved=[
+            ResolvedDependency(
+                spec="RNX3_BRDC",
+                required=False,
+                status="missing",
+                local_path=None,
+            )
+        ],
+    )
+
+    command = proc._build_pdp_command(
+        tmp_path / "obs.rnx",
+        "NCC1",
+        tmp_path / "config_file",
+        resolution=resolution,
+        date=today,
+    )
+
+    assert command[command.index("--system") + 1] == "GR"
+    assert command[command.index("--mapping-func") + 1] == "GMF"
+    frequency_args = command[command.index("--frequency") + 1 : command.index("--loose-edit")]
+    assert frequency_args == ["G12", "R12"]
+
+
+def test_current_day_with_mixed_nav_keeps_configured_constellations(tmp_path: Path) -> None:
+    proc = object.__new__(PrideProcessor)
+    proc._cli_config = processor_module.PrideCLIConfig()
+    today = datetime.datetime.now(datetime.timezone.utc).date()
+    navigation = tmp_path / "BRDC00WRD_R_today_01D_MN.rnx"
+    navigation.write_text("navigation\n")
+    resolution = DependencyResolution(
+        spec_name="test",
+        resolved=[
+            ResolvedDependency(
+                spec="RNX3_BRDC",
+                required=False,
+                status="downloaded",
+                local_path=str(navigation),
+            )
+        ],
+    )
+
+    command = proc._build_pdp_command(
+        tmp_path / "obs.rnx",
+        "NCC1",
+        tmp_path / "config_file",
+        resolution=resolution,
+        date=today,
+    )
+
+    assert "--system" not in command
+    assert command[command.index("--mapping-func") + 1] == "GMF"
+    frequency_args = command[command.index("--frequency") + 1 : command.index("--loose-edit")]
+    assert frequency_args == ["G12", "R12", "E17", "C27", "J12"]
+
+
+def test_historical_day_keeps_requested_vmf1(tmp_path: Path) -> None:
+    proc = object.__new__(PrideProcessor)
+    proc._cli_config = processor_module.PrideCLIConfig(mapping_function="VM1")
+    resolution = DependencyResolution(spec_name="test", resolved=[])
+
+    command = proc._build_pdp_command(
+        tmp_path / "obs.rnx",
+        "NCC1",
+        tmp_path / "config_file",
+        resolution=resolution,
+        date=datetime.date(2025, 1, 1),
+    )
+
+    assert command[command.index("--mapping-func") + 1] == "VM1"
+
+
+def test_current_day_is_clipped_to_common_product_coverage(tmp_path: Path, monkeypatch) -> None:
+    proc = object.__new__(PrideProcessor)
+    proc._cli_config = processor_module.PrideCLIConfig()
+    today = datetime.datetime.now(datetime.timezone.utc).date()
+    rinex = tmp_path / "obs.rnx"
+    rinex.write_text(
+        "                                                            END OF HEADER\n"
+        "> 2026 09 04 00 00 00.0000000  0  1\nG01 observations\n"
+        "> 2026 09 04 09 55 00.0000000  0  1\nG01 observations\n"
+        "> 2026 09 04 10 00 00.0000000  0  1\nG01 observations\n"
+    )
+    orbit = tmp_path / "orbit.SP3"
+    clock = tmp_path / "clock.CLK"
+    orbit.write_text("orbit\n")
+    clock.write_text("clock\n")
+    resolution = DependencyResolution(
+        spec_name="test",
+        resolved=[
+            ResolvedDependency(
+                spec="ORBIT", required=True, status="downloaded", local_path=str(orbit)
+            ),
+            ResolvedDependency(
+                spec="CLOCK", required=True, status="downloaded", local_path=str(clock)
+            ),
+        ],
+    )
+    product_start = datetime.datetime.combine(today, datetime.time(0, 0))
+    orbit_end = datetime.datetime.combine(today, datetime.time(10, 0))
+    clock_end = datetime.datetime.combine(today, datetime.time(10, 4, 30))
+    observation_end = datetime.datetime.combine(today, datetime.time(11, 30))
+    monkeypatch.setattr(
+        processor_module,
+        "product_epoch_bounds",
+        lambda path, product: (
+            product_start,
+            orbit_end if product == "ORBIT" else clock_end,
+        ),
+    )
+    monkeypatch.setattr(
+        processor_module,
+        "rinex_get_time_range",
+        lambda path: (product_start, observation_end),
+    )
+
+    clipped = _prepare_rinex_for_product_coverage(rinex, resolution, today, tmp_path / "work")
+
+    assert clipped != rinex
+    text = clipped.read_text()
+    assert "09 55 00" in text
+    assert "10 00 00" not in text
+
+
 @pytest.fixture
 def processor() -> PrideProcessor:
     """A PrideProcessor without running __init__ — _validate_kinfile is self-free."""
@@ -213,6 +413,24 @@ def test_unparseable_kinfile_returns_false(
         assert processor._validate_kinfile(garbage) is False
 
 
+def test_resolve_forwards_product_download_override(processor: PrideProcessor) -> None:
+    processor._client = MagicMock()
+    processor._dep_spec = MagicMock()
+    processor._override_products_download = True
+    expected = MagicMock()
+    processor._client.resolve_dependencies.return_value = (expected, None)
+
+    result = processor._resolve(datetime.datetime(2026, 8, 27, tzinfo=datetime.timezone.utc))
+
+    assert result is expected
+    processor._client.resolve_dependencies.assert_called_once_with(
+        processor._dep_spec,
+        datetime.datetime(2026, 8, 27, tzinfo=datetime.timezone.utc),
+        sink_id="pride",
+        force_download=True,
+    )
+
+
 class TestRunPdp3:
     """Subprocess handling in _run_pdp3, exercised via a fake pdp3 on PATH."""
 
@@ -230,8 +448,13 @@ def install(script_body: str) -> None:
 
         return install
 
-    def test_outputs_moved_with_extensions(self, fake_pdp3, tmp_path: Path) -> None:
-        fake_pdp3("touch kin_2025254_ncc1 res_2025254_ncc1")
+    def test_outputs_moved_with_extensions(self, fake_pdp3, tmp_path: Path, monkeypatch) -> None:
+        fake_pdp3(
+            "touch kin_2025254_ncc1 res_2025254_ncc1 config.runtime "
+            "log_2025254_ncc1 cst_2025254_ncc1 stt_2025254_ncc1; "
+            "echo run-output"
+        )
+        monkeypatch.setattr(processor_module, "kin_to_kin_position_df", lambda path: MagicMock())
         out = tmp_path / "out"
 
         kin, res, rc, _ = PrideProcessor._run_pdp3(command=["pdp3"], site="NCC1", output_dir=out)
@@ -239,6 +462,11 @@ def test_outputs_moved_with_extensions(self, fake_pdp3, tmp_path: Path) -> None:
         assert rc == 0
         assert kin == out / "kin_2025254_ncc1.kin" and kin.exists()
         assert res == out / "res_2025254_ncc1.res" and res.exists()
+        assert (out / "config_2025254_ncc1.config").exists()
+        assert (out / "log_2025254_ncc1.log").exists()
+        assert (out / "cst_2025254_ncc1.cst").exists()
+        assert (out / "stt_2025254_ncc1.stt").exists()
+        assert "run-output" in (out / "run_2025254_ncc1.log").read_text()
 
     def test_nonzero_exit_and_missing_output_are_logged(
         self, fake_pdp3, tmp_path: Path, caplog
@@ -257,6 +485,22 @@ def test_nonzero_exit_and_missing_output_are_logged(
         assert any("pdp3 exited with code 2" in m for m in caplog.messages)
         assert any("produced no kin output" in m for m in caplog.messages)
 
+    def test_non_utf8_process_output_does_not_abort_job(
+        self, fake_pdp3, tmp_path: Path, caplog
+    ) -> None:
+        fake_pdp3("printf '\\200bad output\\n'; exit 2")
+        out = tmp_path / "out"
+
+        with caplog.at_level(logging.INFO, logger="pride_ppp.factories.processor"):
+            kin, res, rc, stderr = PrideProcessor._run_pdp3(
+                command=["pdp3"], site="NCC1", output_dir=out
+            )
+
+        assert rc == 2
+        assert kin is None and res is None
+        assert stderr == ""
+        assert any("bad output" in message for message in caplog.messages)
+
 
 def _unfulfilled_resolution() -> DependencyResolution:
     return DependencyResolution(
diff --git a/packages/pride-ppp/tests/test_product_validation.py b/packages/pride-ppp/tests/test_product_validation.py
new file mode 100644
index 0000000..1f5b284
--- /dev/null
+++ b/packages/pride-ppp/tests/test_product_validation.py
@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from gnss_product_management.specifications.dependencies.dependencies import (
+    DependencyResolution,
+    ResolvedDependency,
+)
+from pride_ppp.factories.product_validation import (
+    bias_phase_bands,
+    bias_phase_observables,
+    rinex_phase_bands,
+    rinex_phase_observables,
+    validate_pride_products,
+)
+
+
+def _write(path: Path, text: str) -> Path:
+    path.write_text(text)
+    return path
+
+
+def test_phase_band_parsers_cover_all_observed_constellations(tmp_path: Path) -> None:
+    rinex = _write(
+        tmp_path / "obs.rnx",
+        "G    4 C1C L1C C2W L2W                              SYS / # / OBS TYPES\n"
+        "E    6 C1C L1C C7Q L7Q C5Q                          SYS / # / OBS TYPES\n"
+        "       L5Q                                              SYS / # / OBS TYPES\n"
+        "                                                            END OF HEADER\n",
+    )
+    bias = _write(
+        tmp_path / "test.BIA",
+        "+BIAS/SOLUTION\n"
+        " OSB  G01           L1C  2026:001:00000 2026:002:00000 ns 0.0\n"
+        " OSB  G01           L2W  2026:001:00000 2026:002:00000 ns 0.0\n"
+        " OSB  E01           L1C  2026:001:00000 2026:002:00000 ns 0.0\n"
+        "-BIAS/SOLUTION\n",
+    )
+
+    assert rinex_phase_bands([rinex]) == {"G": {"1", "2"}, "E": {"1", "5", "7"}}
+    assert bias_phase_bands(bias) == {"G": {"1", "2"}, "E": {"1"}}
+    assert rinex_phase_observables([rinex])["E"] == {"L1C", "L5Q", "L7Q"}
+    assert bias_phase_observables(bias)["E"] == {"L1C"}
+
+
+def test_validation_is_advisory_and_reports_partial_coverage(tmp_path: Path) -> None:
+    rinex = _write(
+        tmp_path / "obs.rnx",
+        "E    4 C1C L1C C7Q L7Q                              SYS / # / OBS TYPES\n"
+        "  2026     8    18     0     0    0.0000000     GPS         TIME OF FIRST OBS\n"
+        "  2026     8    18    23    59   30.0000000     GPS         TIME OF LAST OBS\n"
+        "                                                            END OF HEADER\n",
+    )
+    bias = _write(
+        tmp_path / "test.BIA",
+        " OSB  E01           L1C  2026:230:00000 2026:231:00000 ns 0.0\n",
+    )
+    orbit = _write(
+        tmp_path / "test.SP3",
+        "*  2026  8 18  1  0  0.00000000\n*  2026  8 18 23  0  0.00000000\n",
+    )
+    resolution = DependencyResolution(
+        spec_name="test",
+        resolved=[
+            ResolvedDependency(
+                spec="BIA", required=True, status="downloaded", local_path=str(bias)
+            ),
+            ResolvedDependency(
+                spec="ORBIT", required=True, status="downloaded", local_path=str(orbit)
+            ),
+        ],
+    )
+
+    messages = validate_pride_products([rinex], resolution, ["E17"])
+
+    assert any("missing bands ['7']" in message for message in messages)
+    assert any("ORBIT: partial epoch coverage" in message for message in messages)
+    assert resolution.all_required_fulfilled
+
+
+def test_missing_bias_has_one_clear_diagnostic(tmp_path: Path) -> None:
+    rinex = _write(
+        tmp_path / "obs.rnx",
+        "E    4 C1C L1C C7Q L7Q                              SYS / # / OBS TYPES\n"
+        "  2026     8    18     0     0    0.0000000     GPS         TIME OF FIRST OBS\n"
+        "  2026     8    18     1     0    0.0000000     GPS         TIME OF LAST OBS\n"
+        "                                                            END OF HEADER\n",
+    )
+    resolution = DependencyResolution(spec_name="test", resolved=[])
+
+    messages = validate_pride_products([rinex], resolution, ["E17"])
+
+    assert messages == ["BIA unavailable; phase-bias capability cannot be evaluated"]