Strengthen chain integration test to verify params carry-forward

The TestJobResubmissionDynamicMultipleIntegration test previously only
asserted the chained dynamic destinations (initial -> secondary ->
tertiary) walked end-to-end without crashing. Extend each rule in
resubmission_rules/rules.py to also read job.destination_params for
a `chain_attempt` counter set by the previous link, and raise
JobMappingException if it does not match the expected value.

This makes the same single integration test simultaneously verify
both behaviours of this PR:

1. Chain re-walk: every resubmit re-evaluates from the persisted
   dynamic intent rather than reusing the cached resolved destination
   of the prior attempt.
2. destination_params carry-forward: the prior attempt's resolved
   destination_params survive the resubmit handler so the next rule
   in the chain can read them on pickup.

A regression in either property surfaces as a JobMappingException
inside the rule, failing the integration test rather than producing
a silently-wrong result.

Verified passing locally (47s) alongside the other dynamic
resubmission integration tests:
- TestJobResubmissionDynamicIntegration::test_dynamic_resubmission
- TestJobResubmissionSmallMemoryResubmitsToLargeIntegration::test_dynamic_resubmission

test/integration/resubmission_rules/rules.py

@@ -22,10 +22,29 @@ def dynamic_resubmit_once(resource_params) -> JobDestination:
     )
 
 
-def dynamic_resubmit_initial() -> JobDestination:
+def _expected_chain_attempt(job, expected: int) -> None:
+    """Assert the prior attempt's destination_params carried forward.
+
+    Reaching the secondary/tertiary rules with the right `chain_attempt` in
+    `job.destination_params` requires both (a) multiple resubmits to walk the
+    chain afresh on each pickup, and (b) the resubmit handler to merge the
+    prior attempt's destination_params into the persisted dispatcher so the
+    rule sees them on re-entry. Raising here surfaces a regression as a
+    JobMappingException rather than silently producing a wrong result.
+    """
+    from galaxy.jobs.mapper import JobMappingException
+
+    actual = int((job.destination_params or {}).get("chain_attempt", 0))
+    if actual != expected:
+        raise JobMappingException(f"chain_attempt carry-forward broken: expected {expected}, got {actual}")
+
+
+def dynamic_resubmit_initial(job) -> JobDestination:
     """First link of a chained dynamic destination: fail and resubmit to the second link."""
+    _expected_chain_attempt(job, 0)
     return JobDestination(
         runner="failure_runner",
+        params={"chain_attempt": 1},
         resubmit=[
             dict(
                 condition="any_failure",
@@ -35,15 +54,19 @@ def dynamic_resubmit_initial() -> JobDestination:
     )
 
 
-def dynamic_resubmit_secondary() -> JobDestination:
+def dynamic_resubmit_secondary(job) -> JobDestination:
     """Second link: fail and resubmit to the third link.
 
     Reaching this rule on the *second* resubmit-attempt requires that the
     chain re-evaluates from the persisted dynamic intent rather than from
-    the cached resolved destination of the previous attempt.
+    the cached resolved destination of the previous attempt. Asserting on
+    `chain_attempt == 1` additionally requires that destination_params from
+    the prior attempt survived the resubmit handler.
     """
+    _expected_chain_attempt(job, 1)
     return JobDestination(
         runner="failure_runner",
+        params={"chain_attempt": 2},
         resubmit=[
             dict(
                 condition="any_failure",
@@ -53,6 +76,10 @@ def dynamic_resubmit_secondary() -> JobDestination:
     )
 
 
-def dynamic_resubmit_tertiary() -> JobDestination:
-    """Third link: succeed on the local runner."""
+def dynamic_resubmit_tertiary(job) -> JobDestination:
+    """Third link: succeed on the local runner.
+
+    Asserts the counter reached 2 to confirm both resubmits carried params.
+    """
+    _expected_chain_attempt(job, 2)
     return JobDestination(runner="local")

test/integration/test_job_resubmission.py

@@ -232,9 +232,20 @@ class TestJobResubmissionDynamicMultipleIntegration(_BaseResubmissionIntegration
     Three dynamic destinations form a chain: initial -> secondary -> tertiary.
     Each link uses ``failure_runner`` and resubmits to the next link via its
     ``resubmit.environment``; the last link routes to ``local`` (which passes).
-    The job only passes if every resubmit re-walks the chain from the
-    persisted dynamic intent rather than reusing the cached resolved
-    destination of the previous attempt.
+
+    Each rule also reads ``job.destination_params["chain_attempt"]`` set by
+    the prior link and asserts on its value, so the test simultaneously
+    verifies that:
+
+    1. Every resubmit re-walks the chain from the persisted dynamic intent
+       rather than reusing the cached resolved destination of the previous
+       attempt (chain re-walk).
+    2. Prior attempt's ``destination_params`` survive the resubmit handler
+       and are visible to the rule on the next pickup (params carry-forward).
+
+    A regression in either property surfaces as a ``JobMappingException``
+    raised inside the rule rather than a silently-wrong destination, so the
+    job fails the test instead of passing with the wrong behaviour.
     """
 
     framework_tool_and_types = True
@@ -244,7 +255,7 @@ class TestJobResubmissionDynamicMultipleIntegration(_BaseResubmissionIntegration
         super().handle_galaxy_config_kwds(config)
         config["job_config_file"] = JOB_RESUBMISSION_DYNAMIC_MULTIPLE_JOB_CONFIG_FILE
 
-    def test_chained_dynamic_resubmission(self):
+    def test_chained_dynamic_resubmission_with_params_carry_forward(self):
         self._assert_job_passes()