[WIP] Add CI job priority system

.github/workflows/benchmark-multinode-tmpl.yml

@@ -6,6 +6,19 @@ on:
       runner:
         required: true
         type: string
+      priority:
+        description: "Higher-is-sooner CI priority score"
+        required: true
+        type: string
+      queue-token:
+        description: "One-shot queue token generated for this job"
+        required: true
+        type: string
+      skip-queue-pr:
+        description: "Requested skip_queue PR number"
+        required: false
+        type: string
+        default: ''
       image:
         required: true
         type: string
@@ -231,10 +244,30 @@ permissions:
 
 jobs:
   benchmark:
-    runs-on: ${{ inputs.runner }}
+    runs-on: >-
+      ${{ fromJSON(
+        vars.PRIORITY_SCHEDULER_ENABLED == 'true' &&
+        (
+          inputs.skip-queue-pr != '' &&
+          format(
+            '["self-hosted",{0},{1},{2},{3}]',
+            toJSON(inputs.runner),
+            toJSON(format('ci-job-{0}-{1}', inputs.priority, inputs.queue-token)),
+            toJSON(format('ci-attempt-{0}', github.run_attempt)),
+            toJSON(format('ci-skip-queue-pr-{0}', inputs.skip-queue-pr))
+          ) ||
+          format(
+            '["self-hosted",{0},{1},{2}]',
+            toJSON(inputs.runner),
+            toJSON(format('ci-job-{0}-{1}', inputs.priority, inputs.queue-token)),
+            toJSON(format('ci-attempt-{0}', github.run_attempt))
+          )
+        ) ||
+        format('[{0}]', toJSON(inputs.runner))
+      ) }}
     timeout-minutes: 480
     name: >-
-      ${{ inputs.model-prefix }} ${{ inputs.precision }} ${{ inputs.runner }} ${{ inputs.framework == 'sglang' && 'sgl' || inputs.framework == 'dynamo-sglang' && 'dyn-sgl' || inputs.framework == 'sglang-disagg' && 'sgl-disagg' || inputs.framework }}
+      p${{ inputs.priority }} | ${{ inputs.model-prefix }} ${{ inputs.precision }} ${{ inputs.runner }} ${{ inputs.framework == 'sglang' && 'sgl' || inputs.framework == 'dynamo-sglang' && 'dyn-sgl' || inputs.framework == 'sglang-disagg' && 'sgl-disagg' || inputs.framework }}
       ${{ inputs.prefill-num-worker }}P (TP${{ inputs.prefill-tp }}${{ inputs.prefill-pp != '' && inputs.prefill-pp != '1' && format('/PP{0}', inputs.prefill-pp) || '' }}${{ inputs.prefill-dcp-size != '' && inputs.prefill-dcp-size != '1' && format('/DCP{0}', inputs.prefill-dcp-size) || '' }}${{ inputs.prefill-pcp-size != '' && inputs.prefill-pcp-size != '1' && format('/PCP{0}', inputs.prefill-pcp-size) || '' }}${{ inputs.prefill-ep != '' && inputs.prefill-ep != '1' && format('/EP{0}', inputs.prefill-ep) || '' }}${{ inputs.prefill-dp-attn == 'true' && '/DPA' || '' }})
       x ${{ inputs.decode-num-worker }}D (TP${{ inputs.decode-tp }}${{ inputs.decode-pp != '' && inputs.decode-pp != '1' && format('/PP{0}', inputs.decode-pp) || '' }}${{ inputs.decode-dcp-size != '' && inputs.decode-dcp-size != '1' && format('/DCP{0}', inputs.decode-dcp-size) || '' }}${{ inputs.decode-pcp-size != '' && inputs.decode-pcp-size != '1' && format('/PCP{0}', inputs.decode-pcp-size) || '' }}${{ inputs.decode-ep != '' && inputs.decode-ep != '1' && format('/EP{0}', inputs.decode-ep) || '' }}${{ inputs.decode-dp-attn == 'true' && '/DPA' || '' }})
       ${{ inputs.spec-decoding != 'none' && inputs.spec-decoding || '' }}

.github/workflows/benchmark-tmpl.yml

@@ -5,6 +5,19 @@ on:
       runner:
         required: true
         type: string
+      priority:
+        description: "Higher-is-sooner CI priority score"
+        required: true
+        type: string
+      queue-token:
+        description: "One-shot queue token generated for this job"
+        required: true
+        type: string
+      skip-queue-pr:
+        description: "Requested skip_queue PR number"
+        required: false
+        type: string
+        default: ''
       image:
         required: true
         type: string
@@ -161,10 +174,30 @@ permissions:
 
 jobs:
   benchmark:
-    runs-on: ${{ inputs.runner }}
+    runs-on: >-
+      ${{ fromJSON(
+        vars.PRIORITY_SCHEDULER_ENABLED == 'true' &&
+        (
+          inputs.skip-queue-pr != '' &&
+          format(
+            '["self-hosted",{0},{1},{2},{3}]',
+            toJSON(inputs.runner),
+            toJSON(format('ci-job-{0}-{1}', inputs.priority, inputs.queue-token)),
+            toJSON(format('ci-attempt-{0}', github.run_attempt)),
+            toJSON(format('ci-skip-queue-pr-{0}', inputs.skip-queue-pr))
+          ) ||
+          format(
+            '["self-hosted",{0},{1},{2}]',
+            toJSON(inputs.runner),
+            toJSON(format('ci-job-{0}-{1}', inputs.priority, inputs.queue-token)),
+            toJSON(format('ci-attempt-{0}', github.run_attempt))
+          )
+        ) ||
+        format('[{0}]', toJSON(inputs.runner))
+      ) }}
     timeout-minutes: 500
     name: >-
-      ${{ inputs.model-prefix }} ${{ inputs.precision }} ${{ inputs.runner }} ${{ inputs.framework == 'sglang' && 'sgl' || inputs.framework == 'dynamo-sglang' && 'dyn-sgl' || inputs.framework == 'sglang-disagg' && 'sgl-disagg' || inputs.framework }}
+      p${{ inputs.priority }} | ${{ inputs.model-prefix }} ${{ inputs.precision }} ${{ inputs.runner }} ${{ inputs.framework == 'sglang' && 'sgl' || inputs.framework == 'dynamo-sglang' && 'dyn-sgl' || inputs.framework == 'sglang-disagg' && 'sgl-disagg' || inputs.framework }}
       TP${{ inputs.tp }}${{ inputs.pp != '' && inputs.pp != '1' && format('/PP{0}', inputs.pp) || '' }}${{ inputs.dcp-size != '' && inputs.dcp-size != '1' && format('/DCP{0}', inputs.dcp-size) || '' }}${{ inputs.pcp-size != '' && inputs.pcp-size != '1' && format('/PCP{0}', inputs.pcp-size) || '' }}${{ inputs.ep != '' && inputs.ep != '1' && format('/EP{0}', inputs.ep) || '' }}${{ inputs.dp-attn && '/DPA' || '' }}
       ${{ inputs.spec-decoding != 'none' && inputs.spec-decoding || '' }}
       ${{ inputs.kv-offloading != '' && inputs.kv-offloading != 'none' && format('{0} KV offload', inputs.kv-offloading) || '' }}

.github/workflows/collectivex-sweep.yml

@@ -34,16 +34,24 @@ jobs:
     outputs:
       matrix: ${{ steps.gen.outputs.matrix }}
       n: ${{ steps.gen.outputs.n }}
+      priority: ${{ steps.score.outputs.priority }}
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v5.0.0
         with: { clean: true, persist-credentials: false }
+      - id: score
+        run: |
+          scored=$(printf '%s' '[{"runner":"collectivex","framework":"collectivex"}]' |
+            python3 utils/ci_priority.py)
+          echo "priority=$(jq -r '.[0].priority' <<<"$scored")" >> "$GITHUB_OUTPUT"
       - id: gen
         working-directory: experimental/CollectiveX
         env:
           INPUT_BACKEND: ${{ inputs.backend }}
           INPUT_ONLY_SKU: ${{ inputs.only_sku }}
           INPUT_EXCLUDE_SKUS: ${{ inputs.exclude_skus }}
           INPUT_EP_SIZES: ${{ inputs.ep_sizes }}
+          RUN_ID: ${{ github.run_id }}
+          RUN_ATTEMPT: ${{ github.run_attempt }}
         run: |
           set -euo pipefail
           args=(--backend "$INPUT_BACKEND")
@@ -52,12 +60,18 @@ jobs:
           [ -n "$INPUT_EP_SIZES" ] && args+=(--ep-sizes "$INPUT_EP_SIZES")
           python3 sweep_matrix.py "${args[@]}" --out matrix_full.json >/dev/null
           python3 - "$GITHUB_OUTPUT" <<'PY'
+          import hashlib
           import json
+          import os
           import pathlib
           import sys
 
           matrix = json.loads(pathlib.Path("matrix_full.json").read_text())
           cells = matrix["include"]
+          for index, cell in enumerate(cells):
+              canonical = json.dumps(cell, sort_keys=True, separators=(",", ":"))
+              material = f"{os.environ['RUN_ID']}:{os.environ['RUN_ATTEMPT']}:{index}:{canonical}".encode()
+              cell["queue-token"] = hashlib.sha256(material).hexdigest()[:32]
           slim = {"include": cells}
           with open(sys.argv[1], "a", encoding="utf-8") as output:
               output.write(f"matrix={json.dumps(slim, separators=(',', ':'))}\n")
@@ -83,7 +97,22 @@ jobs:
       # first jobs under this cap spread over pools instead of queuing on one.
       max-parallel: 10
       matrix: ${{ fromJSON(needs.setup.outputs.matrix) }}
-    runs-on: ${{ matrix.sku }}
+    runs-on: >-
+      ${{ fromJSON(
+        vars.PRIORITY_SCHEDULER_ENABLED == 'true' &&
+        format(
+          '["self-hosted",{0},{1},{2}]',
+          toJSON(matrix.sku),
+          toJSON(format(
+            'ci-job-{0}-{1}',
+            needs.setup.outputs.priority,
+            matrix.queue-token
+          )),
+          toJSON(format('ci-attempt-{0}', github.run_attempt))
+        ) ||
+        format('[{0}]', toJSON(matrix.sku))
+      ) }}
+    name: p${{ needs.setup.outputs.priority }} | ${{ matrix.sku }} ${{ matrix.backend }} shard ${{ matrix.id }}
     timeout-minutes: 350
     env:
       COLLX_BENCH: ${{ matrix.backend }}

.github/workflows/e2e-tests.yml

@@ -63,18 +63,40 @@ jobs:
               uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
               with:
                 ref: ${{ github.sha }}
+            - name: Checkout priority scheduler tooling
+              if: ${{ inputs.ref && inputs.ref != '' }}
+              uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+              with:
+                ref: ${{ github.workflow_sha }}
+                path: .ci-priority
+                persist-credentials: false
+
 
             - id: get-jobs
+              env:
+                  PR_LABELS: ${{ toJson(github.event.pull_request.labels.*.name) }}
               run: |
                   pip install pydantic
                   CONFIG_JSON=$(python3 ${GITHUB_WORKSPACE}/utils/matrix_logic/generate_sweep_configs.py \
                     ${{ inputs.generate-cli-command || github.event.inputs.generate-cli-command }})
-                  AGENTIC=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if x.get('scenario-type') == 'agentic-coding' and 'prefill' not in x]))")
-                  MULTI_AGENTIC=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if x.get('scenario-type') == 'agentic-coding' and 'prefill' in x]))")
-                  SINGLE=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' not in x and x.get('scenario-type') != 'agentic-coding' and not x.get('eval-only', False)]))")
-                  MULTI=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' in x and x.get('scenario-type') != 'agentic-coding' and not x.get('eval-only', False)]))")
-                  EVALS=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' not in x and x.get('scenario-type') != 'agentic-coding' and x.get('run-eval', False)]))")
-                  MULTI_EVAL=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' in x and x.get('run-eval', False)]))")
+                  PRIORITY_ROOT="${GITHUB_WORKSPACE}"
+                  if [ -d "${GITHUB_WORKSPACE}/.ci-priority" ]; then
+                    PRIORITY_ROOT="${GITHUB_WORKSPACE}/.ci-priority"
+                  fi
+                  score_matrix() {
+                    local family="$1"
+                    python3 "${PRIORITY_ROOT}/utils/ci_priority.py" \
+                      --policy "${PRIORITY_ROOT}/configs/ci-priority.yaml" \
+                      --event-name "${{ github.event_name }}" \
+                      --queue-namespace "${{ github.run_id }}:${{ github.run_attempt }}:${family}" \
+                      --labels-json "$PR_LABELS"
+                  }
+                  AGENTIC=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if x.get('scenario-type') == 'agentic-coding' and 'prefill' not in x]))" | score_matrix agentic)
+                  MULTI_AGENTIC=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if x.get('scenario-type') == 'agentic-coding' and 'prefill' in x]))" | score_matrix multi-agentic)
+                  SINGLE=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' not in x and x.get('scenario-type') != 'agentic-coding' and not x.get('eval-only', False)]))" | score_matrix single)
+                  MULTI=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' in x and x.get('scenario-type') != 'agentic-coding' and not x.get('eval-only', False)]))" | score_matrix multi)
+                  EVALS=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' not in x and x.get('scenario-type') != 'agentic-coding' and x.get('run-eval', False)]))" | score_matrix eval)
+                  MULTI_EVAL=$(echo "$CONFIG_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(json.dumps([x for x in d if 'prefill' in x and x.get('run-eval', False)]))" | score_matrix multi-eval)
                   echo "agentic-config=$AGENTIC" >> $GITHUB_OUTPUT
                   echo "multi-node-agentic-config=$MULTI_AGENTIC" >> $GITHUB_OUTPUT
                   echo "single-node-config=$SINGLE" >> $GITHUB_OUTPUT
@@ -97,6 +119,8 @@ jobs:
             osl: ${{ matrix.config.osl }}
             max-model-len: ${{ matrix.config.max-model-len }}
             runner: ${{ matrix.config.runner }}
+            priority: ${{ matrix.config.priority }}
+            queue-token: ${{ matrix.config['queue-token'] }}
             image: ${{ matrix.config.image }}
             model: ${{ matrix.config.model }}
             model-prefix: ${{ matrix.config.model-prefix }}
@@ -146,6 +170,8 @@ jobs:
             osl: ${{ matrix.config.osl }}
             max-model-len: ${{ matrix.config.max-model-len }}
             runner: ${{ matrix.config.runner }}
+            priority: ${{ matrix.config.priority }}
+            queue-token: ${{ matrix.config['queue-token'] }}
             image: ${{ matrix.config.image }}
             model: ${{ matrix.config.model }}
             model-prefix: ${{ matrix.config.model-prefix }}
@@ -195,6 +221,8 @@ jobs:
         with:
             exp-name: ${{ matrix.config.exp-name }}
             runner: ${{ matrix.config.runner }}
+            priority: ${{ matrix.config.priority }}
+            queue-token: ${{ matrix.config['queue-token'] }}
             image: ${{ matrix.config.image }}
             model: ${{ matrix.config.model }}
             model-prefix: ${{ matrix.config.model-prefix }}
@@ -239,6 +267,8 @@ jobs:
             osl: '0'
             max-model-len: '0'
             runner: ${{ matrix.config.runner }}
+            priority: ${{ matrix.config.priority }}
+            queue-token: ${{ matrix.config['queue-token'] }}
             image: ${{ matrix.config.image }}
             model: ${{ matrix.config.model }}
             model-prefix: ${{ matrix.config.model-prefix }}
@@ -292,6 +322,8 @@ jobs:
             osl: ${{ matrix.config.osl }}
             max-model-len: ${{ matrix.config.max-model-len }}
             runner: ${{ matrix.config.runner }}
+            priority: ${{ matrix.config.priority }}
+            queue-token: ${{ matrix.config['queue-token'] }}
             image: ${{ matrix.config.image }}
             model: ${{ matrix.config.model }}
             model-prefix: ${{ matrix.config.model-prefix }}
@@ -327,6 +359,8 @@ jobs:
             osl: ${{ matrix.config.osl }}
             max-model-len: ${{ matrix.config.max-model-len }}
             runner: ${{ matrix.config.runner }}
+            priority: ${{ matrix.config.priority }}
+            queue-token: ${{ matrix.config['queue-token'] }}
             image: ${{ matrix.config.image }}
             model: ${{ matrix.config.model }}
             model-prefix: ${{ matrix.config.model-prefix }}

.github/workflows/priority-scheduler-canary.yml

@@ -0,0 +1,121 @@
+name: Priority Scheduler Canary
+run-name: Priority Scheduler Canary - ${{ inputs.runner }} at p${{ inputs.score }}
+
+on:
+  workflow_dispatch:
+    inputs:
+      runner:
+        description: "Runner label to test"
+        required: true
+        type: choice
+        default: b200
+        options:
+          - b200
+          - b300
+          - h100
+          - h200
+          - mi355x
+      score:
+        description: "Numeric priority score"
+        required: true
+        type: string
+        default: "1.000"
+
+permissions:
+  contents: read
+  actions: write
+
+jobs:
+  prepare:
+    name: Prepare one-shot scheduling label
+    runs-on: ubuntu-latest
+    outputs:
+      job-label-base: ${{ steps.labels.outputs.job-label-base }}
+    steps:
+      - id: labels
+        name: Generate unique canary label
+        shell: python
+        env:
+          RUNNER_LABEL: ${{ inputs.runner }}
+          PRIORITY_SCORE: ${{ inputs.score }}
+          RUN_ID: ${{ github.run_id }}
+          RUN_ATTEMPT: ${{ github.run_attempt }}
+        run: |
+          import hashlib
+          import json
+          import os
+          import re
+
+          score = os.environ["PRIORITY_SCORE"]
+          if re.fullmatch(r"[0-9]+(?:\.[0-9]+)?", score) is None:
+              raise SystemExit(f"Invalid numeric priority: {score}")
+
+          material = f"{os.environ['RUN_ID']}:{os.environ['RUN_ATTEMPT']}:canary"
+          token = hashlib.sha256(material.encode()).hexdigest()[:32]
+          job_label_base = f"ci-job-{score}-{token}"
+
+          with open(os.environ["GITHUB_OUTPUT"], "a", encoding="utf-8") as output:
+              output.write(f"job-label-base={job_label_base}\n")
+
+  canary:
+    name: Verify priority dispatch
+    needs: prepare
+    runs-on: >-
+      ${{ fromJSON(
+        github.run_attempt == 1 &&
+        format(
+          '["self-hosted",{0},{1},{2}]',
+          toJSON(inputs.runner),
+          toJSON(needs.prepare.outputs.job-label-base),
+          toJSON(format('ci-attempt-{0}', github.run_attempt))
+        ) ||
+        '["ubuntu-latest"]'
+      ) }}
+    timeout-minutes: 10
+    steps:
+      - name: Reject partial rerun
+        if: github.run_attempt != 1
+        run: |
+          echo "Priority canaries are one-shot; dispatch a fresh workflow run." >&2
+          exit 1
+      - name: Confirm assignment
+        if: github.run_attempt == 1
+        env:
+          JOB_LABEL: ${{ needs.prepare.outputs.job-label-base }}
+          ATTEMPT_LABEL: ${{ format('ci-attempt-{0}', github.run_attempt) }}
+          RUNNER_LABEL: ${{ inputs.runner }}
+        run: |
+          echo "Controller dispatched $JOB_LABEL $ATTEMPT_LABEL to $RUNNER_LABEL runner $RUNNER_NAME"
+          {
+            echo "### Priority scheduler canary passed"
+            echo "- Job label: \`$JOB_LABEL\`"
+            echo "- Attempt label: \`$ATTEMPT_LABEL\`"
+            echo "- Runner class: \`$RUNNER_LABEL\`"
+            echo "- Runner: \`$RUNNER_NAME\`"
+          } >> "$GITHUB_STEP_SUMMARY"
+
+  watchdog:
+    name: Enforce canary queue deadline
+    needs: prepare
+    runs-on: ubuntu-latest
+    timeout-minutes: 12
+    steps:
+      - name: Cancel if the canary is still queued after ten minutes
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          set -euo pipefail
+          for _ in {1..40}; do
+            status=$(gh api \
+              "repos/${{ github.repository }}/actions/runs/${{ github.run_id }}/jobs?filter=latest" \
+              --jq '.jobs[] | select(.name == "Verify priority dispatch") | .status')
+            if [ "$status" = "in_progress" ] || [ "$status" = "completed" ]; then
+              exit 0
+            fi
+            sleep 15
+          done
+
+          echo "Priority canary remained queued for ten minutes; cancelling the run." >&2
+          gh api --method POST \
+            "repos/${{ github.repository }}/actions/runs/${{ github.run_id }}/cancel"
+          exit 1