feat: support pre-existing job definitions via containerOverrides (#44)

## Summary

The plugin always registered (and later deregistered) a fresh job
definition per job — flexible, but it requires
`batch:RegisterJobDefinition`/`DeregisterJobDefinition` IAM that
security-conscious accounts refuse, churns thousands of single-use
revisions, and keeps resource config out of infra review. A new optional
`--aws-batch-job-definition <name|name:rev|ARN>` setting (with a
per-rule `aws_batch_job_definition` resource override, mirroring
`batch_queue`) switches to static mode:

- No `RegisterJobDefinition` call; per-job specifics travel via
`containerOverrides` (`command`, `environment`, vcpu/mem
`resourceRequirements`, GPU only when >0 — same resource extraction as
the dynamic path)
- Submitted jobs are marked `_preexisting_job_definition` (carried
through `run_job`'s `aux=dict(job_info)`), and every deregister/cleanup
path skips them — we never deregister a definition we didn't register
- Fail-fast `WorkflowError` when combined with `job_role` or
`shared_memory_size_mb` (build-only knobs); `container_image` is
documented-ignored since its default is indistinguishable from explicit
- Tags flow through the same `_build_job_tags()` as the dynamic path
- The optional submit kwargs the dynamic path applies are mirrored via
shared resolvers: task timeout travels as `SubmitJob`'s top-level
`timeout` field (the dynamic path bakes it into the definition), and
scheduling priority forwards `schedulingPriorityOverride` identically
- Platform (EC2/FARGATE) discovery is resolved lazily, so pre-existing
mode never queries the queue — keeping its IAM surface free of
`DescribeJobQueues`/`DescribeComputeEnvironments`
- The default (dynamic) path is byte-identical to before

## Test Plan

- [x] New tests (`TestPreExistingJobDefinition`,
`TestDeregisterSkipsPreexisting`), including an end-to-end
marker-survival test through the real `submit → SubmittedJobInfo(aux=…)
→ _interpret_job_status aux write → _deregister_job` chain, plus
timeout/scheduling-priority mirroring and lazy-platform coverage
- [x] Full builder suite green (98 tests); black clean; flake8 clean

Author: nh13

docs/further.md

@@ -199,3 +199,54 @@ coordinator job can set the variable so that all child jobs it submits
 inherit the run-specific tags. AWS Batch allows at most 50 tags per job;
 malformed pairs (missing `=` or an empty key) raise an error at submission
 time.
+
+# Pre-existing Job Definitions
+
+By default the plugin registers a fresh AWS Batch job definition for every job
+and deregisters it afterward.  Accounts where job definitions are managed by
+infrastructure tooling (Terraform, CloudFormation) can opt out of this with
+`--aws-batch-job-definition`.  When set, the plugin skips
+`RegisterJobDefinition`/`DeregisterJobDefinition` entirely and instead submits
+with the supplied definition, pushing per-job specifics (command, environment
+variables, vcpu/mem/gpu) through `containerOverrides`.
+
+```bash
+snakemake --executor aws-batch \
+    --aws-batch-job-definition my-snakemake-def:3 \
+    ...
+```
+
+The value can be a bare name (`my-def`), a name:revision pair (`my-def:3`),
+or a full ARN
+(`arn:aws:batch:us-east-1:123456789012:job-definition/my-def:3`).
+
+A rule can override the setting for a specific job with the
+`aws_batch_job_definition` resource (mirrors the `batch_queue` pattern):
+
+```python
+rule align:
+    resources:
+        aws_batch_job_definition="gpu-enabled-def:2"
+    ...
+```
+
+**Incompatible combinations** — the following raise a `WorkflowError` at
+submission time because they are only meaningful when the plugin builds the
+definition:
+
+- `--aws-batch-job-role` (`job_role`): the job role is baked into the
+  definition at registration time and cannot be overridden via
+  `containerOverrides`.
+- The per-rule `shared_memory_size_mb` resource: `linuxParameters.sharedMemorySize`
+  is a definition-level field.
+
+The `--aws-batch-container-image` (`container_image`) setting and the per-rule
+`aws_batch_container_image` resource are both silently ignored in this mode — the
+container image is taken from the pre-existing definition.
+
+Task timeout and scheduling priority **are** still honored: `--aws-batch-task-timeout`
+(and the per-rule `aws_batch_task_timeout` resource) and the scheduling priority
+(`--aws-batch-scheduling-priority` / the per-rule `aws_batch_scheduling_priority`
+resource) travel as `SubmitJob`'s top-level `timeout` and `schedulingPriorityOverride`
+fields, so they apply to pre-existing definitions just as they do to dynamically
+registered ones.

snakemake_executor_plugin_aws_batch/__init__.py

@@ -89,6 +89,23 @@ class ExecutorSettings(ExecutorSettingsBase):
             "required": False,
         },
     )
+    job_definition: Optional[str] = field(
+        default=None,
+        metadata={
+            "help": (
+                "Use a pre-existing AWS Batch job definition instead of registering "
+                "one per job. Accepts a definition name (e.g. my-def), a name:revision "
+                "pair (my-def:3), or a full ARN. When set, resource configuration "
+                "(container image, job role, shared memory) is managed externally — "
+                "the container_image setting is ignored. Per-job specifics (command, "
+                "environment variables, vcpu/mem/gpu) are passed via "
+                "containerOverrides. Cannot be combined with --aws-batch-job-role "
+                "or the per-rule shared_memory_size_mb resource."
+            ),
+            "env_var": False,
+            "required": False,
+        },
+    )
 
 
 # Required:
@@ -277,6 +294,15 @@ def _terminate_job(self, job: SubmittedJobInfo):
     def _deregister_job(self, job: SubmittedJobInfo):
         """deregister batch job definition"""
         try:
+            # Pre-existing job definitions are not owned by this executor;
+            # deregistering them would break other workflows using the same definition.
+            if job.aux.get("_preexisting_job_definition"):
+                self.logger.debug(
+                    "skipping deregistration of pre-existing Batch job definition "
+                    f"{job.aux.get('job_definition_arn')} "
+                    "(externally managed, not owned by this executor)"
+                )
+                return
             job_def_arn = job.aux.get("job_definition_arn")
             if job_def_arn is not None:
                 self.logger.debug(f"de-registering Batch job definition {job_def_arn}")