fix(batch): scope Batch job ids to the selected pool

.github/scripts/deploy_apps.sh

@@ -134,7 +134,8 @@ deploy_function() {
             "STATIC_APP_NAME=$STATIC_WEB_APP" \
             "STATIC_APP_DOMAIN=${STATIC_APP_DOMAIN}" \
             "EMAIL_CONNECTION_STRING=${EMAIL_CONNECTION_STRING}" \
-            "EMAIL_SENDER=${EMAIL_SENDER}"
+            "EMAIL_SENDER=${EMAIL_SENDER}" \
+            --output none
     fi
 
     az functionapp restart --name "$FUNCTION_NAME" --resource-group "$RESOURCE_GROUP"

.github/workflows/deploy-apps.yml

@@ -116,15 +116,16 @@ jobs:
           # on Azure's remote build.
           HASTEGEO_WHEEL_URL: ${{ steps.hastegeo.outputs.url }}
           # Batch pool wiring. All optional: unset reproduces the legacy
-          # single-pool behavior. Set these as GitHub Environment variables to
-          # point an environment at pre-created shared pools.
-          BATCH_TRAINING_POOL_ID: ${{ vars.BATCH_TRAINING_POOL_ID }}
-          BATCH_IMAGERYPREP_POOL_ID: ${{ vars.BATCH_IMAGERYPREP_POOL_ID }}
-          BATCH_TRAINING_POOL_IDS: ${{ vars.BATCH_TRAINING_POOL_IDS }}
-          BATCH_INFERENCE_POOL_IDS: ${{ vars.BATCH_INFERENCE_POOL_IDS }}
-          BATCH_IMAGERYPREP_POOL_IDS: ${{ vars.BATCH_IMAGERYPREP_POOL_IDS }}
-          BATCH_USE_SAS: ${{ vars.BATCH_USE_SAS }}
-          BATCH_MANAGE_POOLS: ${{ vars.BATCH_MANAGE_POOLS }}
+          # single-pool behavior. Stored as environment *secrets* (not
+          # variables) so the values are masked in this repo's public
+          # Actions logs, matching the other infra-naming config.
+          BATCH_TRAINING_POOL_ID: ${{ secrets.BATCH_TRAINING_POOL_ID }}
+          BATCH_IMAGERYPREP_POOL_ID: ${{ secrets.BATCH_IMAGERYPREP_POOL_ID }}
+          BATCH_TRAINING_POOL_IDS: ${{ secrets.BATCH_TRAINING_POOL_IDS }}
+          BATCH_INFERENCE_POOL_IDS: ${{ secrets.BATCH_INFERENCE_POOL_IDS }}
+          BATCH_IMAGERYPREP_POOL_IDS: ${{ secrets.BATCH_IMAGERYPREP_POOL_IDS }}
+          BATCH_USE_SAS: ${{ secrets.BATCH_USE_SAS }}
+          BATCH_MANAGE_POOLS: ${{ secrets.BATCH_MANAGE_POOLS }}
         run: |
           chmod +x .github/scripts/deploy_apps.sh
           .github/scripts/deploy_apps.sh \

CHANGELOG.md

@@ -17,7 +17,7 @@ Versioning follows the Docker image tags defined in the CI workflows (see [.gith
 
   > **Operator action:** rename the `AZURE_BATCH_REGISTRY_SERVER_URL` application setting to `AZURE_BATCH_REGISTRY_SERVER` (value: the bare login server, e.g. `myacr.azurecr.io`) on the `api` and `queues` Function Apps. Confirm `AZURE_BATCH_TRAINING_POOL_ID` / `AZURE_BATCH_IMAGERYPREP_POOL_ID` name pools that still exist and are **identical across both apps** — they double as the default Batch job ids.
 
-- **A reused Batch job stayed pinned to the pool that created it** — job ids default to the configured pool id, and `create_job` only re-enabled an existing job, never re-pointing it. Capacity-aware spillover was therefore silently ineffective, and once a pool was renamed or deleted every task queued into a job bound to a pool that no longer existed. The runner now rebinds the job to the pool selected for the submission, and fails loudly when Batch refuses.
+- **A reused Batch job stayed pinned to the pool that created it** — job ids default to the configured pool id, and `create_job` only re-enabled an existing job, never re-pointing it. Capacity-aware spillover was therefore silently ineffective, and once a pool was renamed or deleted every task queued into a job bound to a pool that no longer existed. Job ids are now derived from the pool a task is routed to (one job per pool), so spillover works even while another task is running; `create_job` falls back to a pool-scoped job rather than failing the submission. Environments with a single candidate pool keep their existing job ids unchanged. Full design in [`spec/features/batch-pool-job-binding/`](spec/features/batch-pool-job-binding).
 - **Missing Batch settings now fail fast** — the runner validates its configuration before the first Batch call and names the specific application setting that is unset, instead of surfacing an opaque Azure error from deep inside pool creation.
 - **Pool creation whitelisted only the training image** — pools created by the runner now list both the training and imageryprep images, matching [`infra/modules/batchPool.bicep`](infra/modules/batchPool.bicep).
 

docs/configuration.md

@@ -176,19 +176,29 @@ The runner picks a pool from the candidate list **at submit time** — the first
 with an idle node, otherwise the preferred (first) pool, which scales up / queues.
 
 These are emitted by both deploy paths: set them as GitHub Environment
-variables (`BATCH_TRAINING_POOL_IDS`, `BATCH_INFERENCE_POOL_IDS`,
+**secrets** (`BATCH_TRAINING_POOL_IDS`, `BATCH_INFERENCE_POOL_IDS`,
 `BATCH_IMAGERYPREP_POOL_IDS`, `BATCH_USE_SAS`, `BATCH_MANAGE_POOLS`,
 `BATCH_TRAINING_POOL_ID`, `BATCH_IMAGERYPREP_POOL_ID`) for
 [`deploy-apps.yml`](../.github/workflows/deploy-apps.yml), or as the
 corresponding `infra/main.bicepparam` values for the Bicep path.
 
+> **Secrets, not variables.** Resource naming is treated as sensitive in this
+> repo (as `RESOURCE_PREFIX`, `LOCATION` and friends already are). GitHub masks
+> secret values in Actions logs but does **not** mask variables, and this repo's
+> workflow logs are public.
+
 > **`*_POOL_ID` (singular) also names the Batch job.** `TRAINING_BATCH_JOB_ID`
 > and friends default to the matching pool id, so the value must be **identical
 > on the api and queues apps** — the queues app submits under that job id and
-> the api app reads status back from it. Pointing it at a pool that no longer
-> exists leaves a job permanently bound to a deleted pool; the runner now
-> rebinds such a job to the pool it selected, but the setting should still be
-> corrected.
+> the api app reads status back from it.
+
+**Job ids are scoped to the selected pool.** A Batch job is permanently bound to
+the pool it was created against and can only be re-pointed while it has no
+active tasks, so one static job id cannot span pools. When routing across
+multiple candidates, the runner derives the job id from the pool the task was
+routed to — one job per pool. Environments with a single candidate pool keep
+their existing job id unchanged. This is why a task that spills over to a
+second pool no longer collides with the job created on the preferred pool.
 
 ### Keeping settings in sync
 

hastelib/src/hastegeo/core/processors/artifacts.py

@@ -310,7 +310,7 @@ def submit_zip_job(self):
             task_id = f"{ZIP_PREFIX}-{MetadataUtils.generate_id()}"
             zip_output_prefix = f"{MetadataUtils.hash_string(self.model_artifacts.projectId)}/{task_id}"
 
-            self.runner.add_task(
+            job_id, task_id = self.runner.add_task(
                 job_id=job_id,
                 task_id=task_id,
                 output_prefix=zip_output_prefix,

hastelib/src/hastegeo/core/processors/embedding.py

@@ -169,7 +169,7 @@ def _execute_embedding(self):
                 f"{MetadataUtils.hash_string(self.model_data.projectId)}"
                 f"/{task_id}"
             )
-            self.runner.add_task(
+            job_id, task_id = self.runner.add_task(
                 job_id=job_id,
                 task_id=task_id,
                 output_prefix=output_prefix,

hastelib/src/hastegeo/core/processors/imagery.py

@@ -338,7 +338,7 @@ def _execute_image_preprocess(self):
         imagery_output_prefix = (
             f"{MetadataUtils.hash_string(self.image_data.projectId)}/{task_id}"
         )
-        self.runner.add_task(
+        job_id, task_id = self.runner.add_task(
             job_id=job_id,
             task_id=task_id,
             output_prefix=imagery_output_prefix,

hastelib/src/hastegeo/core/processors/inference.py

@@ -289,7 +289,7 @@ def _execute_inference(self):
             task_id = f"{INFERENCE_PREFIX}-{MetadataUtils.generate_id()}"
             inference_output_prefix = f"{MetadataUtils.hash_string(self.model_data.projectId)}/{task_id}"
 
-            self.runner.add_task(
+            job_id, task_id = self.runner.add_task(
                 job_id=job_id,
                 task_id=task_id,
                 output_prefix=inference_output_prefix,

hastelib/src/hastegeo/core/processors/train.py

@@ -265,7 +265,7 @@ def _execute_training(self):
             job_id = job_id[:64]
             task_id = f"{TRAINING_PREFIX}-{MetadataUtils.generate_id()}"
             training_output_prefix = f"{MetadataUtils.hash_string(self.model_data.projectId)}/{task_id}"
-            self.runner.add_task(
+            job_id, task_id = self.runner.add_task(
                 job_id=job_id,
                 task_id=task_id,
                 output_prefix=training_output_prefix,

hastelib/src/hastegeo/core/runners/azure_batch.py

@@ -49,7 +49,7 @@
 )
 from hastegeo.core.config import Config
 from hastegeo.core.utils.batch_config import (
-    BatchJobPoolMismatchError,
+    resolve_job_id,
     validate_batch_config,
 )
 from hastegeo.core.utils.logs import Logger
@@ -163,6 +163,11 @@ def add_task(
             task_id,
         )
 
+        # A Batch job is pinned to one pool, so the job id has to follow the
+        # pool this task was routed to; otherwise a second task that spills over
+        # to another pool collides with the job the first one created.
+        job_id = resolve_job_id(job_id, selected_pool, self.candidate_pool_ids)
+
         # Pre-created IaC/autoscale pools manage their own lifecycle; only
         # auto-create/resize for legacy single-pool envs (manage_pools=True).
         if self.manage_pools:
@@ -194,7 +199,7 @@ def add_task(
         self.logger.info(
             "Creating job for job_id: %s and task_id: %s", job_id, task_id
         )
-        self.batch_cluster.create_job(job_id)
+        job_id = self.batch_cluster.create_job(job_id)
 
         self.batch_cluster.add_task(
             job_id=job_id,
@@ -451,55 +456,83 @@ def create_pool_if_not_exists(
                 raise
 
     def create_job(self, job_id):
+        """Ensure an active job bound to ``self.pool_id``.
+
+        Returns the job id that was actually used, which may differ from the
+        requested one when an existing job is pinned to another pool and cannot
+        be re-pointed.
+        """
         # Fixed pools we manage need a ready node before the job is created;
         # autoscale / pre-created pools scale up in response to queued tasks, so
         # waiting here would deadlock (0 nodes until tasks exist).
         if self.manage_pools:
             self.wait_for_pool_to_be_ready()
         try:
             job = self.batch_client.job.get(job_id)
-            self._rebind_job_pool(job)
-            if job.state != JobState.active:
-                self.batch_client.job.enable(job_id)
-                self.logger.info(f"Job {job_id} activated.")
         except BatchErrorException as e:
             if e.error.code == "JobNotFound":
-                job = JobAddParameter(
-                    id=job_id, pool_info=PoolInformation(pool_id=self.pool_id)
-                )
-                self.batch_client.job.add(job)
-                self.logger.info(f"Job {job_id} created.")
-
-    def _rebind_job_pool(self, job):
-        # A job keeps the pool it was first created against, so reusing a job id
-        # would pin every later task to that original pool -- defeating
-        # capacity-aware routing, and failing outright once that pool is renamed
-        # or deleted. Re-point the job at the pool selected for this submission.
+                self._add_job(job_id)
+                return job_id
+            raise
+
         bound_pool = getattr(job.pool_info, "pool_id", None)
         if not bound_pool or bound_pool == self.pool_id:
-            return
+            if job.state != JobState.active:
+                self.batch_client.job.enable(job_id)
+                self.logger.info(f"Job {job_id} activated.")
+            return job_id
+
+        # The job belongs to a different pool. Re-pointing it only works while
+        # it has no active tasks, so fall back to a pool-scoped job rather than
+        # failing the submission.
         self.logger.info(
             "Job %s is bound to pool %s; rebinding to selected pool %s.",
-            job.id,
+            job_id,
             bound_pool,
             self.pool_id,
         )
         try:
             self.batch_client.job.patch(
-                job.id,
+                job_id,
                 JobPatchParameter(
                     pool_info=PoolInformation(pool_id=self.pool_id)
                 ),
             )
+            if job.state != JobState.active:
+                self.batch_client.job.enable(job_id)
+            return job_id
         except BatchErrorException as e:
-            # Batch refuses the change while tasks are active. Surface it rather
-            # than silently submitting into a job on the wrong pool.
-            raise BatchJobPoolMismatchError(
-                f"Job {job.id} is bound to pool {bound_pool} but this task "
-                f"targets pool {self.pool_id}, and rebinding failed "
-                f"({getattr(e.error, 'code', e)}). Delete or drain the job, "
-                "or set a distinct *_BATCH_JOB_ID."
-            ) from e
+            fallback_id = resolve_job_id(
+                job_id, self.pool_id, [job_id, self.pool_id]
+            )
+            self.logger.info(
+                "Rebinding job %s failed (%s); using pool-scoped job %s.",
+                job_id,
+                getattr(e.error, "code", e),
+                fallback_id,
+            )
+            return self._ensure_job_on_pool(fallback_id)
+
+    def _ensure_job_on_pool(self, job_id):
+        # Last step of the fallback: the id is derived from the pool, so it can
+        # only be missing or already bound to that same pool.
+        try:
+            job = self.batch_client.job.get(job_id)
+        except BatchErrorException as e:
+            if e.error.code == "JobNotFound":
+                self._add_job(job_id)
+                return job_id
+            raise
+        if job.state != JobState.active:
+            self.batch_client.job.enable(job_id)
+        return job_id
+
+    def _add_job(self, job_id):
+        job = JobAddParameter(
+            id=job_id, pool_info=PoolInformation(pool_id=self.pool_id)
+        )
+        self.batch_client.job.add(job)
+        self.logger.info(f"Job {job_id} created on pool {self.pool_id}.")
 
     def add_task(
         self,

hastelib/src/hastegeo/core/utils/batch_config.py

@@ -8,12 +8,19 @@
 until Azure rejects it — surfacing as an opaque API error far from the cause
 (e.g. ``InvalidPropertyValue`` on ``registryServer``). Validating before the
 first Batch call lets the failure name the setting that is actually missing.
+
+This module also resolves the Batch *job* id, which has to account for
+capacity-aware routing: a Batch job is permanently bound to one pool, but the
+pool is chosen per task, so a single static job id cannot span pools.
 """
 
 import re
 
 PLACEHOLDER_PATTERN = re.compile(r"<[^<>]+>")
 
+# Azure Batch job ids are limited to 64 characters.
+MAX_JOB_ID_LENGTH = 64
+
 # Settings the runner needs for any submission, mapped to the application
 # setting that supplies each one.
 ALWAYS_REQUIRED = {
@@ -37,10 +44,6 @@ class BatchConfigurationError(RuntimeError):
     """Raised when a required Azure Batch application setting is missing."""
 
 
-class BatchJobPoolMismatchError(RuntimeError):
-    """Raised when a reused Batch job is pinned to a different pool."""
-
-
 def has_placeholder(value):
     """True when ``value`` still contains an unresolved ``<...>`` default."""
     return isinstance(value, str) and bool(PLACEHOLDER_PATTERN.search(value))
@@ -75,3 +78,41 @@ def validate_batch_config(batch_config, manage_pools=None):
             + "; ".join(unresolved)
             + ". Set these on the Function App and restart it."
         )
+
+
+def resolve_job_id(base_job_id, selected_pool, candidate_pool_ids=None):
+    """Return the Batch job id to use for a task routed to ``selected_pool``.
+
+    A Batch job is permanently bound to the pool it was created against, and
+    can only be re-pointed while it has no active tasks. Capacity-aware routing
+    picks the pool per task, so reusing one static job id across pools breaks as
+    soon as two tasks are in flight on different pools. Scoping the job id to
+    the selected pool gives one job per pool and removes the conflict.
+
+    Environments that are not routing across multiple pools keep their existing
+    job id, so their jobs are not renamed.
+
+    Args:
+        base_job_id: The configured job id (e.g. ``IMAGERYPREP_BATCH_JOB_ID``).
+        selected_pool: The pool this task was routed to.
+        candidate_pool_ids: The pools routing may choose between.
+    """
+    if not selected_pool:
+        return base_job_id
+    candidates = list(candidate_pool_ids or [])
+
+    # The default convention is job id == pool id, so follow the selected pool
+    # and keep names clean rather than doubling the pool id up.
+    if base_job_id in candidates:
+        return selected_pool[:MAX_JOB_ID_LENGTH]
+
+    # A single candidate cannot spill over, so leave custom ids untouched.
+    if len(candidates) <= 1:
+        return base_job_id
+
+    # Reserve room for the suffix: truncating the base instead of the pool
+    # keeps ids for two different pools distinct.
+    room = MAX_JOB_ID_LENGTH - len(selected_pool) - 1
+    if room <= 0:
+        return selected_pool[:MAX_JOB_ID_LENGTH]
+    return f"{base_job_id[:room]}-{selected_pool}"