diff --git a/src/aind_torch_utils/config.py b/src/aind_torch_utils/config.py
index 06bb930..725a004 100644
--- a/src/aind_torch_utils/config.py
+++ b/src/aind_torch_utils/config.py
@@ -4,6 +4,14 @@
 from pydantic import BaseModel, Field, model_validator
 
 
+CompileMode = Literal[
+    "default",
+    "reduce-overhead",
+    "max-autotune",
+    "max-autotune-no-cudagraphs",
+]
+
+
 class InferenceConfig(BaseModel):
     # Geometry
     patch: Tuple[int, int, int] = Field(
@@ -32,13 +40,20 @@ class InferenceConfig(BaseModel):
         default=False, description="Enable cuDNN benchmarking"
     )
     use_compile: bool = Field(default=False, description="Use torch.compile")
-    compile_mode: str = Field(
-        default="reduce-overhead",
-        description="Torch.compile mode",  # or "max-autotune" if you want extra tuning time
+    compile_mode: CompileMode = Field(
+        default="default",
+        description="Torch.compile mode",
+    )
+    compile_dynamic: Optional[bool] = Field(
+        default=None,
+        description=(
+            "torch.compile dynamic shapes: None (auto) compiles static and "
+            "promotes to dynamic on a shape change; True forces dynamic; "
+            "False recompiles per shape. Input shapes are constant (tail "
+            "batches are padded), so auto yields one static-specialized "
+            "graph."
+        ),
     )
-    compile_dynamic: bool = Field(
-        default=True, description="Torch.compile with dynamic shapes"
-    )  # tolerate last-batch size changes
 
     # Concurrency / queues
     max_inflight_batches: int = Field(default=64, description="Max in-flight batches")
@@ -224,5 +239,25 @@ def _validate(self):
         # Devices
         if not self.devices:
             raise ValueError("devices list must not be empty")
+        cuda_device_count = sum(
+            str(device).lower().startswith("cuda") for device in self.devices
+        )
+        if self.use_compile and cuda_device_count > 1:
+            safe_compile_modes = {
+                "reduce-overhead": "default",
+                "max-autotune": "max-autotune-no-cudagraphs",
+            }
+            safe_mode = safe_compile_modes.get(self.compile_mode)
+            if safe_mode:
+                warnings.warn(
+                    (
+                        f"torch.compile mode '{self.compile_mode}' enables "
+                        "CUDA graphs, "
+                        "which can fail in threaded multi-GPU runs; using "
+                        f"'{safe_mode}' instead."
+                    ),
+                    RuntimeWarning,
+                )
+                self.compile_mode = safe_mode
 
         return self
diff --git a/src/aind_torch_utils/run.py b/src/aind_torch_utils/run.py
index c626f52..7c93146 100644
--- a/src/aind_torch_utils/run.py
+++ b/src/aind_torch_utils/run.py
@@ -484,14 +484,25 @@ def _parse_args(argv: List[str]) -> argparse.Namespace:
     ap.add_argument(
         "--compile-mode",
         type=str,
-        default="reduce-overhead",
-        choices=["reduce-overhead", "max-autotune"],
+        default="default",
+        choices=[
+            "default",
+            "reduce-overhead",
+            "max-autotune",
+            "max-autotune-no-cudagraphs",
+        ],
         help="torch.compile mode",
     )
     ap.add_argument(
-        "--no-compile-dynamic",
-        action="store_true",
-        help="Disable dynamic shape support for torch.compile",
+        "--compile-dynamic",
+        type=str,
+        default="auto",
+        choices=["auto", "true", "false"],
+        help=(
+            "torch.compile dynamic shapes: 'auto' compiles static and "
+            "promotes to dynamic if a shape changes; 'true' forces dynamic; "
+            "'false' recompiles per shape"
+        ),
     )
     ap.add_argument(
         "--max-inflight-batches",
@@ -627,7 +638,9 @@ def main(argv: Optional[List[str]] = None) -> None:
         cudnn_benchmark=args.cudnn_benchmark,
         use_compile=args.compile,
         compile_mode=args.compile_mode,
-        compile_dynamic=not args.no_compile_dynamic,
+        compile_dynamic={"auto": None, "true": True, "false": False}[
+            args.compile_dynamic
+        ],
         max_inflight_batches=args.max_inflight_batches,
         seam_mode=args.seam_mode,
         trim_voxels=args.trim_voxels,
diff --git a/src/aind_torch_utils/workers.py b/src/aind_torch_utils/workers.py
index d7acdbe..44cc9ae 100644
--- a/src/aind_torch_utils/workers.py
+++ b/src/aind_torch_utils/workers.py
@@ -34,7 +34,11 @@ class Batch:
         List of (z, y, x) start coordinates for each patch in the batch,
         relative to the expanded block.
     host_in : torch.Tensor
-        The input tensor of patches, pinned to host memory.
+        The input tensor of patches, pinned to host memory. When compiling
+        (cfg.use_compile), tail batches are zero-padded up to batch_size so
+        the model sees a constant input shape, and rows beyond
+        len(starts_in_block) are padding. In eager mode it has exactly
+        len(starts_in_block) rows.
     valid_sizes : List[Tuple[int, int, int]]
         List of (dz, dy, dx) valid dimensions for each patch, handling
         boundary conditions.
@@ -260,10 +264,18 @@ def run(self, stop_event: threading.Event) -> None:
             # batch over those starts
             for i in range(0, total_patches, self.cfg.batch_size):
                 batch_starts = starts[i : i + self.cfg.batch_size]
-                B = len(batch_starts)
+                n_real = len(batch_starts)
                 pin_memory = any("cuda" in d for d in self.cfg.devices)
+                # When compiling, pad the tail batch up to batch_size so the
+                # model always sees a constant input shape; this prevents
+                # torch.compile from recompiling at runtime (not thread-safe
+                # across GPU workers). Writers ignore padded rows since they
+                # only index rows in starts_in_block. In eager mode there is
+                # no shape constraint, so allocate exactly n_real rows and
+                # avoid wasting compute and copy bandwidth on padding.
+                n_rows = self.cfg.batch_size if self.cfg.use_compile else n_real
                 host_in = torch.zeros(
-                    (B, 1, pz, py, px),
+                    (n_rows, 1, pz, py, px),
                     dtype=torch.float16 if self.cfg.amp else torch.float32,
                     pin_memory=pin_memory,
                 )
@@ -351,18 +363,68 @@ def __init__(
         self.copy_stream = torch.cuda.Stream(device=self.device)
 
         if getattr(torch, "compile", None) and self.cfg.use_compile:
+            self._compile_model()
+
+    def _autocast_context(self):
+        return (
+            torch.autocast(device_type="cuda", dtype=torch.float16)
+            if self.cfg.amp
+            else nullcontext()
+        )
+
+    def _compile_model(self) -> None:
+        # Keep a handle to the original module so we can fall back to eager
+        # execution if compilation fails. torch.compile returns a new wrapper
+        # and does not mutate the original, so this reference stays valid.
+        eager_model = self.model
+        try:
             try:
-                # dynamic=True avoids recompiles when the final batch is smaller
+                # PrepWorker pads tail batches to batch_size, so input shapes
+                # are constant and no runtime recompiles are expected
+                # regardless of the `dynamic` setting.
                 self.model = torch.compile(
                     self.model,
                     mode=self.cfg.compile_mode,
                     dynamic=self.cfg.compile_dynamic,
                 )
-                logger.info("Successfully compiled model.")
+                logger.info("Compiled model on %s.", self.device)
             except TypeError:
                 # older PyTorch without `dynamic` kwarg
                 self.model = torch.compile(self.model, mode=self.cfg.compile_mode)
-                logger.info("Successfully compiled model (older pytorch).")
+                logger.info("Compiled model on %s (older pytorch).", self.device)
+
+            # Compilation is lazy: the graph is traced on the first forward,
+            # so tracing/guard errors surface here in warmup, not above.
+            self._warmup_compiled_model()
+        except Exception as exc:
+            # Some models do host-side numpy/Python work in forward that
+            # dynamo cannot trace. Fall back to eager so the run proceeds
+            # instead of aborting. Warmup runs on the main thread, so this
+            # also keeps the failure off the worker threads.
+            logger.warning(
+                "torch.compile failed on %s (%s); falling back to eager "
+                "execution.",
+                self.device,
+                type(exc).__name__,
+                exc_info=True,
+            )
+            self.model = eager_model
+
+    def _warmup_compiled_model(self) -> None:
+        torch.cuda.set_device(self.device)
+        dtype = torch.float16 if self.cfg.amp else torch.float32
+        shape = (self.cfg.batch_size, 1, *self.cfg.patch)
+        warmup_in = torch.zeros(shape, dtype=dtype, device=self.device)
+
+        logger.info(
+            "Warming compiled model on %s with shape %s.", self.device, shape
+        )
+        with torch.inference_mode():
+            with self._autocast_context():
+                warmup_out = self.model(warmup_in)
+        torch.cuda.synchronize(self.device)
+        del warmup_in, warmup_out
+        logger.info("Finished compiled model warmup on %s.", self.device)
 
     def run(self, stop_event: threading.Event) -> None:
         """
@@ -376,11 +438,7 @@ def run(self, stop_event: threading.Event) -> None:
         stop_event : threading.Event
             An event that signals the worker to stop.
         """
-        autocast_ctx = (
-            torch.autocast(device_type="cuda", dtype=torch.float16)
-            if self.cfg.amp
-            else nullcontext()
-        )
+        autocast_ctx = self._autocast_context()
 
         # Ensure the current device matches self.device for streams/events
         torch.cuda.set_device(self.device)
diff --git a/tests/test_config.py b/tests/test_config.py
index 9839829..d71b7f7 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -13,6 +13,8 @@ def test_precision_performance_defaults_to_false():
 
     assert cfg.use_tf32 is False
     assert cfg.cudnn_benchmark is False
+    assert cfg.compile_mode == "default"
+    assert cfg.compile_dynamic is None
 
 
 @pytest.mark.parametrize("field", ["use_tf32", "cudnn_benchmark"])
@@ -26,3 +28,41 @@ def test_precision_performance_config_overrides(field):
     )
 
     assert getattr(cfg, field) is True
+
+
+@pytest.mark.parametrize(
+    ("requested_mode", "expected_mode"),
+    [
+        ("reduce-overhead", "default"),
+        ("max-autotune", "max-autotune-no-cudagraphs"),
+    ],
+)
+def test_compile_cudagraph_modes_are_downgraded_for_multi_cuda(
+    requested_mode, expected_mode
+):
+    with pytest.warns(RuntimeWarning, match="threaded multi-GPU"):
+        cfg = InferenceConfig(
+            patch=(16, 16, 16),
+            overlap=4,
+            trim_voxels=2,
+            block=(32, 32, 32),
+            devices=["cuda:0", "cuda:1"],
+            use_compile=True,
+            compile_mode=requested_mode,
+        )
+
+    assert cfg.compile_mode == expected_mode
+
+
+def test_compile_cudagraph_mode_allowed_for_single_cuda():
+    cfg = InferenceConfig(
+        patch=(16, 16, 16),
+        overlap=4,
+        trim_voxels=2,
+        block=(32, 32, 32),
+        devices=["cuda:0"],
+        use_compile=True,
+        compile_mode="reduce-overhead",
+    )
+
+    assert cfg.compile_mode == "reduce-overhead"
diff --git a/tests/test_workers.py b/tests/test_workers.py
index f45a4f9..fb63a45 100644
--- a/tests/test_workers.py
+++ b/tests/test_workers.py
@@ -2,11 +2,103 @@
 import threading
 from typing import Optional
 
+import numpy as np
 import pytest
+import tensorstore as ts
 import torch
 
 from aind_torch_utils.config import InferenceConfig
-from aind_torch_utils.workers import Preds, WriterWorker
+from aind_torch_utils.workers import GpuWorker, Preds, PrepWorker, WriterWorker
+
+
+def _make_input_store(shape):
+    spec = {
+        "driver": "zarr",
+        "kvstore": {"driver": "memory"},
+        "metadata": {
+            "shape": shape,
+            "chunks": (1, 1, 16, 16, 16),
+            "dtype": "<u2",
+        },
+    }
+    store = ts.open(spec, create=True).result()
+    arr = np.random.randint(1, 65535, size=shape, dtype=np.uint16)
+    store.write(arr).result()
+    return store
+
+
+def _prep_cfg(use_compile):
+    return InferenceConfig(
+        patch=(16, 16, 16),
+        overlap=4,
+        trim_voxels=2,
+        seam_mode="trim",
+        block=(32, 32, 32),
+        batch_size=4,
+        devices=["cpu"],
+        amp=False,
+        normalize=False,
+        use_compile=use_compile,
+    )
+
+
+def _drain(prep_q):
+    batches = []
+    while not prep_q.empty():
+        batches.append(prep_q.get_nowait())
+    return batches
+
+
+def test_prep_worker_pads_tail_batch_to_constant_shape_when_compiling():
+    """With torch.compile, every batch must have exactly batch_size rows so
+    the compiled model never sees a varying input shape; padded rows must be
+    zero."""
+    store = _make_input_store((1, 1, 32, 32, 32))
+    cfg = _prep_cfg(use_compile=True)
+    prep_q = queue.Queue()
+    PrepWorker(cfg, store, prep_q, cfg.patch).run(threading.Event())
+
+    batches = _drain(prep_q)
+    assert batches
+
+    saw_partial = False
+    total_real = 0
+    for b in batches:
+        assert b.host_in.shape == (cfg.batch_size, 1, *cfg.patch)
+        n_real = len(b.starts_in_block)
+        assert 0 < n_real <= cfg.batch_size
+        assert len(b.valid_sizes) == n_real
+        assert len(b.per_block_minmax) == n_real
+        total_real += n_real
+        if n_real < cfg.batch_size:
+            saw_partial = True
+            assert torch.all(b.host_in[n_real:] == 0)
+
+    # 32^3 block, patch 16, overlap 4 -> 3 starts per axis -> 27 patches
+    assert total_real == batches[0].total_patches_in_block == 27
+    assert saw_partial, "geometry should produce a partial tail batch"
+
+
+def test_prep_worker_does_not_pad_in_eager_mode():
+    """Without torch.compile there is no constant-shape requirement, so the
+    tail batch keeps its true row count instead of wasting compute and copy
+    bandwidth on zero padding (matching the pre-compile behavior)."""
+    store = _make_input_store((1, 1, 32, 32, 32))
+    cfg = _prep_cfg(use_compile=False)
+    prep_q = queue.Queue()
+    PrepWorker(cfg, store, prep_q, cfg.patch).run(threading.Event())
+
+    batches = _drain(prep_q)
+    assert batches
+
+    saw_partial = False
+    for b in batches:
+        n_real = len(b.starts_in_block)
+        # No padding: the allocation matches the real number of patches.
+        assert b.host_in.shape == (n_real, 1, *cfg.patch)
+        if n_real < cfg.batch_size:
+            saw_partial = True
+    assert saw_partial, "geometry should produce a partial tail batch"
 
 
 def test_writer_raises_on_mismatched_output_channels_and_writers():
@@ -34,3 +126,57 @@ def test_writer_raises_on_mismatched_output_channels_and_writers():
 
     with pytest.raises(ValueError, match="Mismatch between model output channels"):
         worker.run(stop_event=threading.Event())
+
+
+def _make_compile_worker():
+    """Build a GpuWorker shell without running __init__ (which needs CUDA),
+    wired with just the attributes _compile_model touches."""
+    worker = object.__new__(GpuWorker)
+    worker.cfg = InferenceConfig(devices=["cpu"], use_compile=True)
+    worker.device = torch.device("cpu")
+    worker.model = torch.nn.Identity()
+    return worker
+
+
+def test_compile_model_falls_back_to_eager_when_compile_call_raises(monkeypatch):
+    worker = _make_compile_worker()
+    eager = worker.model
+
+    def boom(*args, **kwargs):
+        raise RuntimeError("backend unavailable")
+
+    monkeypatch.setattr(torch, "compile", boom)
+
+    worker._compile_model()
+
+    assert worker.model is eager
+
+
+def test_compile_model_falls_back_to_eager_when_warmup_raises(monkeypatch):
+    """The real failure mode: torch.compile returns lazily, then tracing
+    blows up on the first forward inside warmup."""
+    worker = _make_compile_worker()
+    eager = worker.model
+
+    monkeypatch.setattr(torch, "compile", lambda model, **kwargs: torch.nn.Identity())
+
+    def warmup_boom(self):
+        raise RuntimeError("Guard failed on the same frame it was created")
+
+    monkeypatch.setattr(GpuWorker, "_warmup_compiled_model", warmup_boom)
+
+    worker._compile_model()
+
+    assert worker.model is eager
+
+
+def test_compile_model_keeps_compiled_module_on_success(monkeypatch):
+    worker = _make_compile_worker()
+    compiled = torch.nn.Identity()
+
+    monkeypatch.setattr(torch, "compile", lambda model, **kwargs: compiled)
+    monkeypatch.setattr(GpuWorker, "_warmup_compiled_model", lambda self: None)
+
+    worker._compile_model()
+
+    assert worker.model is compiled