From bd5b7169049d983cb3b359a5306c591b65ad898d Mon Sep 17 00:00:00 2001
From: Jay W <git.jaydoubleu@gmail.com>
Date: Sun, 11 Oct 2026 11:29:04 +0100
Subject: [PATCH 1/3] Fix #5231: retain CI environments after branch validation

---
 .github/scripts/package-lock.json             |   4 +-
 .github/scripts/package.json                  |   2 +-
 .../workflows/build_validation_develop.yml    |   1 +
 CHANGELOG.md                                  |   1 +
 devops/scripts/ci_cleanup_activity.py         | 127 +++++++++++++
 devops/scripts/ci_cleanup_scope.py            |  19 +-
 devops/scripts/clean_ci_validation_envs.sh    |  24 ++-
 devops/scripts/recover_bootstrap_lease.py     |   1 +
 devops/tests/test_ci_cleanup_activity.py      | 178 ++++++++++++++++++
 devops/tests/test_ci_cleanup_scope.py         |  35 +++-
 devops/tests/test_clean_ci_validation_envs.py |  40 ++++
 devops/tests/test_validate_ci_cleanup.py      |   2 +
 devops/version.txt                            |   2 +-
 13 files changed, 422 insertions(+), 14 deletions(-)
 create mode 100644 devops/scripts/ci_cleanup_activity.py
 create mode 100644 devops/tests/test_ci_cleanup_activity.py

diff --git a/.github/scripts/package-lock.json b/.github/scripts/package-lock.json
index 11613c9df..b28d9ccc7 100644
--- a/.github/scripts/package-lock.json
+++ b/.github/scripts/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "build-scripts",
-  "version": "1.0.3",
+  "version": "1.0.4",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "build-scripts",
-      "version": "1.0.3",
+      "version": "1.0.4",
       "license": "MIT",
       "devDependencies": {
         "jest": "^30.2.0",
diff --git a/.github/scripts/package.json b/.github/scripts/package.json
index 58e755bb0..0452315bc 100644
--- a/.github/scripts/package.json
+++ b/.github/scripts/package.json
@@ -1,6 +1,6 @@
 {
   "name": "build-scripts",
-  "version": "1.0.3",
+  "version": "1.0.4",
   "description": "Scripts used in the GitHub workflows",
   "main": "build.js",
   "license": "MIT",
diff --git a/.github/workflows/build_validation_develop.yml b/.github/workflows/build_validation_develop.yml
index 956c16646..b88d6c0fa 100644
--- a/.github/workflows/build_validation_develop.yml
+++ b/.github/workflows/build_validation_develop.yml
@@ -74,6 +74,7 @@ jobs:
               - 'devops/scripts/recover_bootstrap_lease.py'
               - 'devops/scripts/ci_environment_id.py'
               - 'devops/scripts/ci_cleanup_scope.py'
+              - 'devops/scripts/ci_cleanup_activity.py'
               - 'devops/scripts/destroy_ci_ref_env.py'
               - 'devops/scripts/destroy_env_no_terraform.sh'
               - 'devops/scripts/validate_ci_cleanup.py'
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ffa456623..6dabe2c68 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,6 +27,7 @@ See [Legacy Airlock & migration](docs/azure-tre-overview/airlock.md#legacy-airlo
 * Add weekly and manual builds for Dockerfiles and Porter bundles, with selected pull request builds. Retry a failed target once and report final results. ([#5088](https://github.com/microsoft/AzureTRE/issues/5088))
 
 BUG FIXES:
+* Defer CI environment cleanup for queued operations and recent native-branch validation activity. Preserve scoped ownership checks and orphan deletion. (`devops` 0.6.14, `build-scripts` 1.0.4) ([#5231](https://github.com/microsoft/AzureTRE/issues/5231))
 * Fix denial-of-service dependency alerts by updating `source-map-js` to 1.2.2 in the UI and removing transitive `sprintf-js` from workflow tests through an `argparse` 2 override for `js-yaml` 3. (`ui` 0.10.3, `build-scripts` 1.0.3) ([#5118](https://github.com/microsoft/AzureTRE/pull/5118))
 * Align the `create_aad_groups` schema and Porter defaults to `true` across all three workspace bundles, preventing an unrelated edit from unintentionally removing workspace role groups and app-role assignments.
   Add regression checks for each bundle. (`tre-workspace-base` 2.12.2, `tre-workspace-unrestricted` 0.15.1, `tre-workspace-airlock-import-review` 0.18.1) ([#5050](https://github.com/microsoft/AzureTRE/issues/5050))
diff --git a/devops/scripts/ci_cleanup_activity.py b/devops/scripts/ci_cleanup_activity.py
new file mode 100644
index 000000000..0148b230d
--- /dev/null
+++ b/devops/scripts/ci_cleanup_activity.py
@@ -0,0 +1,127 @@
+"""Defer scoped cleanup while matching validation is active or recently used a branch."""
+
+from datetime import datetime, timezone
+from urllib.parse import urlencode
+
+from recover_bootstrap_lease import RecoveryError, github, require
+
+ACTIVE_STATUSES = frozenset(("in_progress", "queued", "pending", "waiting", "requested"))
+CLEANUP_WORKFLOW = ".github/workflows/clean_validation_envs.yml"
+BRANCH_WORKFLOW = ".github/workflows/deploy_tre_branch.yml"
+
+
+def pending_work(ctx, members):
+    """Ignore only identified cleanup successors in the validated reference queue."""
+    for member in members:
+        if member["status"] == "in_progress":
+            continue  # verify_concurrency already proved this is the current cleanup owner.
+        run = github(ctx, f"/actions/runs/{member['run_id']}")
+        require(
+            isinstance(run, dict)
+            and run.get("id") == member["run_id"]
+            and isinstance(run.get("repository"), dict)
+            and run["repository"].get("full_name") == ctx["repository"],
+            "Cannot identify queued cleanup work.",
+        )
+        if run.get("path") != CLEANUP_WORKFLOW or run.get("event") not in ("schedule", "workflow_dispatch"):
+            return True
+        if "job_id" not in member:
+            return True
+        job = github(ctx, f"/actions/jobs/{member['job_id']}")
+        if not (
+            isinstance(job, dict)
+            and job.get("id") == member["job_id"]
+            and job.get("run_id") == member["run_id"]
+            and type(run.get("run_attempt")) is int
+            and run["run_attempt"] > 0
+            and job.get("run_attempt") == run["run_attempt"]
+            and job.get("name") == "Clean " + ctx["ref"]
+            and job.get("status") in ACTIVE_STATUSES
+        ):
+            return True
+    return False
+
+
+def timestamp(value):
+    require(isinstance(value, str), "Missing branch validation timestamp.")
+    try:
+        result = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as error:
+        raise RecoveryError("Invalid branch validation timestamp.") from error
+    require(result.tzinfo is not None, "Branch validation timestamp has no time zone.")
+    return result.timestamp()
+
+
+def branch_activity(ctx, commit_time, now=None):
+    """Read all matching branch runs, including old runs with recent rerun activity."""
+    require(
+        ctx["ref"].startswith("refs/heads/") and type(commit_time) is int and commit_time > 0,
+        "Invalid branch cleanup activity request.",
+    )
+    now = datetime.now(timezone.utc).timestamp() if now is None else now
+    branch = ctx["ref"].removeprefix("refs/heads/")
+    latest = commit_time
+    runs, total = {}, None
+    for page in range(1, 12):
+        query = urlencode({"branch": branch, "event": "workflow_dispatch", "per_page": 100, "page": page})
+        data = github(ctx, "/actions/workflows/deploy_tre_branch.yml/runs?" + query)
+        require(
+            isinstance(data, dict)
+            and type(data.get("total_count")) is int
+            and 0 <= data["total_count"] <= 1000
+            and isinstance(data.get("workflow_runs"), list),
+            "Incomplete branch validation history.",
+        )
+        total = data["total_count"] if total is None else total
+        require(total == data["total_count"], "Branch validation history changed during cleanup.")
+        for run in data["workflow_runs"]:
+            require(
+                isinstance(run, dict)
+                and type(run.get("id")) is int
+                and run["id"] > 0
+                and run["id"] not in runs
+                and run.get("path") == BRANCH_WORKFLOW
+                and run.get("event") == "workflow_dispatch"
+                and run.get("head_branch") == branch
+                and all(
+                    isinstance(run.get(key), dict) and run[key].get("full_name") == ctx["repository"]
+                    for key in ("repository", "head_repository")
+                ),
+                "Invalid branch validation identity.",
+            )
+            runs[run["id"]] = run
+            if run.get("status") in ACTIVE_STATUSES:
+                # Preparation can be active before the run joins the reference queue.
+                return now
+            require(run.get("status") == "completed", "Unknown branch validation status.")
+            created = timestamp(run.get("created_at"))
+            updated = timestamp(run.get("updated_at"))
+            require(created <= updated <= now, "Inconsistent branch validation activity time.")
+            # updated_at conservatively retains recent failed, cancelled and rerun attempts.
+            # It is an activity time, not a claim about exact job completion.
+            latest = max(latest, updated)
+        require(len(runs) <= total, "Inconsistent branch validation count.")
+        if len(runs) == total:
+            return latest
+        require(len(data["workflow_runs"]) == 100, "GitHub omitted branch validation runs.")
+    raise RecoveryError("Could not read all branch validation history.")
+
+
+def mutation_allowed(ctx, members, commit_time=None, inactive_hours=None, now=None):
+    if pending_work(ctx, members):
+        return False
+    require((commit_time is None) == (inactive_hours is None), "Incomplete cleanup inactivity threshold.")
+    if commit_time is None:
+        # Closed PRs and deleted branches retain their orphan policy. The queue still applies.
+        return True
+    require(
+        ctx["ref"].startswith("refs/heads/")
+        and type(commit_time) is int
+        and commit_time > 0
+        and type(inactive_hours) is int
+        and inactive_hours >= 0,
+        "Invalid branch cleanup inactivity threshold.",
+    )
+    now = datetime.now(timezone.utc).timestamp() if now is None else now
+    latest = branch_activity(ctx, commit_time, now)
+    return (now - latest) // 3600 > inactive_hours
diff --git a/devops/scripts/ci_cleanup_scope.py b/devops/scripts/ci_cleanup_scope.py
index efec4160a..031a2a64a 100644
--- a/devops/scripts/ci_cleanup_scope.py
+++ b/devops/scripts/ci_cleanup_scope.py
@@ -8,6 +8,7 @@
 import sys
 
 from ci_environment_id import validate_ref
+from ci_cleanup_activity import branch_activity, mutation_allowed
 from recover_bootstrap_lease import RecoveryError, github, require, verify_concurrency
 
 
@@ -124,7 +125,7 @@ def verify_lock(ctx):
         raise RecoveryError("Could not verify all cleanup jobs.")
     owners = [job for job in jobs.values() if job["name"] == expected_name and job.get("status") == "in_progress"]
     require(len(owners) == 1, "The current cleanup job is missing or ambiguous.")
-    verify_concurrency(ctx, job_id=owners[0]["id"])
+    return verify_concurrency(ctx, job_id=owners[0]["id"])
 
 
 def verify_target(groups, ref, group):
@@ -142,19 +143,29 @@ def verify_target(groups, ref, group):
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("operation", choices=("plan", "verify-lock", "verify-target"))
+    parser.add_argument("operation", choices=("plan", "verify-lock", "verify-target", "may-clean", "activity-time"))
     parser.add_argument("--group")
+    parser.add_argument("--commit-time", type=int)
+    parser.add_argument("--inactive-hours", type=int)
     args = parser.parse_args()
     try:
         if args.operation == "plan":
             print(json.dumps(references(inventory()), separators=(",", ":")))
         else:
             ctx = context()
-            verify_lock(ctx)
-            if args.operation == "verify-target":
+            members = verify_lock(ctx)
+            if args.operation in ("verify-target", "may-clean"):
                 verify_target(inventory(), ctx["ref"], args.group or "")
+                if args.operation == "may-clean":
+                    allowed = mutation_allowed(ctx, members, args.commit_time, args.inactive_hours)
+                    # Recheck ownership and queued work after reading history.
+                    if allowed:
+                        allowed = mutation_allowed(ctx, verify_lock(ctx))
+                    print("proceed" if allowed else "defer")
             else:
                 verify_account()
+                if args.operation == "activity-time":
+                    print(int(branch_activity(ctx, args.commit_time)))
     except (RecoveryError, ValueError, OSError, subprocess.SubprocessError) as error:
         print(f"ERROR: CI cleanup refused: {error}", file=sys.stderr)
         return 1
diff --git a/devops/scripts/clean_ci_validation_envs.sh b/devops/scripts/clean_ci_validation_envs.sh
index 48ebe1993..2e6287d0a 100755
--- a/devops/scripts/clean_ci_validation_envs.sh
+++ b/devops/scripts/clean_ci_validation_envs.sh
@@ -14,7 +14,13 @@ set -o nounset
 
 function stopEnv ()
 {
-  python3 devops/scripts/ci_cleanup_scope.py verify-target --group "$1"
+  local decision
+  decision=$(python3 devops/scripts/ci_cleanup_scope.py may-clean --group "$1" "${@:2}")
+  if [[ "$decision" == defer ]]; then
+    echo "Validation activity defers stopping $1."
+    return 0
+  fi
+  [[ "$decision" == proceed ]] || { echo "Invalid cleanup activity decision" >&2; exit 1; }
   local tre_rg="$1"
   if [[ "$tre_rg" == *-mgmt ]]; then
     echo "Management-only environment ${tre_rg} has no core services to stop. Keeping it until the destroy threshold."
@@ -26,7 +32,13 @@ function stopEnv ()
 
 function destroyEnv ()
 {
-  python3 devops/scripts/ci_cleanup_scope.py verify-target --group "$1"
+  local decision
+  decision=$(python3 devops/scripts/ci_cleanup_scope.py may-clean --group "$1" "${@:2}")
+  if [[ "$decision" == defer ]]; then
+    echo "Validation activity defers destroying $1."
+    return 0
+  fi
+  [[ "$decision" == proceed ]] || { echo "Invalid cleanup activity decision" >&2; exit 1; }
   # The destroy helper accepts the core name even when only management exists.
   devops/scripts/destroy_env_no_terraform.sh --core-tre-rg "${1%-mgmt}"
 }
@@ -146,14 +158,16 @@ while read -r rg_name rg_ref_name; do
        # checking when was the last commit on the branch.
       last_commit_date_string=$(git for-each-ref --sort='-committerdate:iso8601' --format=' %(committerdate:iso8601)%09%(refname)' "${ref_in_remote}" | cut -f1)
       echo "Native ref is ${rg_ref_name}, last commit was on: ${last_commit_date_string}"
-      diff_in_hours=$(( ($(date +%s) - $(date -d "${last_commit_date_string}" +%s) )/(60*60) ))
+      last_commit_time=$(date -d "${last_commit_date_string}" +%s)
+      last_activity_time=$(python3 devops/scripts/ci_cleanup_scope.py activity-time --commit-time "$last_commit_time")
+      diff_in_hours=$(( ($(date +%s) - last_activity_time )/(60*60) ))
 
       if (( diff_in_hours > BRANCH_LAST_ACTIVITY_IN_HOURS_FOR_DESTROY )); then
         echo "No recent activity on ${rg_ref_name}. Environment in ${rg_name} will be destroyed."
-        destroyEnv "${rg_name}"
+        destroyEnv "${rg_name}" --commit-time "$last_commit_time" --inactive-hours "$BRANCH_LAST_ACTIVITY_IN_HOURS_FOR_DESTROY"
       elif (( diff_in_hours > BRANCH_LAST_ACTIVITY_IN_HOURS_FOR_STOP )); then
         echo "No recent activity on ${rg_ref_name}. Environment in ${rg_name} will be stopped."
-        stopEnv "${rg_name}"
+        stopEnv "${rg_name}" --commit-time "$last_commit_time" --inactive-hours "$BRANCH_LAST_ACTIVITY_IN_HOURS_FOR_STOP"
       fi
     fi
   fi
diff --git a/devops/scripts/recover_bootstrap_lease.py b/devops/scripts/recover_bootstrap_lease.py
index 3300a1321..ca362123d 100644
--- a/devops/scripts/recover_bootstrap_lease.py
+++ b/devops/scripts/recover_bootstrap_lease.py
@@ -226,6 +226,7 @@ def verify_concurrency(ctx, job_id=None):
             and job.get("name") == "Deploy PR / Deploy Management",
             f"The concurrency job is not this recovery attempt's management deployment; target={target}.",
         )
+    return members
 
 
 def verify_workflows(ctx, cache=None):
diff --git a/devops/tests/test_ci_cleanup_activity.py b/devops/tests/test_ci_cleanup_activity.py
new file mode 100644
index 000000000..6464255ca
--- /dev/null
+++ b/devops/tests/test_ci_cleanup_activity.py
@@ -0,0 +1,178 @@
+"""Protect recently used environments without retaining unrelated or orphaned groups."""
+
+from copy import deepcopy
+from datetime import datetime, timezone
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+from urllib.parse import parse_qs, urlsplit
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
+import ci_cleanup_activity as activity  # noqa: E402
+
+
+class ActivityTests(unittest.TestCase):
+    def setUp(self):
+        self.now = datetime(2026, 10, 11, 12, tzinfo=timezone.utc).timestamp()
+        self.ctx = {"repository": "microsoft/AzureTRE", "ref": "refs/heads/feature/test", "run_id": 123}
+        self.members = [{"run_id": 123, "job_id": 10, "status": "in_progress"}]
+        self.runs = []
+        self.requests = []
+
+    def run_record(self, number=200, age_hours=1, **kwargs):
+        return {
+            "id": number,
+            "path": activity.BRANCH_WORKFLOW,
+            "event": "workflow_dispatch",
+            "head_branch": "feature/test",
+            "head_sha": "a" * 40,
+            "repository": {"full_name": self.ctx["repository"]},
+            "head_repository": {"full_name": self.ctx["repository"]},
+            "status": "completed",
+            "conclusion": "success",
+            "created_at": "2026-09-01T00:00:00Z",
+            "updated_at": datetime.fromtimestamp(self.now - age_hours * 3600, timezone.utc).isoformat(),
+            **kwargs,
+        }
+
+    def github(self, ctx, suffix):
+        self.requests.append(suffix)
+        self.assertEqual(ctx, self.ctx)
+        self.assertTrue(suffix.startswith("/actions/workflows/deploy_tre_branch.yml/runs?"))
+        query = parse_qs(urlsplit(suffix).query)
+        self.assertEqual(query["branch"], ["feature/test"])
+        self.assertEqual(query["event"], ["workflow_dispatch"])
+        self.assertNotIn("created", query)  # Older runs can have a recent rerun.
+        page = int(query["page"][0])
+        return {"total_count": len(self.runs), "workflow_runs": deepcopy(self.runs[(page - 1) * 100 : page * 100])}  # noqa: E203
+
+    def allowed(self, age=100, threshold=4):
+        with patch.object(activity, "github", side_effect=self.github):
+            return activity.mutation_allowed(self.ctx, self.members, int(self.now - age * 3600), threshold, self.now)
+
+    def test_old_commit_and_recent_success_failure_or_cancellation_are_retained(self):
+        for conclusion in ("success", "failure", "cancelled", "timed_out"):
+            with self.subTest(conclusion=conclusion):
+                self.runs = [self.run_record(conclusion=conclusion, age_hours=1 / 60)]
+                self.assertFalse(self.allowed())
+
+    def test_recent_rerun_of_old_sha_extends_activity(self):
+        self.runs = [self.run_record(run_attempt=4, head_sha="b" * 40)]
+        self.assertFalse(self.allowed())
+
+    def test_preparation_and_pending_runs_are_active_without_queue_membership(self):
+        for status in activity.ACTIVE_STATUSES:
+            with self.subTest(status=status):
+                self.runs = [self.run_record(status=status, age_hours=100)]
+                self.assertFalse(self.allowed())
+
+    def test_old_activity_preserves_thresholds_and_new_commits_extend_activity(self):
+        self.runs = [self.run_record(age_hours=6)]
+        self.assertTrue(self.allowed(threshold=4))
+        self.assertFalse(self.allowed(threshold=48))
+        self.assertFalse(self.allowed(age=1))
+        self.runs = [self.run_record(age_hours=60)]
+        self.assertTrue(self.allowed(threshold=48))
+        self.runs = []
+        self.assertTrue(self.allowed(threshold=48))
+
+    def test_second_page_recent_run_prevents_destruction(self):
+        self.runs = [self.run_record(number=n, age_hours=100) for n in range(200, 300)] + [self.run_record(number=300)]
+        self.assertFalse(self.allowed(threshold=48))
+        self.assertTrue(self.requests[-1].endswith("page=2"))
+
+    def test_unrelated_history_is_not_requested(self):
+        self.runs = []
+        self.assertTrue(self.allowed())
+        self.assertEqual(len(self.requests), 1)
+        self.assertIn("branch=feature%2Ftest", self.requests[0])
+
+    def test_unexpected_identity_status_or_time_refuses_cleanup(self):
+        for change in (
+            {"id": True},
+            {"path": activity.CLEANUP_WORKFLOW},
+            {"event": "issue_comment"},
+            {"head_branch": "other"},
+            {"repository": {"full_name": "other/repo"}},
+            {"head_repository": None},
+            {"status": "unknown"},
+            {"updated_at": None},
+            {"updated_at": "not-a-date"},
+            {"updated_at": "2026-10-11T11:00:00"},
+            {"updated_at": "2026-10-12T11:00:00Z"},
+            {"updated_at": "2026-08-01T11:00:00Z"},
+        ):
+            with self.subTest(change=change), self.assertRaises(activity.RecoveryError):
+                self.runs = [self.run_record(age_hours=100, **change)]
+                self.allowed()
+
+    def test_incomplete_duplicate_or_unbounded_history_refuses_cleanup(self):
+        for response in (
+            {},
+            {"total_count": 1001, "workflow_runs": []},
+            {"total_count": True, "workflow_runs": []},
+            {"total_count": 2, "workflow_runs": [self.run_record(age_hours=100)]},
+            {"total_count": 2, "workflow_runs": [self.run_record(age_hours=100)] * 2},
+            {"total_count": 0, "workflow_runs": [self.run_record(age_hours=100)]},
+        ):
+            with (
+                self.subTest(response=response),
+                patch.object(activity, "github", return_value=response),
+                self.assertRaises(activity.RecoveryError),
+            ):
+                activity.mutation_allowed(self.ctx, self.members, int(self.now - 100 * 3600), 4, self.now)
+
+    def test_changed_page_count_and_api_error_refuse_cleanup(self):
+        first = {
+            "total_count": 101,
+            "workflow_runs": [self.run_record(number=n, age_hours=100) for n in range(200, 300)],
+        }
+        for responses in ([first, {"total_count": 102, "workflow_runs": []}], activity.RecoveryError("rate limit")):
+            with patch.object(activity, "github", side_effect=responses), self.assertRaises(activity.RecoveryError):
+                activity.mutation_allowed(self.ctx, self.members, int(self.now - 100 * 3600), 4, self.now)
+
+    def test_queued_deployment_or_unknown_work_defers_even_orphan_deletion(self):
+        self.members.append({"run_id": 456, "status": "pending"})
+        for path in (activity.BRANCH_WORKFLOW, ".github/workflows/pr_comment_bot.yml", None, activity.CLEANUP_WORKFLOW):
+            run = {"id": 456, "path": path, "event": "schedule", "repository": {"full_name": self.ctx["repository"]}}
+            with patch.object(activity, "github", return_value=run):
+                self.assertFalse(activity.mutation_allowed(self.ctx, self.members))
+
+    def test_verified_cleanup_successors_do_not_extend_activity(self):
+        self.members.append({"run_id": 456, "job_id": 50, "status": "pending"})
+        run = {
+            "id": 456,
+            "path": activity.CLEANUP_WORKFLOW,
+            "event": "schedule",
+            "run_attempt": 2,
+            "repository": {"full_name": self.ctx["repository"]},
+        }
+        job = {"id": 50, "run_id": 456, "run_attempt": 2, "name": "Clean " + self.ctx["ref"], "status": "pending"}
+        with patch.object(activity, "github", side_effect=[run, job]):
+            self.assertTrue(activity.mutation_allowed(self.ctx, self.members))
+        for change in ({"run_attempt": 1}, {"name": "Unidentified"}, {"id": 60}, {"status": "completed"}):
+            with patch.object(activity, "github", side_effect=[run, {**job, **change}]):
+                self.assertFalse(activity.mutation_allowed(self.ctx, self.members))
+
+    def test_unverified_queue_identity_refuses_cleanup(self):
+        self.members.append({"run_id": 456, "job_id": 50, "status": "pending"})
+        with patch.object(activity, "github", return_value={}), self.assertRaises(activity.RecoveryError):
+            activity.mutation_allowed(self.ctx, self.members)
+
+    def test_orphan_decision_has_no_completed_run_grace_period(self):
+        with patch.object(activity, "github") as github:
+            self.assertTrue(activity.mutation_allowed(self.ctx, self.members))
+            github.assert_not_called()
+
+    def test_partial_or_invalid_age_requests_refuse_cleanup(self):
+        for commit, hours in ((None, 4), (1, None), (-1, 4), (1, -1), (True, 4)):
+            with self.subTest(commit=commit, hours=hours), self.assertRaises(activity.RecoveryError):
+                activity.mutation_allowed(self.ctx, self.members, commit, hours, self.now)
+        self.ctx["ref"] = "refs/pull/123/merge"
+        with self.assertRaises(activity.RecoveryError):
+            activity.mutation_allowed(self.ctx, self.members, int(self.now), 4, self.now)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/devops/tests/test_ci_cleanup_scope.py b/devops/tests/test_ci_cleanup_scope.py
index cad8f9c74..138a0ddbe 100644
--- a/devops/tests/test_ci_cleanup_scope.py
+++ b/devops/tests/test_ci_cleanup_scope.py
@@ -1,7 +1,9 @@
 """Verify cleanup ownership, pagination and the shared workflow lock contract."""
 
 from copy import deepcopy
+from contextlib import redirect_stderr, redirect_stdout
 import importlib.util
+import io
 import json
 import os
 from pathlib import Path
@@ -15,6 +17,7 @@
 SPEC = importlib.util.spec_from_file_location("ci_cleanup_scope", SCRIPTS / "ci_cleanup_scope.py")
 scope = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(scope)
+import ci_cleanup_activity as activity  # noqa: E402
 
 
 class ScopeTests(unittest.TestCase):
@@ -125,6 +128,36 @@ def test_rate_limit_or_lock_failure_never_allows_cleanup(self):
             with self.assertRaises(scope.RecoveryError):
                 scope.verify_lock(self.ctx)
 
+    def test_new_queued_deployment_defers_command_after_history_check(self):
+        owner = [{"run_id": 123, "job_id": 10, "status": "in_progress"}]
+        successor = {"run_id": 456, "job_id": 11, "status": "pending"}
+        run = {"id": 456, "path": activity.BRANCH_WORKFLOW, "repository": {"full_name": self.ctx["repository"]}}
+        output = io.StringIO()
+        with (
+            patch.object(sys, "argv", ["ci_cleanup_scope.py", "may-clean", "--group", "rg-tretest"]),
+            patch.object(scope, "verify_lock", side_effect=[owner, owner + [successor]]) as lock,
+            patch.object(scope, "inventory", return_value={"rg-tretest": self.ctx["ref"]}),
+            patch.object(activity, "github", return_value=run),
+            redirect_stdout(output),
+        ):
+            self.assertEqual(scope.main(), 0)
+        self.assertEqual(lock.call_count, 2)
+        self.assertEqual(output.getvalue(), "defer\n")
+
+    def test_lost_second_lock_fails_command_without_permission_to_mutate(self):
+        owner = [{"run_id": 123, "job_id": 10, "status": "in_progress"}]
+        output, error = io.StringIO(), io.StringIO()
+        with (
+            patch.object(sys, "argv", ["ci_cleanup_scope.py", "may-clean", "--group", "rg-tretest"]),
+            patch.object(scope, "verify_lock", side_effect=[owner, scope.RecoveryError("owner changed")]),
+            patch.object(scope, "inventory", return_value={"rg-tretest": self.ctx["ref"]}),
+            redirect_stdout(output),
+            redirect_stderr(error),
+        ):
+            self.assertEqual(scope.main(), 1)
+        self.assertEqual(output.getvalue(), "")
+        self.assertIn("owner changed", error.getvalue())
+
     def test_unscoped_non_ci_and_invalid_references_are_refused(self):
         for key, value in (
             ("CI_CLEANUP_REF", ""),
@@ -220,7 +253,7 @@ def test_cleanup_waits_until_all_resource_group_deletions_complete(self):
         self.assertNotIn("--no-wait", script)
         self.assertNotIn("actions/runs?", script)
         self.assertLess(script.index("ci_cleanup_scope.py verify-lock"), script.index("az group list"))
-        self.assertRegex(script, re.compile(r"function destroyEnv.*?verify-target.*?destroy_env_no_terraform", re.S))
+        self.assertRegex(script, re.compile(r"function destroyEnv.*?may-clean.*?destroy_env_no_terraform", re.S))
 
     def test_cleanup_job_keeps_the_lock_for_long_deletions(self):
         scheduled = (SCRIPTS.parents[1] / ".github/workflows/clean_validation_envs.yml").read_text()
diff --git a/devops/tests/test_clean_ci_validation_envs.py b/devops/tests/test_clean_ci_validation_envs.py
index b011e29af..52bdd2971 100644
--- a/devops/tests/test_clean_ci_validation_envs.py
+++ b/devops/tests/test_clean_ci_validation_envs.py
@@ -29,6 +29,12 @@
 if command == "python3":
     if config.get("guard_error") or not os.environ.get("CI_CLEANUP_REF"):
         sys.exit("CI cleanup refused: exclusive ownership unavailable")
+    if "may-clean" in args:
+        if config.get("activity_error"):
+            sys.exit("CI cleanup refused: history incomplete")
+        print(config.get("activity_decision", "proceed"))
+    elif "activity-time" in args:
+        print(2000000000 - config.get("activity_age_hours", config["age_hours"]) * 3600)
     sys.exit(0)
 elif command == "gh":
     if args[0] == "api":
@@ -257,6 +263,40 @@ def test_scoped_cleanup_ignores_unrelated_workflows_and_other_targets(self):
         self.assertEqual(self.environment_actions(), [["destroy_env_no_terraform.sh", "--core-tre-rg", "rg-tretest"]])
         self.assertFalse(any(call[0] == "gh" and any("actions/runs?" in item for item in call) for call in self.calls))
 
+    def test_native_branch_uses_latest_activity_for_stop_and_destroy(self):
+        self.config.update(groups=["rg-tretest\trefs/heads/feature/test"], age_hours=100)
+        for hours, action in ((1, None), (6, "control_tre.sh"), (60, "destroy_env_no_terraform.sh")):
+            with self.subTest(hours=hours):
+                self.config["activity_age_hours"] = hours
+                result = self.run_cleanup()
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual([call[0] for call in self.environment_actions()], [] if action is None else [action])
+                if action:
+                    guards = [call for call in self.calls if "may-clean" in call]
+                    self.assertEqual(len(guards), 1)
+                    self.assertIn("--commit-time", guards[0])
+                    self.assertIn("--inactive-hours", guards[0])
+
+    def test_activity_recheck_defers_old_and_orphaned_environments(self):
+        for ref, missing, prs in (
+            ("refs/pull/5085/merge", False, self.config["open_prs"]),
+            ("refs/pull/5085/merge", False, []),
+            ("refs/heads/feature/deleted", True, []),
+        ):
+            with self.subTest(ref=ref, missing=missing, prs=prs):
+                self.config.update(
+                    groups=["rg-tretest\t" + ref], missing_branch=missing, open_prs=prs, activity_decision="defer"
+                )
+                result = self.run_cleanup()
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(self.environment_actions(), [])
+
+    def test_activity_api_failure_prevents_mutation(self):
+        self.config["activity_error"] = True
+        result = self.run_cleanup()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(self.environment_actions(), [])
+
     def test_main_workspace_cleanup_is_separate_and_synchronous(self):
         self.config["selected_ref"] = True
         self.env["CI_CLEANUP_REF"] = "refs/heads/main"
diff --git a/devops/tests/test_validate_ci_cleanup.py b/devops/tests/test_validate_ci_cleanup.py
index 670349bc0..a7c5c42e7 100644
--- a/devops/tests/test_validate_ci_cleanup.py
+++ b/devops/tests/test_validate_ci_cleanup.py
@@ -123,6 +123,7 @@ def values(name):
     os.execv(sys.executable, [sys.executable, *sys.argv[1:]])
 sys.path.insert(0, str(Path(sys.argv[1]).resolve().parent))
 import ci_cleanup_scope as scope
+import ci_cleanup_activity as activity
 import recover_bootstrap_lease as recovery
 
 def github(ctx, suffix):
@@ -137,6 +138,7 @@ def github(ctx, suffix):
     raise AssertionError("Unexpected GitHub request: " + suffix)
 
 scope.github = github
+activity.github = github
 recovery.github = github
 sys.argv = sys.argv[1:]
 sys.exit(scope.main())
diff --git a/devops/version.txt b/devops/version.txt
index c98d6e334..2e77e1073 100644
--- a/devops/version.txt
+++ b/devops/version.txt
@@ -1 +1 @@
-__version__ = "0.6.13"
+__version__ = "0.6.14"

From 3b5ca6ba296db64cade5a0b3d19fa0f59a2c345b Mon Sep 17 00:00:00 2001
From: Jay W <git.jaydoubleu@gmail.com>
Date: Sun, 11 Oct 2026 15:10:53 +0100
Subject: [PATCH 2/3] Fix #5231: acquire the deployment lock before branch
 preparation

---
 .github/linters/actionlint.yml                |   3 +
 .../scripts/ci_environment_workflow.test.js   |  81 +++++++-
 .../workflows/build_validation_develop.yml    |   1 +
 .github/workflows/deploy_tre_branch.yml       |  10 +-
 .github/workflows/deploy_tre_reusable.yml     |  45 ++++-
 CHANGELOG.md                                  |   2 +-
 devops/scripts/verify_deployment_lock.py      |  81 ++++++++
 devops/tests/test_ci_cleanup_scope.py         |   5 +-
 devops/tests/test_verify_deployment_lock.py   | 180 ++++++++++++++++++
 9 files changed, 396 insertions(+), 12 deletions(-)
 create mode 100644 devops/scripts/verify_deployment_lock.py
 create mode 100644 devops/tests/test_verify_deployment_lock.py

diff --git a/.github/linters/actionlint.yml b/.github/linters/actionlint.yml
index 7a0c2e52f..daba70a4f 100644
--- a/.github/linters/actionlint.yml
+++ b/.github/linters/actionlint.yml
@@ -4,6 +4,9 @@
 # Remove these exact parser exceptions when actionlint supports queue.
 # ci_environment_workflow.test.js validates all reference group settings.
 paths:
+  .github/workflows/deploy_tre_branch.yml:
+    ignore:
+      - '^unexpected key "queue" for "concurrency" section\. expected one of "cancel-in-progress", "group"$'
   .github/workflows/clean_validation_envs.yml:
     ignore:
       - '^unexpected key "queue" for "concurrency" section\. expected one of "cancel-in-progress", "group"$'
diff --git a/.github/scripts/ci_environment_workflow.test.js b/.github/scripts/ci_environment_workflow.test.js
index 6500a5147..4853383be 100644
--- a/.github/scripts/ci_environment_workflow.test.js
+++ b/.github/scripts/ci_environment_workflow.test.js
@@ -3,6 +3,7 @@ const os = require('os');
 const path = require('path');
 const { execFileSync } = require('child_process');
 const yaml = require('js-yaml');
+const vm = require('vm');
 
 const root = path.resolve(__dirname, '../..');
 const workflow = name => yaml.load(fs.readFileSync(path.join(root, '.github/workflows', name), 'utf8'));
@@ -10,7 +11,6 @@ const workflow = name => yaml.load(fs.readFileSync(path.join(root, '.github/work
 describe('reference concurrency queues', () => {
   test.each([
     ['deploy_tre.yml', 'run-deploy-tre-main', '${{ github.ref }}'],
-    ['deploy_tre_branch.yml', 'run-deploy-tre-not-main', '${{ github.ref }}'],
     ['pr_comment_bot.yml', 'run_test', '${{ needs.pr_comment.outputs.ciGitRef }}']
   ])('%s routes deployment into the shared queue without an outer queue', (file, job, ref) => {
     const caller = workflow(file);
@@ -18,6 +18,7 @@ describe('reference concurrency queues', () => {
     expect(caller.jobs[job].concurrency).toBeUndefined();
     expect(caller.jobs[job].uses).toBe('./.github/workflows/deploy_tre_reusable.yml');
     expect(caller.jobs[job].with.ciGitRef).toBe(ref);
+    expect(caller.jobs[job].with.callerOwnsDeploymentLock).toBeUndefined();
     // Preparation may run independently, but must not add another waiting group.
     const needs = caller.jobs[job].needs || [];
     for (const dependency of Array.isArray(needs) ? needs : [needs]) {
@@ -25,8 +26,30 @@ describe('reference concurrency queues', () => {
     }
   });
 
+  test('branch validation owns the reference queue before preparation', () => {
+    const caller = workflow('deploy_tre_branch.yml');
+    expect(caller.concurrency).toEqual({ group: 'deploy-${{ github.ref }}', 'cancel-in-progress': false, queue: 'max' });
+    const job = caller.jobs['run-deploy-tre-not-main'];
+    expect(job.uses).toBe('./.github/workflows/deploy_tre_reusable.yml');
+    expect(job.with.ciGitRef).toBe('${{ github.ref }}');
+    expect(job.with.callerOwnsDeploymentLock).toBe(true);
+    expect(Object.values(caller.jobs).every(value => value.concurrency === undefined)).toBe(true);
+  });
+
+  test('delegation avoids a nested reference lock and preserves the default queue', () => {
+    const reusable = workflow('deploy_tre_reusable.yml');
+    expect(reusable.on.workflow_call.inputs.callerOwnsDeploymentLock).toMatchObject({
+      type: 'boolean', required: false, default: false
+    });
+    expect(reusable.concurrency).toEqual({
+      group: "${{ inputs.callerOwnsDeploymentLock && format('deploy-delegated-{0}-{1}', github.run_id, github.run_attempt) " +
+        "|| format('deploy-{0}', inputs.ciGitRef) }}",
+      'cancel-in-progress': false,
+      queue: 'max'
+    });
+  });
+
   test.each([
-    ['deploy_tre_reusable.yml', null, 'deploy-${{ inputs.ciGitRef }}'],
     ['pr_comment_bot.yml', 'destroy_pr_env', 'deploy-${{ needs.pr_comment.outputs.ciGitRef }}'],
     ['pr_comment_bot.yml', 'destroy_branch_env', 'deploy-${{ needs.pr_comment.outputs.branchCiGitRef }}'],
     ['clean_validation_envs.yml', 'clean', 'deploy-${{ matrix.ref }}'],
@@ -53,6 +76,60 @@ describe('reference concurrency queues', () => {
   });
 });
 
+function e2eCondition(job, lockResult, skipDeployment) {
+  const needs = job.needs.map(name => name === 'verify_deployment_lock' ? lockResult : (skipDeployment ? 'skipped' : 'success'));
+  const expression = job.if.trim().slice(3, -2)
+    .replaceAll('always()', 'true')
+    .replaceAll('inputs.e2eTestsCustomSelector', JSON.stringify('airlock'))
+    .replaceAll('inputs.skipDeployment', JSON.stringify(skipDeployment))
+    .replaceAll('needs.verify_deployment_lock.result', JSON.stringify(lockResult))
+    .replace(/contains\(needs\.\*\.result, '([^']+)'\)/g, (_, result) => JSON.stringify(needs.includes(result)));
+  return vm.runInNewContext(expression, {}, { timeout: 100 });
+}
+
+describe('delegated deployment guard', () => {
+  test('the guard reads workflow source and has no Azure credentials', () => {
+    const job = workflow('deploy_tre_reusable.yml').jobs.verify_deployment_lock;
+    expect(job.permissions).toEqual({ actions: 'read', contents: 'read' });
+    expect(job.steps[0].if).toBe('inputs.callerOwnsDeploymentLock');
+    expect(job.steps[0].with).toEqual({ 'persist-credentials': false, ref: '${{ github.sha }}' });
+    const check = job.steps.find(step => step.run?.includes('verify_deployment_lock.py'));
+    expect(check.if).toBe('inputs.callerOwnsDeploymentLock');
+    expect(check.env).toEqual({
+      CI_DEPLOYMENT_LOCK_DELEGATED: '${{ inputs.callerOwnsDeploymentLock }}',
+      CI_GIT_REF: '${{ inputs.ciGitRef }}',
+      GITHUB_TOKEN: '${{ github.token }}'
+    });
+    const fallback = job.steps.find(step => step.if === '${{ !inputs.callerOwnsDeploymentLock }}');
+    expect(fallback.run).toBe('echo "The reusable workflow holds the deployment reference lock."');
+  });
+
+  test('every Azure job depends on the guard', () => {
+    const jobs = workflow('deploy_tre_reusable.yml').jobs;
+    expect(Object.keys(jobs).filter(name => !jobs[name].needs)).toEqual(['verify_deployment_lock']);
+    expect(jobs.deploy_management.needs).toEqual(['verify_deployment_lock']);
+    for (const name of ['e2e_tests_smoke', 'e2e_tests_custom']) {
+      expect(jobs[name].needs).toContain('verify_deployment_lock');
+    }
+  });
+
+  test.each(['success', 'failure', 'cancelled', 'skipped'])('E2E requires guard success when its result is %s', result => {
+    const jobs = workflow('deploy_tre_reusable.yml').jobs;
+    for (const name of ['e2e_tests_smoke', 'e2e_tests_custom']) {
+      for (const skipDeployment of [true, false]) {
+        expect(e2eCondition(jobs[name], result, skipDeployment)).toBe(result === 'success');
+      }
+    }
+  });
+
+  test('guard changes select the DevOps tests', () => {
+    const filter = workflow('build_validation_develop.yml').jobs;
+    const steps = Object.values(filter).flatMap(job => job.steps || []);
+    const paths = yaml.load(steps.find(step => step.id === 'filter').with.filters);
+    expect(paths.ci_cleanup).toContain('devops/scripts/verify_deployment_lock.py');
+  });
+});
+
 function prepare(job, location, ref, output) {
   const step = job.steps.find(value => value.id === 'environment-id' || value.id === 'run-id');
   execFileSync('bash', ['-c', step.run], {
diff --git a/.github/workflows/build_validation_develop.yml b/.github/workflows/build_validation_develop.yml
index b88d6c0fa..3a54f57f1 100644
--- a/.github/workflows/build_validation_develop.yml
+++ b/.github/workflows/build_validation_develop.yml
@@ -75,6 +75,7 @@ jobs:
               - 'devops/scripts/ci_environment_id.py'
               - 'devops/scripts/ci_cleanup_scope.py'
               - 'devops/scripts/ci_cleanup_activity.py'
+              - 'devops/scripts/verify_deployment_lock.py'
               - 'devops/scripts/destroy_ci_ref_env.py'
               - 'devops/scripts/destroy_env_no_terraform.sh'
               - 'devops/scripts/validate_ci_cleanup.py'
diff --git a/.github/workflows/deploy_tre_branch.yml b/.github/workflows/deploy_tre_branch.yml
index c37788e1a..cd1d01cb5 100644
--- a/.github/workflows/deploy_tre_branch.yml
+++ b/.github/workflows/deploy_tre_branch.yml
@@ -35,8 +35,13 @@ on: # yamllint disable-line rule:truthy
         default: false
         required: false
 
-# The reusable deployment workflow holds the shared reference queue.
-# Separate caller queues can discard deployments before they reach that group.
+# Register branch validation before Preparation can start. The reusable workflow
+# delegates this reference lock to the caller. Older workflow reruns keep their
+# original locking behaviour; dispatch this workflow to use the early lock.
+concurrency:
+  group: deploy-${{ github.ref }}
+  cancel-in-progress: false
+  queue: max
 
 permissions:
   contents: read
@@ -90,6 +95,7 @@ jobs:
       actions: read # Required by the reusable management job
     with:
       ciGitRef: ${{ github.ref }}
+      callerOwnsDeploymentLock: true
       ciEnvironmentId: ${{ needs.prepare-not-main.outputs.refid }}
       prHeadSha: ${{ github.sha }}
       e2eTestsCustomSelector: ${{ github.event.inputs.e2eTestsCustomSelector }}
diff --git a/.github/workflows/deploy_tre_reusable.yml b/.github/workflows/deploy_tre_reusable.yml
index 1a794f107..475da2707 100644
--- a/.github/workflows/deploy_tre_reusable.yml
+++ b/.github/workflows/deploy_tre_reusable.yml
@@ -18,6 +18,11 @@ on: # yamllint disable-line rule:truthy
         description: The git ref to use in annotations to associate a deployment with the code that triggered it
         type: string
         required: true
+      callerOwnsDeploymentLock:
+        description: The branch caller holds the deployment reference lock for its entire workflow
+        type: boolean
+        required: false
+        default: false
       ciEnvironmentId:
         description: Isolated CI environment ID derived from the Git reference, Azure cloud and location
         type: string
@@ -128,10 +133,12 @@ on: # yamllint disable-line rule:truthy
         description: ""
         required: false
 
-# This will prevent multiple runs of this entire workflow.
-# We should NOT cancel in progress runs as that can destabilize the environment.
+# Other callers retain the reference lock here. A branch caller already holds
+# it, so use a run-specific inner group and verify the outer owner before work.
 concurrency:
-  group: deploy-${{ inputs.ciGitRef }}
+  group: >-
+    ${{ inputs.callerOwnsDeploymentLock && format('deploy-delegated-{0}-{1}', github.run_id, github.run_attempt)
+    || format('deploy-{0}', inputs.ciGitRef) }}
   cancel-in-progress: false
   queue: max
 
@@ -139,8 +146,34 @@ env:
   CI_ENVIRONMENT_ID: ${{ inputs.ciEnvironmentId }}
 
 jobs:
+  verify_deployment_lock:
+    name: Verify deployment lock
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    permissions:
+      actions: read
+      contents: read
+    steps:
+      - name: Checkout workflow source
+        if: inputs.callerOwnsDeploymentLock
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
+        with:
+          persist-credentials: false
+          ref: ${{ github.sha }}
+      - name: Verify the delegated reference lock
+        if: inputs.callerOwnsDeploymentLock
+        env:
+          CI_DEPLOYMENT_LOCK_DELEGATED: ${{ inputs.callerOwnsDeploymentLock }}
+          CI_GIT_REF: ${{ inputs.ciGitRef }}
+          GITHUB_TOKEN: ${{ github.token }}
+        run: python3 devops/scripts/verify_deployment_lock.py
+      - name: Use the reusable workflow reference lock
+        if: ${{ !inputs.callerOwnsDeploymentLock }}
+        run: echo "The reusable workflow holds the deployment reference lock."
+
   deploy_management:
     name: Deploy Management
+    needs: [verify_deployment_lock]
     if: ${{ !inputs.skipDeployment }}
     runs-on: ubuntu-latest
     permissions:
@@ -861,12 +894,13 @@ jobs:
     # always() lets this job still evaluate its guard and run against the existing PR environment.
     if: >-
       ${{ always() &&
+      needs.verify_deployment_lock.result == 'success' &&
       (inputs.skipDeployment || !contains(needs.*.result, 'skipped')) &&
       !contains(needs.*.result, 'failure') &&
       !contains(needs.*.result, 'cancelled') }}
     runs-on: ubuntu-latest
     environment: ${{ inputs.environmentName }}
-    needs: [deploy_shared_services, register_bundles, deploy_ui]
+    needs: [verify_deployment_lock, deploy_shared_services, register_bundles, deploy_ui]
     timeout-minutes: 10
     steps:
       - name: Checkout
@@ -911,13 +945,14 @@ jobs:
     # Same as smoke tests: skipDeployment intentionally bypasses deploy jobs but still runs E2E tests.
     if: >-
       ${{ always() && inputs.e2eTestsCustomSelector != '' &&
+      needs.verify_deployment_lock.result == 'success' &&
       (inputs.skipDeployment || !contains(needs.*.result, 'skipped')) &&
       !contains(needs.*.result, 'failure') &&
       !contains(needs.*.result, 'cancelled') }}
     runs-on: ubuntu-latest
     environment: ${{ inputs.environmentName }}
     needs:
-      [deploy_shared_services, register_bundles, register_user_resource_bundles]
+      [verify_deployment_lock, deploy_shared_services, register_bundles, register_user_resource_bundles]
     timeout-minutes: 300
     steps:
       - name: Checkout
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6dabe2c68..6b3e877a6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,7 +27,7 @@ See [Legacy Airlock & migration](docs/azure-tre-overview/airlock.md#legacy-airlo
 * Add weekly and manual builds for Dockerfiles and Porter bundles, with selected pull request builds. Retry a failed target once and report final results. ([#5088](https://github.com/microsoft/AzureTRE/issues/5088))
 
 BUG FIXES:
-* Defer CI environment cleanup for queued operations and recent native-branch validation activity. Preserve scoped ownership checks and orphan deletion. (`devops` 0.6.14, `build-scripts` 1.0.4) ([#5231](https://github.com/microsoft/AzureTRE/issues/5231))
+* Defer CI environment cleanup for queued operations and recent native-branch validation activity. Lock new branch validation runs before preparation and verify delegated deployment ownership. Preserve scoped ownership checks and orphan deletion. (`devops` 0.6.14, `build-scripts` 1.0.4) ([#5231](https://github.com/microsoft/AzureTRE/issues/5231))
 * Fix denial-of-service dependency alerts by updating `source-map-js` to 1.2.2 in the UI and removing transitive `sprintf-js` from workflow tests through an `argparse` 2 override for `js-yaml` 3. (`ui` 0.10.3, `build-scripts` 1.0.3) ([#5118](https://github.com/microsoft/AzureTRE/pull/5118))
 * Align the `create_aad_groups` schema and Porter defaults to `true` across all three workspace bundles, preventing an unrelated edit from unintentionally removing workspace role groups and app-role assignments.
   Add regression checks for each bundle. (`tre-workspace-base` 2.12.2, `tre-workspace-unrestricted` 0.15.1, `tre-workspace-airlock-import-review` 0.18.1) ([#5050](https://github.com/microsoft/AzureTRE/issues/5050))
diff --git a/devops/scripts/verify_deployment_lock.py b/devops/scripts/verify_deployment_lock.py
new file mode 100644
index 000000000..1c7119dce
--- /dev/null
+++ b/devops/scripts/verify_deployment_lock.py
@@ -0,0 +1,81 @@
+"""Verify that a branch caller holds the deployment lock before Azure work starts."""
+
+import os
+import re
+import sys
+
+from ci_environment_id import validate_ref
+from recover_bootstrap_lease import RecoveryError, github, require, verify_concurrency
+
+ACTIVE_RUN_STATUSES = frozenset(("in_progress", "queued", "pending", "waiting", "requested"))
+
+
+def context():
+    require(os.environ.get("GITHUB_ACTIONS") == "true", "Delegated deployment requires GitHub Actions.")
+    ref = os.environ.get("CI_GIT_REF", "")
+    validate_ref(ref)
+    require(
+        ref.startswith("refs/heads/") and ref == os.environ.get("GITHUB_REF"),
+        "The delegated deployment reference differs from the branch caller.",
+    )
+    repository = os.environ.get("GITHUB_REPOSITORY", "")
+    require(
+        re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9-]*/[A-Za-z0-9_.-]+", repository)
+        and repository.split("/")[-1] not in (".", ".."),
+        "Invalid deployment repository.",
+    )
+    values = [os.environ.get(key, "") for key in ("GITHUB_RUN_ID", "GITHUB_RUN_ATTEMPT")]
+    require(
+        all(value.isdecimal() and int(value) > 0 for value in values), "The deployment run identity is unavailable."
+    )
+    sha = os.environ.get("GITHUB_SHA", "")
+    require(re.fullmatch(r"[0-9a-f]{40}", sha), "The deployment workflow commit is unavailable.")
+    return {"ref": ref, "repository": repository, "run_id": int(values[0]), "attempt": int(values[1]), "sha": sha}
+
+
+def verify_delegated_lock(ctx):
+    run = github(ctx, f"/actions/runs/{ctx['run_id']}")
+    require(
+        isinstance(run, dict)
+        and type(run.get("id")) is int
+        and run["id"] == ctx["run_id"]
+        and type(run.get("run_attempt")) is int
+        and run["run_attempt"] == ctx["attempt"]
+        and run.get("status") in ACTIVE_RUN_STATUSES
+        and run.get("event") == "workflow_dispatch"
+        and run.get("path") == ".github/workflows/deploy_tre_branch.yml"
+        and run.get("head_branch") == ctx["ref"].removeprefix("refs/heads/")
+        and run.get("head_sha") == ctx["sha"]
+        and all(
+            isinstance(run.get(key), dict) and run[key].get("full_name") == ctx["repository"]
+            for key in ("repository", "head_repository")
+        ),
+        "The delegated deployment is not the current branch workflow run and attempt.",
+    )
+    # The parent run can be queued while a job executes. The canonical group's
+    # in-progress owner, checked below, proves that this run holds the lock.
+    members = verify_concurrency(ctx)
+    # A reusable or job-level lease does not protect caller preparation.
+    require(
+        all("job_id" not in member for member in members if member["status"] == "in_progress"),
+        "The branch caller does not hold a workflow-level deployment lock.",
+    )
+
+
+def main():
+    try:
+        delegated = os.environ.get("CI_DEPLOYMENT_LOCK_DELEGATED", "")
+        require(delegated in ("true", "false"), "The deployment lock delegation flag is invalid.")
+        if delegated == "false":
+            print("The reusable workflow holds the deployment reference lock.")
+            return 0
+        verify_delegated_lock(context())
+        print("The branch workflow owns the deployment reference lock for this run and attempt.")
+    except (RecoveryError, ValueError, OSError) as error:
+        print(f"ERROR: Deployment lock verification failed: {error}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/devops/tests/test_ci_cleanup_scope.py b/devops/tests/test_ci_cleanup_scope.py
index 138a0ddbe..112d893b6 100644
--- a/devops/tests/test_ci_cleanup_scope.py
+++ b/devops/tests/test_ci_cleanup_scope.py
@@ -130,7 +130,8 @@ def test_rate_limit_or_lock_failure_never_allows_cleanup(self):
 
     def test_new_queued_deployment_defers_command_after_history_check(self):
         owner = [{"run_id": 123, "job_id": 10, "status": "in_progress"}]
-        successor = {"run_id": 456, "job_id": 11, "status": "pending"}
+        # The updated caller queues the whole run before Preparation starts.
+        successor = {"run_id": 456, "status": "pending"}
         run = {"id": 456, "path": activity.BRANCH_WORKFLOW, "repository": {"full_name": self.ctx["repository"]}}
         output = io.StringIO()
         with (
@@ -236,7 +237,7 @@ class WorkflowContractTests(unittest.TestCase):
     def test_all_supported_writers_share_the_reference_group(self):
         workflows = SCRIPTS.parents[1] / ".github/workflows"
         deploy = (workflows / "deploy_tre_reusable.yml").read_text()
-        self.assertIn("group: deploy-${{ inputs.ciGitRef }}", deploy)
+        self.assertIn("|| format('deploy-{0}', inputs.ciGitRef)", deploy)
         self.assertIn("CI_GIT_REF: ${{ inputs.ciGitRef }}", deploy)
         explicit = (workflows / "pr_comment_bot.yml").read_text()
         for name in ("ciGitRef", "branchCiGitRef"):
diff --git a/devops/tests/test_verify_deployment_lock.py b/devops/tests/test_verify_deployment_lock.py
new file mode 100644
index 000000000..bedc98dad
--- /dev/null
+++ b/devops/tests/test_verify_deployment_lock.py
@@ -0,0 +1,180 @@
+"""Check delegated deployment ownership against GitHub run and queue responses."""
+
+from contextlib import redirect_stderr, redirect_stdout
+from copy import deepcopy
+import io
+import os
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "devops/scripts"))
+import recover_bootstrap_lease as recovery  # noqa: E402
+import verify_deployment_lock as guard  # noqa: E402
+
+
+class DelegatedLockTests(unittest.TestCase):
+    def setUp(self):
+        self.env = patch.dict(
+            os.environ,
+            {
+                "CI_DEPLOYMENT_LOCK_DELEGATED": "true",
+                "CI_GIT_REF": "refs/heads/feature/test",
+                "GITHUB_ACTIONS": "true",
+                "GITHUB_REF": "refs/heads/feature/test",
+                "GITHUB_REPOSITORY": "microsoft/AzureTRE",
+                "GITHUB_RUN_ID": "123",
+                "GITHUB_RUN_ATTEMPT": "2",
+                "GITHUB_SHA": "a" * 40,
+            },
+            clear=True,
+        )
+        self.env.start()
+        self.addCleanup(self.env.stop)
+        self.ctx = guard.context()
+        self.run = {
+            "id": 123,
+            "run_attempt": 2,
+            "status": "in_progress",
+            "event": "workflow_dispatch",
+            "path": ".github/workflows/deploy_tre_branch.yml",
+            "head_branch": "feature/test",
+            "head_sha": "a" * 40,
+            "repository": {"full_name": "microsoft/AzureTRE"},
+            "head_repository": {"full_name": "microsoft/AzureTRE"},
+        }
+        self.group = {
+            "group_name": "deploy-refs/heads/feature/test",
+            "total_count": 1,
+            "group_members": [{"run_id": 123, "status": "in_progress"}],
+        }
+        self.requests = []
+        for module in (guard, recovery):
+            mock = patch.object(module, "github", side_effect=self.github)
+            mock.start()
+            self.addCleanup(mock.stop)
+
+    def github(self, ctx, suffix):
+        self.assertEqual(ctx, self.ctx)
+        self.requests.append(suffix)
+        if suffix == "/actions/runs/123":
+            return deepcopy(self.run)
+        if suffix == "/actions/concurrency_groups/deploy-refs%2Fheads%2Ffeature%2Ftest":
+            return deepcopy(self.group)
+        if suffix == "/actions/jobs/456":
+            return {
+                "id": 456,
+                "run_id": 123,
+                "run_attempt": 2,
+                "status": "in_progress",
+                "name": "Deploy PR / Deploy Management",
+            }
+        self.fail(f"Unexpected GitHub request: {suffix}")
+
+    def test_current_branch_workflow_owner_is_accepted_with_waiting_cleanup(self):
+        self.group["group_members"].append({"run_id": 999, "job_id": 101, "status": "pending"})
+        self.group["total_count"] = 2
+        guard.verify_delegated_lock(self.ctx)
+        self.assertEqual(len(self.requests), 2)
+
+    def test_active_parent_run_status_uses_the_canonical_in_progress_owner(self):
+        for status in ("queued", "pending", "waiting", "requested"):
+            self.run["status"] = status
+            with self.subTest(status=status):
+                guard.verify_delegated_lock(self.ctx)
+
+    def test_wrong_run_attempt_source_or_repository_is_refused(self):
+        for key, value in (
+            ("id", 456),
+            ("id", True),
+            ("run_attempt", 1),
+            ("run_attempt", "2"),
+            ("status", "unknown"),
+            ("status", "completed"),
+            ("event", "issue_comment"),
+            ("path", ".github/workflows/pr_comment_bot.yml"),
+            ("head_branch", "another-branch"),
+            ("head_sha", "b" * 40),
+            ("repository", {"full_name": "another/repo"}),
+            ("head_repository", {"full_name": "another/repo"}),
+        ):
+            before = self.run[key]
+            self.run[key] = value
+            with self.subTest(key=key, value=value), self.assertRaises(guard.RecoveryError):
+                guard.verify_delegated_lock(self.ctx)
+            self.run[key] = before
+
+    def test_wrong_pending_missing_or_ambiguous_owner_is_refused(self):
+        for members in (
+            [],
+            [{"run_id": 999, "status": "in_progress"}],
+            [{"run_id": 123, "status": "pending"}],
+            [{"run_id": 123, "status": "in_progress"}, {"run_id": 999, "status": "in_progress"}],
+        ):
+            self.group["group_members"] = members
+            self.group["total_count"] = len(members)
+            with self.subTest(members=members), self.assertRaises(guard.RecoveryError):
+                guard.verify_delegated_lock(self.ctx)
+
+    def test_management_job_lease_cannot_substitute_for_caller_workflow_lease(self):
+        # The existing bootstrap verifier accepts this job owner. Delegation must
+        # reject it because it does not protect branch preparation.
+        self.group["group_members"][0]["job_id"] = 456
+        with self.assertRaisesRegex(guard.RecoveryError, "workflow-level"):
+            guard.verify_delegated_lock(self.ctx)
+        self.assertIn("/actions/jobs/456", self.requests)
+
+    def test_wrong_group_or_incomplete_queue_is_refused(self):
+        for key, value in (("group_name", "deploy-other"), ("total_count", 2)):
+            before = self.group[key]
+            self.group[key] = value
+            with self.subTest(key=key), self.assertRaises(guard.RecoveryError):
+                guard.verify_delegated_lock(self.ctx)
+            self.group[key] = before
+
+    def test_invalid_context_is_refused_before_any_request(self):
+        for key, value in (
+            ("GITHUB_ACTIONS", "false"),
+            ("CI_GIT_REF", "refs/pull/123/merge"),
+            ("CI_GIT_REF", "refs/heads/main"),
+            ("CI_GIT_REF", "refs/heads/another"),
+            ("GITHUB_REF", "refs/heads/another"),
+            ("GITHUB_REPOSITORY", "../another/repo"),
+            ("GITHUB_RUN_ID", ""),
+            ("GITHUB_RUN_ATTEMPT", "0"),
+            ("GITHUB_SHA", "missing"),
+        ):
+            with (
+                self.subTest(key=key),
+                patch.dict(os.environ, {key: value}),
+                self.assertRaises((guard.RecoveryError, ValueError)),
+            ):
+                guard.context()
+        self.assertEqual(self.requests, [])
+
+    def test_other_callers_keep_the_reusable_lock_without_delegation_checks(self):
+        output = io.StringIO()
+        with patch.dict(os.environ, {"CI_DEPLOYMENT_LOCK_DELEGATED": "false"}), redirect_stdout(output):
+            self.assertEqual(guard.main(), 0)
+        self.assertEqual(self.requests, [])
+        self.assertIn("reusable workflow holds", output.getvalue())
+
+    def test_invalid_delegation_flag_and_api_failure_fail_the_command(self):
+        for value in ("", "yes", "TRUE"):
+            with patch.dict(os.environ, {"CI_DEPLOYMENT_LOCK_DELEGATED": value}), redirect_stderr(io.StringIO()):
+                self.assertEqual(guard.main(), 1)
+        output, error = io.StringIO(), io.StringIO()
+        with (
+            patch.object(recovery, "github", side_effect=guard.RecoveryError("API unavailable")),
+            redirect_stdout(output),
+            redirect_stderr(error),
+        ):
+            self.assertEqual(guard.main(), 1)
+        self.assertEqual(output.getvalue(), "")
+        self.assertIn("API unavailable", error.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()

From 3aa149223fd1e8f8c8915eb624a274fa5a7621ec Mon Sep 17 00:00:00 2001
From: Jay W <git.jaydoubleu@gmail.com>
Date: Mon, 12 Oct 2026 22:32:48 +0100
Subject: [PATCH 3/3] Keep deployment guard checks compatible with ordered E2E
 jobs (#5231)

---
 .../scripts/ci_environment_workflow.test.js    | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/.github/scripts/ci_environment_workflow.test.js b/.github/scripts/ci_environment_workflow.test.js
index 4853383be..14cf61608 100644
--- a/.github/scripts/ci_environment_workflow.test.js
+++ b/.github/scripts/ci_environment_workflow.test.js
@@ -77,14 +77,16 @@ describe('reference concurrency queues', () => {
 });
 
 function e2eCondition(job, lockResult, skipDeployment) {
-  const needs = job.needs.map(name => name === 'verify_deployment_lock' ? lockResult : (skipDeployment ? 'skipped' : 'success'));
-  const expression = job.if.trim().slice(3, -2)
-    .replaceAll('always()', 'true')
-    .replaceAll('inputs.e2eTestsCustomSelector', JSON.stringify('airlock'))
-    .replaceAll('inputs.skipDeployment', JSON.stringify(skipDeployment))
-    .replaceAll('needs.verify_deployment_lock.result', JSON.stringify(lockResult))
-    .replace(/contains\(needs\.\*\.result, '([^']+)'\)/g, (_, result) => JSON.stringify(needs.includes(result)));
-  return vm.runInNewContext(expression, {}, { timeout: 100 });
+  const needs = Object.fromEntries(job.needs.map(name => [name, {
+    result: name === 'verify_deployment_lock' ? lockResult
+      : name.startsWith('e2e_tests_') ? 'success' : (skipDeployment ? 'skipped' : 'success')
+  }]));
+  const inputs = { e2eBundle: '', e2eTestsCustomSelector: 'airlock', skipDeployment };
+  const results = Object.values(needs).map(value => value.result);
+  const expression = job.if.trim().slice(3, -2).replaceAll('needs.*.result', 'results');
+  return vm.runInNewContext(expression, {
+    needs, inputs, results, always: () => true, contains: (values, value) => values.includes(value)
+  }, { timeout: 100 });
 }
 
 describe('delegated deployment guard', () => {