Pre-warm the IWC workflow manifest cache on a celery beat schedule. The new

refresh_manifest helper in iwc.py is a force-fetch sibling of fetch_manifest
that holds the cache lock across the network call, so concurrent on-demand
readers see either the previous value or the new one but never an empty cache
mid-write. The celery task wraps it with the periodic-task error contract
(swallow failures, retain prior copy). Registration uses the same
inference_services gate as the GTN refresh -- IWC pre-warm only matters when
the agent-ops layer is exposed.

lib/galaxy/agents/iwc.py

@@ -53,6 +53,27 @@ def fetch_manifest(timeout: float = 30.0) -> list[dict[str, Any]]:
         return manifest
 
 
+def refresh_manifest(timeout: float = 30.0) -> list[dict[str, Any]]:
+    """Force-fetch the manifest and replace the cached entry.
+
+    Used by the celery-beat pre-warm task. The lock is held across the
+    network fetch so a concurrent on-demand call from a request handler
+    sees either the previous value or the new one -- never an empty cache
+    mid-write. Kept as a separate function from ``fetch_manifest`` because
+    the error contracts differ: lazy fetch propagates (caller can't
+    continue without the data); this one is called from a periodic task
+    that has to tolerate transient failure.
+    """
+    with _manifest_cache_lock:
+        response = requests.get(IWC_MANIFEST_URL, timeout=timeout)
+        response.raise_for_status()
+        manifest = response.json()
+        if not isinstance(manifest, list):
+            raise ValueError(f"IWC manifest at {IWC_MANIFEST_URL} did not return a JSON array")
+        _manifest_cache[_CACHE_KEY] = manifest
+        return manifest
+
+
 def all_workflows(manifest: list[dict[str, Any]]) -> list[dict[str, Any]]:
     """Flatten the manifest into a single list of workflow entries."""
     workflows: list[dict[str, Any]] = []

lib/galaxy/celery/__init__.py

@@ -294,6 +294,11 @@ def setup_periodic_tasks(config, celery_app):
     ):
         schedule_task("refresh_gtn_database", config.gtn_database_refresh_interval)
 
+    # IWC manifest pre-warm only matters when the agent-ops layer is exposed --
+    # same inference_services gate as the GTN refresh above.
+    if getattr(config, "inference_services", None) and config.iwc_manifest_refresh_interval:
+        schedule_task("refresh_iwc_manifest", config.iwc_manifest_refresh_interval)
+
     if config.celery_user_concurrency_limit:
         # Run cleanup every 5 minutes (300 seconds)
         schedule_task("cleanup_stale_concurrency_slots", 300)

lib/galaxy/celery/tasks.py

@@ -830,6 +830,29 @@ def renew_vault_token(vault: Vault):
     renew_vault_token_if_needed(vault)
 
 
+@galaxy_task(action="refreshing IWC workflow manifest cache")
+def refresh_iwc_manifest(config: GalaxyAppConfiguration):
+    """Pre-warm the in-process IWC manifest cache.
+
+    The agent-ops layer caches the manifest at module scope with an hour
+    TTL; without this task the first user-driven IWC call after a worker
+    restart pays the full network fetch. Failures are logged and swallowed
+    so an iwc.galaxyproject.org outage doesn't kill the periodic queue --
+    on-demand callers still get the prior cached copy until the TTL lapses.
+    """
+    # Local import keeps the iwc module (and its cachetools / requests
+    # imports) out of celery's startup graph -- they only load on the
+    # workers that actually run this task.
+    from galaxy.agents import iwc
+
+    try:
+        manifest = iwc.refresh_manifest()
+    except Exception as e:  # noqa: BLE001 -- best-effort warm; resilience over precision
+        log.warning("refresh_iwc_manifest: fetch failed, keeping existing cache: %s", e)
+        return
+    log.info("refresh_iwc_manifest: cached %s top-level manifest entries", len(manifest))
+
+
 @galaxy_task(action="refreshing GTN training database")
 def refresh_gtn_database(config: GalaxyAppConfiguration):
     """HEAD depot for the GTN search database and re-download only when newer.

test/unit/agents/test_iwc.py

@@ -48,6 +48,47 @@ def test_fetch_manifest_caches_response():
     assert mock_get.call_count == 1
 
 
+def test_refresh_manifest_replaces_cached_value():
+    with patch("galaxy.agents.iwc.requests.get") as mock_get:
+        mock_get.return_value.json.return_value = SAMPLE_MANIFEST
+        mock_get.return_value.raise_for_status.return_value = None
+
+        first = iwc.refresh_manifest()
+        assert first == SAMPLE_MANIFEST
+
+        new_manifest = [{"workflows": [{"trsID": "#workflow/x/y/main"}]}]
+        mock_get.return_value.json.return_value = new_manifest
+
+        second = iwc.refresh_manifest()
+        assert second == new_manifest
+        # And the next on-demand fetch sees the refreshed value
+        assert iwc.fetch_manifest() == new_manifest
+
+
+def test_refresh_manifest_failure_leaves_prior_cache():
+    with patch("galaxy.agents.iwc.requests.get") as mock_get:
+        mock_get.return_value.json.return_value = SAMPLE_MANIFEST
+        mock_get.return_value.raise_for_status.return_value = None
+        iwc.fetch_manifest()  # prime the cache
+
+        mock_get.side_effect = RuntimeError("boom")
+        with pytest.raises(RuntimeError):
+            iwc.refresh_manifest()
+
+        # fetch_manifest still returns the previously-cached value
+        mock_get.side_effect = None
+        assert iwc.fetch_manifest() == SAMPLE_MANIFEST
+
+
+def test_refresh_manifest_rejects_non_list_payload():
+    with patch("galaxy.agents.iwc.requests.get") as mock_get:
+        mock_get.return_value.json.return_value = {"not": "a list"}
+        mock_get.return_value.raise_for_status.return_value = None
+
+        with pytest.raises(ValueError, match="did not return a JSON array"):
+            iwc.refresh_manifest()
+
+
 def test_clean_readme_summary_strips_headers_and_truncates():
     body = "First line that has plenty of content. Second line continues the thought. "
     readme = "# Heading\n\n" + (body * 10)

test/unit/app/test_celery.py

@@ -27,9 +27,10 @@ def test_default_configuration():
         "task": "galaxy.cleanup_short_term_storage",
         "schedule": galaxy_conf.short_term_storage_cleanup_interval,
     }
-    # GTN refresh is gated on inference_services being configured; default
-    # config doesn't set it, so the schedule isn't registered here.
+    # GTN and IWC refreshes are gated on inference_services being configured;
+    # default config doesn't set it, so neither schedule is registered here.
     assert "refresh-gtn-database" not in conf.beat_schedule
+    assert "refresh-iwc-manifest" not in conf.beat_schedule
 
 
 def test_gtn_refresh_schedules_when_inference_configured():
@@ -43,6 +44,17 @@ def test_gtn_refresh_schedules_when_inference_configured():
     }
 
 
+def test_iwc_refresh_schedules_when_inference_configured():
+    config = GalaxyAppConfiguration(override_tempdir=False)
+    config.inference_services = {"default": {"model": "test"}}
+    app = GalaxyCelery("test-iwc-schedule")
+    setup_periodic_tasks(config, app)
+    assert app.conf.beat_schedule["refresh-iwc-manifest"] == {
+        "task": "galaxy.refresh_iwc_manifest",
+        "schedule": config.iwc_manifest_refresh_interval,
+    }
+
+
 def test_galaxycelery_trim_module_name():
     gc = GalaxyCelery()
     assert gc.trim_module_name("notgalaxy.celery.tasks") == "notgalaxy.celery.tasks"