Thanks for the detailed context — that helps a lot. What’s happening - In Prefect 3.x, block types from integrations (like prefect-gcp) are registered via Python entry points loaded per process when Prefect first needs them. - In your deployment run, Prefect is trying to resolve the default task result storage from settings inside the Ray worker process: - hydrated_context → Block.new → get_block_class_from_key('gcs-bucket') - That lookup triggers plugin discovery in that Ray worker process. - The UnknownBlockType means that, in that Ray worker process, plugin discovery didn’t find/register the

prefect_gcp

blocks, even though the package may be importable in a shell. Why it only fails in deployments - In your ad‑hoc/local run, you don’t have task result persistence enabled, so the Ray worker never tries to resolve the default result storage → no lookup → no failure. - In the deployment, you set PREFECT_TASKS_DEFAULT_PERSIST_RESULT=true and PREFECT_DEFAULT_RESULT_STORAGE_BLOCK=gcs-bucket/batch-results on the work‑pool job. Prefect snapshots these settings and applies them in the Ray worker process during task startup — which forces the lookup and fails if plugins aren’t registered in that process. Most likely root cause - The Ray worker process environment is not seeing the

prefect.collections

entry point for

prefect_gcp

(even if

import prefect_gcp

works in an interactive shell), usually because: - The package is not installed in the exact Python environment the Ray worker processes use (KubeRay often runs workers in a conda env different from your kubectl exec shell). - Long‑lived Ray workers did not pick up runtime_env["pip"] installs (they won’t retroactively install packages on existing worker processes). - Entry points metadata isn’t discoverable from that interpreter (importlib.metadata.entry_points does not include

prefect_gcp

). Quickest workarounds - Prefer an explicit result storage object over the env var so the engine doesn’t have to resolve by slug:

from prefect import flow
  from prefect_gcp.cloud_storage import GcsBucket

  results = GcsBucket.load("batch-results")

  @flow(result_storage=results)
  def my_flow(...):
      ...

This avoids get_block_class_from_key('gcs-bucket') in the worker. Note: - You must have

prefect_gcp

installed in both the work‑pool image and Ray workers (for unpickling/usage), but this bypasses the entry point slug resolution path that’s failing. - Force-load the collection in the Ray worker before any task starts (works even if entry points aren’t visible, as long as the package is installed):

from prefect.plugins import load_prefect_collections
  import prefect_gcp  # ensure module import runs block registration
  load_prefect_collections()

You can put this at module import time in your flow file so it’s included in the Ray task payload and runs as soon as the worker imports your module. Recommended durable fix (KubeRay) - Bake

prefect-gcp==0.6.16

(and

prefect==3.6.1

,

prefect-ray==0.4.5

) into the Ray worker image used by KubeRay, in the same interpreter/env the workers use. For typical KubeRay images that’s the

ray

conda env:

# Example Dockerfile fragment
  RUN pip install "prefect==3.6.1" "prefect-ray==0.4.5" "prefect-gcp==0.6.16"

- If you were relying on

runtime_env["pip"]

, note that long‑lived workers won’t be updated. You’d need to: - Restart the Ray cluster to recreate workers with the new env, or - Build an image with the deps preinstalled. Diagnostics to confirm the mismatch (run on a Ray worker process) - Add a tiny Prefect task that runs on Ray and prints what the worker sees: ``` import sys, importlib.metadata as im from prefect import task @task def ray_env_probe(): print("python:", sys.executable) print("sys.path[03]", sys.path[0:3]) eps = [ep.name for ep in im.entry_points(group="prefect.collections")]

print("prefect.collections entry_points:", eps) try: from prefect.plugins import load_prefect_collections load_prefect_collections() from prefect.blocks.core import Block print("gcs-bucket class:", Block.get_block_class_from_key("gcs-bucket")) except Exception as e: print("lookup error:", repr(e)) ``` Call this once early in your flow to capture logs from the Ray worker. If

prefect_gcp

is missing from the entry_points list, that confirms the entry‐point discovery gap. Helpful logging - Set in Ray worker pods to see plugin load attempts: -

PREFECT_LOGGING_LEVEL=DEBUG

or

PREFECT_DEBUG_MODE=1

You should see messages about loading Prefect collections. Why your current checks can pass but tasks still fail - A “fresh kubectl exec shell” may not be the exact Python environment that the Ray worker processes use (different conda env/interpreter). Entry points are tied to the interpreter that installed the wheel. That’s why the shell can see

prefect_gcp

and even list entry points, while the Ray worker process can’t. References - Prefect plugin loading (source): prefect/plugins.py - Blocks overview: Blocks - Ray integration: Prefect + Ray If you can share: - The Ray image you’re using for workers (and how

prefect

and

prefect-gcp

are installed) - Output from the ray_env_probe task above during a failing deployment …I can help pinpoint exactly where the environment diverges and suggest the minimal change.