diff --git a/.github-staging/workflows/test-integration.yml b/.github-staging/workflows/test-integration.yml
new file mode 100644
index 0000000000..2a736cbb0e
--- /dev/null
+++ b/.github-staging/workflows/test-integration.yml
@@ -0,0 +1,181 @@
+name: Integration Tests
+
+on:
+  workflow_call:
+    outputs:
+      passed:
+        description: 'Whether all integration tests passed'
+        value: ${{ jobs.aggregate.outputs.passed }}
+  workflow_dispatch:  # Allow manual testing
+
+jobs:
+  integration:
+    name: Integration Tests
+    runs-on: ubuntu-latest
+    # The 30-min budget set on the caller (.github/workflows/test.yml)
+    # is repeated here as defense in depth so the budget is enforced
+    # whether this workflow is invoked via `uses:` or via
+    # `workflow_dispatch`.
+    timeout-minutes: 30
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.14"
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Install dependencies
+        run: uv sync --extra dev
+
+      - name: Build containers
+        run: |
+          docker build -t egg-gateway -f gateway/Dockerfile .
+          docker build -t egg-sandbox -f sandbox/Dockerfile .
+
+      - name: Set up k3s
+        run: |
+          curl -sfL https://get.k3s.io | INSTALL_K3S_EXEC="--flannel-backend=none --disable-network-policy --write-kubeconfig-mode=644" sh -
+          export KUBECONFIG=/etc/rancher/k3s/k3s.yaml
+          echo "KUBECONFIG=/etc/rancher/k3s/k3s.yaml" >> "$GITHUB_ENV"
+          # Install Calico CNI
+          scripts/install-calico.sh
+          # Wait for node to be ready
+          kubectl wait --for=condition=Ready node --all --timeout=120s
+
+      - name: Import images into k3s
+        # Flake guard (HITL Q1 of #2474): retry the image-import step
+        # up to 3 attempts with a short backoff. Transient k3s
+        # containerd import failures (rare but observed in #2556's
+        # early runs) should not flake the entire integration tier.
+        # `set -o pipefail` makes a `docker save` failure on the
+        # left side of the pipe propagate into the `if`'s test
+        # result (rather than being swallowed by an exit-0
+        # `k3s ctr images import` on the right side reading an
+        # empty stream); the surrounding `if … then exit 0 … fi`
+        # then keeps `set -e` suspended for the pipeline so a
+        # failure falls through to the next retry instead of
+        # aborting the script.
+        run: |
+          set -eo pipefail
+          attempt=0
+          max_attempts=3
+          until [ "$attempt" -ge "$max_attempts" ]; do
+            attempt=$((attempt + 1))
+            echo "::group::Image import attempt ${attempt}/${max_attempts}"
+            if docker save egg-gateway:latest | sudo k3s ctr images import - \
+              && docker save egg-sandbox:latest | sudo k3s ctr images import -; then
+              echo "::endgroup::"
+              echo "Image import succeeded on attempt ${attempt}"
+              exit 0
+            fi
+            echo "::endgroup::"
+            echo "Image import failed on attempt ${attempt}"
+            if [ "$attempt" -lt "$max_attempts" ]; then
+              echo "Sleeping 5s before retry..."
+              sleep 5
+            fi
+          done
+          echo "Image import failed after ${max_attempts} attempts" >&2
+          exit 1
+
+      - name: Deploy egg to k3s
+        run: |
+          kubectl apply -k k8s/overlays/local/
+          kubectl -n egg-system wait --for=condition=Available deployment/egg-gateway --timeout=120s
+
+      - name: Run integration and security tests
+        env:
+          EGG_RUNTIME: kubernetes
+          KUBECONFIG: /etc/rancher/k3s/k3s.yaml
+        run: |
+          PYTHONPATH=shared .venv/bin/pytest integration_tests -v \
+            -m "integration or security" \
+            --timeout=300
+
+      - name: Collect k3s debug artifacts on failure
+        # Flake guard (HITL Q1 of #2474): on any prior-step failure,
+        # capture cluster-wide events and per-pod logs so the
+        # debugger doesn't have to reproduce locally to triage a CI
+        # flake. Uploaded as a single `k3s-debug` workflow artifact.
+        # NB: `kubectl logs` requires an explicit pod name or a
+        # non-empty label selector - there is no "all pods in
+        # namespace" primitive, so we enumerate pods per namespace
+        # via `kubectl get pods -o name` and tail each individually.
+        if: failure()
+        run: |
+          set +e
+          echo "Collecting cluster events..."
+          kubectl get events --all-namespaces -o yaml > k3s-debug-events.yaml 2>&1 || true
+          echo "Collecting pod logs..."
+          {
+            for ns in egg-system egg-test-agents; do
+              echo "===== Namespace: ${ns} ====="
+              for pod in $(kubectl get pods -n "${ns}" -o name 2>/dev/null); do
+                echo "----- ${ns}/${pod} (current) -----"
+                kubectl logs -n "${ns}" "${pod}" --all-containers=true --tail=-1 --prefix=true 2>&1 || true
+                # CrashLoopBackOff is the exact failure mode this artifact
+                # is meant to triage, and `kubectl logs` without
+                # `--previous` returns ONLY the current container
+                # instance's logs — so the crashing instance's stderr
+                # (the part the debugger actually needs) is invisible.
+                # Add a second pass for the previous instance; `|| true`
+                # keeps the call safe for pods with no previous instance.
+                echo "----- ${ns}/${pod} (previous) -----"
+                kubectl logs -n "${ns}" "${pod}" --all-containers=true --previous --tail=-1 --prefix=true 2>&1 || true
+              done
+              echo
+              echo "===== Pods in ${ns} ====="
+              kubectl get pods -n "${ns}" -o wide 2>&1 || true
+              echo
+            done
+          } > k3s-debug-pods.log 2>&1 || true
+          ls -la k3s-debug-events.yaml k3s-debug-pods.log || true
+
+      - name: Upload k3s debug artifacts
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: k3s-debug
+          path: |
+            k3s-debug-events.yaml
+            k3s-debug-pods.log
+          if-no-files-found: warn
+          retention-days: 14
+
+      - name: Cleanup
+        if: always()
+        run: |
+          kubectl delete namespace egg-test-agents --ignore-not-found=true 2>/dev/null || true
+          kubectl delete namespace egg-system --ignore-not-found=true 2>/dev/null || true
+          /usr/local/bin/k3s-uninstall.sh 2>/dev/null || true
+
+  aggregate:
+    name: Aggregate Integration Test Results
+    runs-on: ubuntu-latest
+    if: always()
+    needs: [integration]
+    outputs:
+      passed: ${{ steps.check.outputs.passed }}
+    steps:
+      - name: Check all jobs passed
+        id: check
+        run: |
+          if [[ "${{ needs.integration.result }}" != "success" ]]; then
+            echo "passed=false" >> "$GITHUB_OUTPUT"
+            echo "Integration tests failed"
+            echo "  integration: ${{ needs.integration.result }}"
+            # Without `exit 1`, the failure branch falls through with a
+            # zero exit code and this aggregate job reports success even
+            # when the integration tier was red. For the reusable-workflow
+            # `uses:` path the caller's job-failure propagation usually
+            # also fails the parent's `integration` job, but the
+            # standalone `workflow_dispatch` path has no such backstop.
+            exit 1
+          else
+            echo "passed=true" >> "$GITHUB_OUTPUT"
+            echo "All integration tests passed"
+          fi
diff --git a/.github-staging/workflows/test.yml b/.github-staging/workflows/test.yml
new file mode 100644
index 0000000000..9bd3c528e7
--- /dev/null
+++ b/.github-staging/workflows/test.yml
@@ -0,0 +1,108 @@
+name: Test
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened]
+  workflow_call:
+    outputs:
+      passed:
+        description: 'Whether all tests passed'
+        value: ${{ jobs.aggregate.outputs.passed }}
+  workflow_dispatch:  # Allow manual testing
+
+concurrency:
+  group: test-${{ github.head_ref || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  unit:
+    name: Unit Tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.14"
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Install dependencies
+        run: uv sync --extra dev
+
+      - name: Run unit tests
+        run: |
+          make test-all PYTEST_ARGS="--cov=gateway --cov=shared --cov=sandbox --cov-report=term-missing --cov-fail-under=80"
+
+  security:
+    name: Security Scan
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.14"
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Install dependencies
+        run: uv sync --extra dev
+
+      - name: Run security scan
+        run: make security
+
+  integration:
+    name: Integration Tests
+    # Required-from-day-1 per decision-3 of #2474. The reusable
+    # workflow lives at .github/workflows/test-integration.yml after
+    # the human reviewer performs the pre-merge `git mv` from
+    # .github-staging/. Folded into `aggregate` below so the
+    # canonical required-check name stays `Test / aggregate`.
+    #
+    # `timeout-minutes` historically wasn't in the documented
+    # keyword set for `uses:` caller jobs in GitHub Actions, but is
+    # honored in practice. The test-integration.yml reusable
+    # workflow also carries per-job `timeout-minutes` as a defense
+    # in depth so the budget is enforced regardless.
+    uses: ./.github/workflows/test-integration.yml
+    timeout-minutes: 30
+
+  aggregate:
+    # Intentionally NO `name:` override here: the GitHub-rendered
+    # check name must be `Test / aggregate` (lowercase job-id) so it
+    # matches the canonical required-check name documented in
+    # decision-3 / `manual_steps` of #2474 ("repo admin flips
+    # `Test / aggregate` to required-for-merge"). Adding a `name:`
+    # override would render the check as `Test / <override>` and
+    # silently desync from the operator-flipped required-check.
+    runs-on: ubuntu-latest
+    if: always()
+    needs: [unit, security, integration]
+    outputs:
+      passed: ${{ steps.check.outputs.passed }}
+    steps:
+      - name: Check all jobs passed
+        id: check
+        run: |
+          if [[ "${{ needs.unit.result }}" != "success" || \
+                "${{ needs.security.result }}" != "success" || \
+                "${{ needs.integration.result }}" != "success" ]]; then
+            echo "passed=false" >> "$GITHUB_OUTPUT"
+            echo "Some tests failed"
+            echo "  unit: ${{ needs.unit.result }}"
+            echo "  security: ${{ needs.security.result }}"
+            echo "  integration: ${{ needs.integration.result }}"
+            # Without this `exit 1`, the failure branch falls through with
+            # a zero exit code and the aggregate job — and the canonical
+            # required-for-merge `Test / aggregate` check — would report
+            # success regardless of which tier was red.
+            exit 1
+          else
+            echo "passed=true" >> "$GITHUB_OUTPUT"
+            echo "All tests passed"
+          fi
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index 8e8f5576b8..04b2c5ed7f 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -156,6 +156,11 @@ jobs:
             echo "  docker: ${{ needs.docker.result }}"
             echo "  actions: ${{ needs.actions.result }}"
             echo "  custom-checks: ${{ needs.custom-checks.result }}"
+            # Without this `exit 1`, the failure branch falls through
+            # with a zero exit code and this aggregate would report
+            # success even when a lint tier was red — defeating any
+            # branch-protection rule that requires `Lint / aggregate`.
+            exit 1
           else
             echo "passed=true" >> "$GITHUB_OUTPUT"
             echo "All lint checks passed"
diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py
index 656297f5c3..0a49093db8 100644
--- a/orchestrator/routes/pipelines.py
+++ b/orchestrator/routes/pipelines.py
@@ -9041,6 +9041,35 @@ def _build_github_staging_manual_step(worktree_repo_path: Path) -> str:
     if not staged_paths:
         return ""
 
+    # Compute concrete move commands per staged file, choosing
+    # ``git mv`` vs ``git rm`` + ``git mv`` based on whether the target
+    # ``.github/<rest>`` already exists.  ``git mv`` refuses to
+    # overwrite an existing destination, so a template that always
+    # emits the plain form breaks for replacement scenarios (e.g.
+    # restaging an existing workflow).
+    staging_prefix = ".github-staging/"
+    target_prefix = ".github/"
+    mkdir_dirs: list[str] = []
+    move_cmds: list[str] = []
+    for rel in staged_paths:
+        if not rel.startswith(staging_prefix):
+            continue
+        rest = rel[len(staging_prefix) :]
+        target_rel = f"{target_prefix}{rest}"
+        target_dir = target_rel.rsplit("/", 1)[0] if "/" in rest else target_prefix.rstrip("/")
+        if target_dir and target_dir not in mkdir_dirs:
+            mkdir_dirs.append(target_dir)
+        target_abs = worktree_repo_path / target_rel
+        # ``Path.exists()`` follows symlinks and returns False for a
+        # broken link, so an existing-but-broken symlink would slip
+        # through the existence check and ``git mv`` would still refuse
+        # to overwrite it. ``Path.is_symlink()`` returns True regardless
+        # of whether the target resolves, so the disjunction catches
+        # regular files, valid symlinks, and broken symlinks.
+        if target_abs.is_symlink() or target_abs.exists():
+            move_cmds.append(f"git rm {target_rel}  # target exists; remove before mv")
+        move_cmds.append(f"git mv {rel} {target_rel}")
+
     lines = [
         "### Move staged `.github/` changes (auto-generated, issue #2508)",
         "",
@@ -9060,12 +9089,21 @@ def _build_github_staging_manual_step(worktree_repo_path: Path) -> str:
             "",
             "1. Review each staged file for correctness — these are proposed "
             "CI / repo-config changes that bypass the agent's normal sandbox.",
-            "2. Move each file from `.github-staging/<path>` to `.github/<path>`. For example:",
+            "2. Run the following to move each staged file into `.github/` "
+            "(commands below are pre-computed for this PR; replacement targets "
+            "are handled via `git rm` + `git mv` since `git mv` refuses to "
+            "overwrite an existing destination):",
             "   ```",
-            "   mkdir -p .github/workflows",
-            "   git mv .github-staging/workflows/test-e2e.yml .github/workflows/test-e2e.yml",
+        ]
+    )
+    for d in mkdir_dirs:
+        lines.append(f"   mkdir -p {d}")
+    for cmd in move_cmds:
+        lines.append(f"   {cmd}")
+    lines.extend(
+        [
             "   ```",
-            "   After the `git mv`, `.github-staging/` is no longer tracked "
+            "   After the moves, `.github-staging/` is no longer tracked "
             "by git (git doesn't track empty directories). Run "
             "`rm -rf .github-staging` locally if you want to clear any "
             "leftover empty subdirectories from your worktree.",
diff --git a/orchestrator/tests/test_auto_pr.py b/orchestrator/tests/test_auto_pr.py
index 21a54b9973..7403b52c83 100644
--- a/orchestrator/tests/test_auto_pr.py
+++ b/orchestrator/tests/test_auto_pr.py
@@ -525,6 +525,44 @@ def test_drops_step_when_only_symlinks_staged(self, tmp_path):
 
         assert "Move staged" not in body
 
+    def test_replacement_target_uses_git_rm_then_mv(self, tmp_path):
+        """When `.github/<rest>` already exists, emit `git rm` before `git mv`.
+
+        `git mv` refuses to overwrite an existing destination
+        (``fatal: destination exists … specify -f to overwrite``), so
+        the historic template that always emitted the plain form
+        broke for replacement scenarios (e.g. restaging an existing
+        workflow). The helper must detect occupied targets and emit
+        the `git rm`+`git mv` sequence so the documented procedure
+        actually runs cleanly.
+        """
+        pipeline = _make_pipeline()
+        # Staged file with a counterpart already living under `.github/`.
+        staging = tmp_path / ".github-staging" / "workflows"
+        staging.mkdir(parents=True)
+        (staging / "test.yml").write_text("name: test (new)\n")
+        existing = tmp_path / ".github" / "workflows"
+        existing.mkdir(parents=True)
+        (existing / "test.yml").write_text("name: test (old)\n")
+        # Also a brand-new staged file with no existing target — the
+        # rendered block should use plain `git mv` for that one.
+        (staging / "test-integration.yml").write_text("name: integration\n")
+
+        _title, body, _ = _build_pr_body(pipeline, tmp_path)
+
+        assert "git rm .github/workflows/test.yml" in body, (
+            "replacement target must be removed with `git rm` before "
+            "`git mv` (otherwise `git mv` aborts with 'destination "
+            "exists')"
+        )
+        assert "git mv .github-staging/workflows/test.yml .github/workflows/test.yml" in body
+        # New file (no existing target) gets the plain `git mv`, no `git rm`.
+        assert (
+            "git mv .github-staging/workflows/test-integration.yml "
+            ".github/workflows/test-integration.yml"
+        ) in body
+        assert "git rm .github/workflows/test-integration.yml" not in body
+
     def test_drops_step_when_staging_dir_is_symlink(self, tmp_path):
         """When `.github-staging` itself is a symlink, no step is emitted.
 
diff --git a/tests/config/test_workflows_structure.py b/tests/config/test_workflows_structure.py
new file mode 100644
index 0000000000..c527a692d4
--- /dev/null
+++ b/tests/config/test_workflows_structure.py
@@ -0,0 +1,414 @@
+"""Structural assertions over the CI `Test` workflow YAMLs.
+
+The repository's PR-level CI is two workflow files:
+
+* ``.github/workflows/test.yml`` — the ``Test`` workflow with
+  ``unit`` / ``security`` / ``integration`` jobs and an
+  ``aggregate`` job whose rendered check name is the canonical
+  ``Test / aggregate`` required-for-merge target.
+* ``.github/workflows/test-integration.yml`` — the reusable
+  integration workflow invoked by ``test.yml``'s ``integration``
+  job. Carries HITL-Q1 flake guards introduced in slice-2 of
+  #2474: image-import retry, explicit ``kubectl wait`` timeouts,
+  and an on-failure ``k3s-debug`` artifact capturing
+  ``kubectl get events --all-namespaces`` and pod logs.
+
+These tests guard the structural invariants in perpetuity.
+During slice-2 of #2474 the files live under
+``.github-staging/workflows/`` (the coder role is gateway-blocked
+from ``.github/``, so the human reviewer performs the
+``git mv .github-staging/workflows/*.yml .github/workflows/``
+before merging slice-2's PR). The path resolver below prefers the
+staged copies when present so the same assertions cover both the
+pre-merge staging state AND the post-merge production state — and
+a future regression that drops the integration job from
+``.github/workflows/test.yml`` is caught by the same suite.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import pytest
+import yaml
+
+REPO_ROOT = Path(__file__).resolve().parent.parent.parent
+_STAGED_DIR = REPO_ROOT / ".github-staging" / "workflows"
+_FINAL_DIR = REPO_ROOT / ".github" / "workflows"
+
+
+def _resolve_workflow(filename: str) -> Path | None:
+    """Resolve a workflow file, preferring the staged copy.
+
+    Returns the path to ``.github-staging/workflows/<filename>`` if it
+    exists (slice-2 pre-merge state), else
+    ``.github/workflows/<filename>`` (post-merge production state),
+    else ``None`` so the fixture can skip cleanly.
+    """
+    staged = _STAGED_DIR / filename
+    if staged.exists():
+        return staged
+    final = _FINAL_DIR / filename
+    if final.exists():
+        return final
+    return None
+
+
+def _load_yaml(path: Path) -> dict:
+    """Parse a YAML file, leaving GitHub Actions' bare ``on:`` key as bool True.
+
+    PyYAML's safe_load maps the unquoted ``on:`` top-level key to the
+    Python boolean ``True``, not the string ``"on"``. Callers that
+    need the trigger block use ``data.get(True) or data.get("on")``.
+    """
+    with path.open(encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+@pytest.fixture
+def test_yml_path() -> Path:
+    resolved = _resolve_workflow("test.yml")
+    if resolved is None:
+        pytest.skip(
+            "neither .github-staging/workflows/test.yml nor "
+            ".github/workflows/test.yml found — repo is in an "
+            "unexpected state"
+        )
+    return resolved
+
+
+@pytest.fixture
+def test_yml(test_yml_path: Path) -> dict:
+    return _load_yaml(test_yml_path)
+
+
+@pytest.fixture
+def test_integration_yml_path() -> Path:
+    resolved = _resolve_workflow("test-integration.yml")
+    if resolved is None:
+        pytest.skip(
+            "neither .github-staging/workflows/test-integration.yml nor "
+            ".github/workflows/test-integration.yml found — repo is in "
+            "an unexpected state"
+        )
+    return resolved
+
+
+@pytest.fixture
+def test_integration_yml_text(test_integration_yml_path: Path) -> str:
+    return test_integration_yml_path.read_text(encoding="utf-8")
+
+
+@pytest.fixture
+def test_integration_yml(test_integration_yml_text: str) -> dict:
+    return yaml.safe_load(test_integration_yml_text)
+
+
+@pytest.mark.parametrize(
+    "workflow_filename",
+    [
+        pytest.param("test.yml", id="test.yml"),
+        pytest.param("test-integration.yml", id="test-integration.yml"),
+        pytest.param("lint.yml", id="lint.yml"),
+    ],
+)
+def test_aggregate_failure_branch_exits_nonzero(workflow_filename: str) -> None:
+    """Every aggregate-style gate's failure branch must ``exit 1``.
+
+    Every aggregate in the repo has the same shape: an ``if [[ … !=
+    success ]]; then …; else …; fi`` with ``passed=false`` /
+    ``passed=true`` written into ``$GITHUB_OUTPUT``. Under the default
+    GitHub Actions ``bash -eo pipefail`` shell, a failure branch that
+    only echoes diagnostics and falls through returns 0 — so the step
+    reports success, the aggregate job reports success, and a PR with
+    a red tier merges through the required check. The fix is a
+    trailing ``exit 1`` (or ``false`` — same exit code).
+
+    Covers all three aggregate gates simultaneously so the regression
+    cannot be reintroduced in any one of them without the suite
+    going red:
+
+    * ``test.yml::aggregate`` — the canonical ``Test / aggregate``
+      required-for-merge gate (slice-2 of #2474).
+    * ``test-integration.yml::aggregate`` — the reusable workflow's
+      internal aggregate (defense in depth for the
+      ``workflow_dispatch`` path, since ``uses:`` propagation usually
+      already fails the parent).
+    * ``lint.yml::aggregate`` — the ``Lint / aggregate`` gate, which
+      historically had the same fall-through bug.
+
+    Implementation note — the ``exit 1`` match is anchored to a
+    standalone-statement line via
+    ``re.search(r"^\\s*(exit\\s+1|false)\\s*$", …, re.MULTILINE)``,
+    NOT a plain substring check. The earlier substring version was
+    silently satisfied by the literal text ``exit 1`` inside the
+    warning comment immediately above the real ``exit 1`` statement
+    — so removing only the statement (a realistic regression mode
+    when a future developer "cleans up obsolete commentary") was not
+    caught. The anchored match requires the literal statement to be
+    the entire content of a line.
+    """
+    path = _resolve_workflow(workflow_filename)
+    if path is None:
+        pytest.skip(
+            f"neither .github-staging/workflows/{workflow_filename} nor "
+            f".github/workflows/{workflow_filename} found — repo is in "
+            "an unexpected state"
+        )
+    data = _load_yaml(path)
+    aggregate = data.get("jobs", {}).get("aggregate")
+    assert aggregate is not None, f"{workflow_filename}: missing `aggregate` job"
+    steps = aggregate.get("steps", [])
+    script_text = "\n".join(step.get("run", "") for step in steps if isinstance(step, dict))
+    # Locate the failure branch by anchoring on the ``passed=false``
+    # write into ``$GITHUB_OUTPUT`` (stable across all three
+    # aggregates) and consuming through to the ``else`` keyword.
+    failure_branch_match = re.search(
+        r"passed=false.*?(?=\belse\b)",
+        script_text,
+        re.DOTALL,
+    )
+    assert failure_branch_match is not None, (
+        f"{workflow_filename}: could not locate the aggregate's failure "
+        "branch — expected a `passed=false` write into $GITHUB_OUTPUT "
+        "followed by an `else` keyword"
+    )
+    failure_branch = failure_branch_match.group(0)
+    # Match ``exit 1`` (or ``false`` — same exit code, both fail the
+    # step under ``set -e``) as a STANDALONE statement on a line by
+    # itself. ``re.MULTILINE`` makes ``^`` / ``$`` line-anchored, so
+    # an ``exit 1`` inside a comment like ``# Without `exit 1`, …``
+    # does NOT satisfy this assertion.
+    assert re.search(r"^\s*(exit\s+1|false)\s*$", failure_branch, re.MULTILINE), (
+        f"{workflow_filename}: aggregate's failure branch does not "
+        "terminate with `exit 1` (or `false`) as a standalone "
+        "statement — the script falls through with a zero exit code "
+        "under `bash -eo pipefail` and the aggregate job reports "
+        "success even when a tier was red, defeating any "
+        "branch-protection rule that requires the aggregate check"
+    )
+
+
+class TestTestYmlStructure:
+    """Invariants over ``test.yml`` (staged or final).
+
+    Originally introduced as slice-2 task-2-1 acceptance scaffolding
+    for #2474; promoted to perpetual coverage so the post-merge
+    ``.github/workflows/test.yml`` is guarded by the same suite.
+    """
+
+    def test_integration_job_exists(self, test_yml: dict) -> None:
+        """A new ``integration`` job must be defined as a sibling of unit/security."""
+        jobs = test_yml.get("jobs", {})
+        assert "integration" in jobs, (
+            "missing `integration` job in test.yml — slice-2 task-2-1 "
+            "acceptance criterion: integration job sibling of "
+            "unit/security"
+        )
+        assert "unit" in jobs and "security" in jobs, (
+            "unit/security jobs missing — these are the historic "
+            "siblings of the new integration job"
+        )
+
+    def test_integration_job_uses_reusable_workflow(self, test_yml: dict) -> None:
+        """integration.uses must reference the reusable workflow path."""
+        integration = test_yml["jobs"]["integration"]
+        uses = integration.get("uses", "")
+        assert uses == "./.github/workflows/test-integration.yml", (
+            f"jobs.integration.uses={uses!r}; expected "
+            "'./.github/workflows/test-integration.yml' — the path "
+            "must reference the post-mv production location, never "
+            "`.github-staging/...` (the staged path is never invoked)"
+        )
+
+    def test_integration_job_has_30_minute_timeout(self, test_yml: dict) -> None:
+        """integration job must set ``timeout-minutes: 30`` (plan task-2-1 (b))."""
+        integration = test_yml["jobs"]["integration"]
+        assert integration.get("timeout-minutes") == 30, (
+            f"jobs.integration.timeout-minutes={integration.get('timeout-minutes')!r}; "
+            "expected 30 (plan task-2-1 (b))"
+        )
+
+    def test_aggregate_needs_includes_integration(self, test_yml: dict) -> None:
+        """aggregate.needs must contain unit, security, AND integration."""
+        aggregate = test_yml["jobs"]["aggregate"]
+        needs = aggregate.get("needs", [])
+        assert isinstance(needs, list), (
+            f"jobs.aggregate.needs is {type(needs).__name__}; expected list"
+        )
+        assert set(needs) == {"unit", "security", "integration"}, (
+            f"jobs.aggregate.needs={needs!r}; expected "
+            "['unit', 'security', 'integration'] (any order)"
+        )
+
+    def test_aggregate_check_inspects_integration_result(self, test_yml: dict) -> None:
+        """aggregate's check_all_passed script must reference needs.integration.result.
+
+        The if-all-passed check inspects every needed job's ``result``
+        so a red tier fails the aggregate. Slice-2 added the
+        integration tier; if a future change drops the
+        ``needs.integration.result`` reference, a red integration tier
+        would silently no-op the aggregate. (The companion check that
+        the failure branch actually ``exit 1``s lives in the
+        module-level parametrized
+        ``test_aggregate_failure_branch_exits_nonzero`` so it covers
+        ``test-integration.yml`` and ``lint.yml`` too.)
+        """
+        aggregate = test_yml["jobs"]["aggregate"]
+        steps = aggregate.get("steps", [])
+        script_text = "\n".join(step.get("run", "") for step in steps if isinstance(step, dict))
+        assert "needs.integration.result" in script_text, (
+            "aggregate job's check_all_passed script does not inspect "
+            "`needs.integration.result` — a red integration tier would "
+            "not fail the aggregate (plan task-2-1 (c))"
+        )
+
+    def test_workflow_call_output_passed_preserved(self, test_yml: dict) -> None:
+        """workflow_call output `passed` must remain so callers don't break.
+
+        PyYAML maps the bare GitHub Actions ``on:`` key to the Python
+        bool ``True``; access the trigger block via ``True`` so the
+        assertion runs against the actual loaded structure.
+        """
+        on_block = test_yml.get(True) or test_yml.get("on")
+        assert on_block is not None, "missing top-level `on:` block"
+        workflow_call = on_block.get("workflow_call")
+        assert workflow_call is not None, "workflow_call trigger removed"
+        outputs = workflow_call.get("outputs") or {}
+        assert "passed" in outputs, (
+            "workflow_call output `passed` missing — downstream callers "
+            "(e.g. branch protection aggregate) read this output"
+        )
+
+    def test_concurrency_block_preserved(self, test_yml: dict) -> None:
+        """Existing concurrency block must keep PR-scoped semantics.
+
+        The historical group is ``test-${{ github.head_ref ||
+        github.ref }}`` — one in-flight run per PR (head_ref) and one
+        per branch (ref) for non-PR triggers, with mid-flight
+        cancellation on a new push. A future regression that flipped
+        ``group`` to e.g. ``test-${{ github.run_id }}`` would give
+        every run a unique group, break PR concurrency entirely, and
+        an assertion of just ``"group" in concurrency`` would not
+        catch it. Require the group to actually reference
+        ``github.head_ref`` so the PR-scoping invariant is held.
+        """
+        concurrency = test_yml.get("concurrency") or {}
+        group = concurrency.get("group")
+        assert group, "concurrency.group removed"
+        assert "github.head_ref" in group, (
+            f"concurrency.group={group!r}; expected the group expression "
+            "to reference `github.head_ref` so concurrency is scoped "
+            "per PR. A group keyed on `github.run_id` or similar would "
+            "give every run a unique key and silently disable PR "
+            "concurrency."
+        )
+        assert concurrency.get("cancel-in-progress") is True, (
+            "concurrency.cancel-in-progress flipped to false — the "
+            "existing PR concurrency semantics must be preserved"
+        )
+
+
+class TestTestIntegrationYmlFlakeGuards:
+    """Invariants over ``test-integration.yml`` (staged or final).
+
+    Originally introduced as slice-2 task-2-2 acceptance scaffolding
+    for #2474; promoted to perpetual coverage so the HITL-Q1 flake
+    guards on the post-merge ``.github/workflows/test-integration.yml``
+    cannot silently regress.
+    """
+
+    def test_image_import_step_has_retry(self, test_integration_yml_text: str) -> None:
+        """`Import images into k3s` step must run inside a retry loop.
+
+        HITL Q1: 2-3 attempts with a sleep between attempts to absorb
+        transient image-import flakes. We accept any retry shape —
+        ``for i in 1 2 3``, ``until``, ``--retry`` flag — but require
+        evidence of both a retry loop and the image-import command.
+        """
+        text = test_integration_yml_text
+        assert "Import images into k3s" in text, (
+            "image-import step removed from test-integration.yml"
+        )
+        # Cheap heuristic: a `for` / `until` / `retry` token in the
+        # neighborhood of the image-import step. Capture the step
+        # body via a regex over the YAML text so we don't depend on
+        # YAML key ordering.
+        step_block_match = re.search(
+            r"-\s+name:\s+Import images into k3s.*?(?=\n\s*-\s+name:|\Z)",
+            text,
+            re.DOTALL,
+        )
+        assert step_block_match is not None, "could not locate `Import images into k3s` step body"
+        step_body = step_block_match.group(0)
+        retry_markers = ("for i in", "for attempt", "until ", "while ", "--retry")
+        assert any(m in step_body for m in retry_markers), (
+            f"`Import images into k3s` step has no retry loop "
+            f"(looked for any of {retry_markers!r}) — HITL Q1 flake guard"
+        )
+
+    def test_every_kubectl_wait_has_timeout(self, test_integration_yml_text: str) -> None:
+        """Every ``kubectl wait`` call must carry an explicit ``--timeout=`` flag."""
+        text = test_integration_yml_text
+        wait_lines = [
+            line.strip()
+            for line in text.splitlines()
+            if "kubectl" in line and "wait" in line and "--for" in line
+        ]
+        assert wait_lines, (
+            "no `kubectl wait` calls found in test-integration.yml — "
+            "either the workflow lost its readiness checks or this scan "
+            "missed them"
+        )
+        for line in wait_lines:
+            assert "--timeout=" in line, (
+                f"`kubectl wait` without `--timeout=` flag: {line!r} — "
+                "every wait must be deadline-bounded to defend against "
+                "hung-pod flakes (HITL Q1)"
+            )
+
+    def test_on_failure_artifact_upload_present(
+        self,
+        test_integration_yml: dict,
+        test_integration_yml_text: str,
+    ) -> None:
+        """An ``if: failure()`` step must upload a k3s-debug artifact.
+
+        Plan task-2-2 (c): on failure, capture ``kubectl get events
+        --all-namespaces -o yaml`` plus pod logs and upload as the
+        ``k3s-debug`` artifact via ``actions/upload-artifact@v4``.
+        """
+        text = test_integration_yml_text
+        assert "if: failure()" in text, (
+            "no `if: failure()` step in test-integration.yml — "
+            "HITL Q1 requires an on-failure diagnostic capture"
+        )
+        assert "kubectl get events --all-namespaces" in text, (
+            "on-failure step does not run `kubectl get events "
+            "--all-namespaces` — required for HITL Q1 k3s-debug artifact"
+        )
+        # Pod-log capture: tolerate either explicit `kubectl logs` or a
+        # script wrapper. Require evidence of a logs collection.
+        assert "kubectl logs" in text, (
+            "on-failure step does not capture pod logs via `kubectl logs` — "
+            "HITL Q1 requires pod logs in the k3s-debug artifact"
+        )
+        assert "actions/upload-artifact@v4" in text, (
+            "on-failure step does not upload the k3s-debug artifact via "
+            "`actions/upload-artifact@v4`"
+        )
+        assert "k3s-debug" in text, (
+            "uploaded artifact is not named `k3s-debug` — plan task-2-2 "
+            "(c) requires this exact artifact name so reviewers know "
+            "where to look"
+        )
+
+    def test_workflow_call_trigger_preserved(self, test_integration_yml: dict) -> None:
+        """workflow_call must remain so the parent `test.yml` can invoke it."""
+        on_block = test_integration_yml.get(True) or test_integration_yml.get("on")
+        assert on_block is not None, "missing top-level `on:` block"
+        assert "workflow_call" in on_block, (
+            "workflow_call trigger removed — test.yml's integration "
+            "job references this as a reusable workflow"
+        )