feat(ci): start chunk 0 at the head, or wherever the operator points

The regular rows began wherever (sweep * capacity) % count pointed. On run 210
that was offset 68: chunk 0 opened at candidate 69, the coverage-first head of
the list went undeployed, and no two runs could be compared because the window
moved with the run number. A run that looked wrong turned out to be merely
rotated, and finding that out meant recomputing the arithmetic by hand.

The rotation is gone. Chunk 0 reads the ranking from the top unless an offset
says otherwise, and the offset is an input now, threaded from the manual form
through the orchestrator and the deploy call down to the matrix builder. The
plan renderer takes the same flag, so the table shows the window the run will
take.

The offset is a row count or a selection token. A token is the honest form:
the ranking shifts whenever a variant appears, so 'offset 84' means something
different tomorrow while 'web-app-nextcloud#5' still means the same row. Axes
the token leaves open match anything, so a bare role name starts at its first
row and a pasted job label starts at exactly the job it names. A token that
matches no regular row aborts rather than silently starting at the head an
operator meant to skip.

The run title carries it as ⏩ followed by the value, and only when one was
asked for, so a rotated-looking run says so in its own name and the retrigger
reads it back from there when a job log has expired.

resume_from_chunk leaves the manual form with it. Skipping whole chunks was
the coarse way to say 'carry on further down'; the offset says it exactly and
keeps the rest of the chain intact. The orchestrator input stays for the other
entry points.

What this gives up is automatic tail coverage. Rows beyond the budget are
reached by moving the offset, widening the budget or narrowing the selection,
not by waiting for consecutive sweeps to walk the list.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.github/workflows/call-orchestrator.yml

@@ -27,10 +27,15 @@ on:
         type: string
         default: ""
       sweep:
-        description: "Sweep number driving the mode/tor rotation and the regular-line offset. Empty: the run number."
+        description: "Sweep number driving the mode and tor rotation. Empty: the run number."
         required: false
         type: string
         default: ""
+      offset:
+        description: "Regular rows to skip before filling the chunks. 0 starts at the head of the ranking; raise it to reach rows the budget leaves behind."
+        required: false
+        type: string
+        default: "0"
       resume_from_chunk:
         description: "⏭️ Skip every chunk below this index. Use to continue a sweep after fixing what broke it, instead of re-running the green chunks. Typed string, not number: a workflow_dispatch number input reaches the caller as a string, and handing that to a number-typed workflow_call input makes GitHub reject the job before it is created."
         required: false
@@ -180,6 +185,7 @@ jobs:
       distros: ${{ inputs.distros }}
       index: "0"
       sweep: ${{ inputs.sweep != '' && inputs.sweep || github.run_number }}
+      offset: ${{ inputs.offset }}
       whitelist: ${{ inputs.whitelist }}
       priority: ${{ inputs.priority }}
       lifecycles: ${{ inputs.lifecycles }}
@@ -211,6 +217,7 @@ jobs:
       distros: ${{ inputs.distros }}
       index: "1"
       sweep: ${{ inputs.sweep != '' && inputs.sweep || github.run_number }}
+      offset: ${{ inputs.offset }}
       whitelist: ${{ inputs.whitelist }}
       priority: ${{ inputs.priority }}
       lifecycles: ${{ inputs.lifecycles }}
@@ -242,6 +249,7 @@ jobs:
       distros: ${{ inputs.distros }}
       index: "2"
       sweep: ${{ inputs.sweep != '' && inputs.sweep || github.run_number }}
+      offset: ${{ inputs.offset }}
       whitelist: ${{ inputs.whitelist }}
       priority: ${{ inputs.priority }}
       lifecycles: ${{ inputs.lifecycles }}

.github/workflows/call-test-deploy.yml

@@ -15,6 +15,11 @@ on:
         description: "Sweep number driving the mode and tor rotation and the regular-line offset."
         required: true
         type: string
+      offset:
+        description: "Regular rows to skip before filling the chunks; 0 starts at the head of the ranking."
+        required: false
+        type: string
+        default: "0"
       whitelist:
         description: "Optional space-separated whitelist of app ids"
         required: false
@@ -168,6 +173,7 @@ jobs:
           INFINITO_TOR: ${{ inputs.tor }}
           INFINITO_CI_SWEEP: ${{ inputs.sweep }}
           INFINITO_CI_CHUNK: ${{ inputs.index }}
+          INFINITO_CI_OFFSET: ${{ inputs.offset }}
         run: ./scripts/github/resolve/output_apps.sh
 
       - name: 🗺️ Render the plan into the run summary

.github/workflows/entry-manual-steer.yml

@@ -1,5 +1,5 @@
 name: "🕹️ Manual"
-run-name: "🕹️ ${{ inputs.chunk_gate && '🛑 ' || '' }}${{ inputs.resume_from_chunk != 0 && format('⏭️{0} ', inputs.resume_from_chunk) || '' }}${{ inputs.workspace != 'auto' && format('🧑‍💻{0} ', inputs.workspace == 'true' && '是' || '否') || '' }}${{ inputs.instructions != 'auto' && format('📖{0} ', inputs.instructions == 'true' && '是' || '否') || '' }}${{ inputs.tor != 'auto' && format('{0} ', inputs.tor == 'disabled' && '🌐' || inputs.tor == 'enforced' && '🧅强制' || '🧅独占') || '' }}${{ inputs.filesystem != '' && format('🗄️{0} ', inputs.filesystem) || '' }}${{ inputs.distros != '' && format('🐧{0} ', inputs.distros) || '' }}${{ inputs.lifecycles != '' && format('♻️{0} ', inputs.lifecycles) || '' }}${{ inputs.mode != 'auto' && format('🚀{0} ', inputs.mode) || '' }}${{ inputs.priority != '' && format('⭐{0} ', inputs.priority) || '' }}${{ inputs.whitelist != '' && format('🎯{0}', inputs.whitelist) || '🔀 diff (origin/main)' }}"
+run-name: "🕹️ ${{ inputs.chunk_gate && '🛑 ' || '' }}${{ inputs.offset != '0' && inputs.offset != '' && format('⏩{0} ', inputs.offset) || '' }}${{ inputs.workspace != 'auto' && format('🧑‍💻{0} ', inputs.workspace == 'true' && '是' || '否') || '' }}${{ inputs.instructions != 'auto' && format('📖{0} ', inputs.instructions == 'true' && '是' || '否') || '' }}${{ inputs.tor != 'auto' && format('{0} ', inputs.tor == 'disabled' && '🌐' || inputs.tor == 'enforced' && '🧅强制' || '🧅独占') || '' }}${{ inputs.filesystem != '' && format('🗄️{0} ', inputs.filesystem) || '' }}${{ inputs.distros != '' && format('🐧{0} ', inputs.distros) || '' }}${{ inputs.lifecycles != '' && format('♻️{0} ', inputs.lifecycles) || '' }}${{ inputs.mode != 'auto' && format('🚀{0} ', inputs.mode) || '' }}${{ inputs.priority != '' && format('⭐{0} ', inputs.priority) || '' }}${{ inputs.whitelist != '' && format('🎯{0}', inputs.whitelist) || '🔀 diff (origin/main)' }}"
 
 concurrency:
   group: global-ci-${{ github.repository }}-${{ github.ref_name }}
@@ -53,8 +53,8 @@ on:
           - "enforced"
           - "exclusive"
           - "disabled"
-      resume_from_chunk:
-        description: "⏭️ Skip every chunk below this index after fixing what broke a sweep"
+      offset:
+        description: "⏩ Where chunk 0 starts: a row count, or a token like web-app-x#2"
         required: false
         default: "0"
         type: string
@@ -64,7 +64,7 @@ on:
         default: true
         type: boolean
       sweep:
-        description: "🔁 Sweep number driving rotation and offset. Empty: the run number"
+        description: "🔁 Sweep number driving the mode and tor rotation. Empty: the run number"
         required: false
         default: ""
         type: string
@@ -131,7 +131,7 @@ jobs:
       mode: ${{ inputs.mode }}
       priority: ${{ inputs.priority }}
       sweep: ${{ inputs.sweep }}
-      resume_from_chunk: ${{ inputs.resume_from_chunk }}
+      offset: ${{ inputs.offset }}
       chunk_gate: ${{ inputs.chunk_gate }}
       filesystem: ${{ inputs.filesystem }}
       workspace: ${{ inputs.workspace }}

cli/meta/ci/chunks.py

@@ -11,15 +11,18 @@ queue window (:mod:`cli.meta.ci.slots` sizes them). Two rules shape the split:
   every priority row is done before the first regular one starts. Priority is
   never subject to the sweep offset: it leads every sweep.
 
-* **The regular tail rotates.** A sweep cannot always afford every row (the
-  run job cap bounds it), so the regular rows start at an offset that
-  advances by the sweep's own capacity. Consecutive sweeps therefore walk the
-  whole list instead of re-testing the same head forever. The offset is a
-  closed form over the sweep number, so no state is carried between runs.
+* **The head leads unless an operator says otherwise.** Chunk 0 starts at the
+  top of the ranking. The list is ordered coverage-first, so its head is what a
+  run most needs proven, and a sweep that cannot afford the whole list leaves
+  the tail behind rather than sliding a window over it.
 
-The rotation deliberately does NOT apply when every regular row fits: with
-nothing to leave behind there is nothing to rotate to, and a stable order
-keeps the plan output comparable between runs.
+  ``offset`` moves that start deliberately: it is how an operator reaches the
+  tail without waiting for anything. It defaults to 0, so every run that does
+  not ask for it deploys the same head and can be compared against the last
+  one. This used to be derived from the sweep number instead, which bought tail
+  coverage at the price of the head -- chunk 0 began wherever the arithmetic
+  pointed, the rows that matter most ran only every few sweeps, and no two runs
+  were comparable.
 """
 
 from __future__ import annotations
@@ -40,41 +43,28 @@ def slice_chunks(rows: Sequence[T], size: int) -> list[list[T]]:
     return [list(rows[start : start + size]) for start in range(0, len(rows), size)]
 
 
-def offset(count: int, capacity: int, sweep: int) -> int:
-    """Where a sweep starts reading the regular rows.
-
-    Args:
-        count: regular rows available.
-        capacity: regular rows this sweep can afford.
-        sweep: sweep number; consecutive numbers walk the list.
-
-    Returns:
-        the start index, or 0 when the whole list fits (nothing is left
-        behind, so there is nothing to rotate towards).
-    """
-    if count < 1 or capacity < 1 or capacity >= count:
-        return 0
-    return (sweep * capacity) % count
-
-
 def plan(
     priority: Sequence[T],
     regular: Sequence[T],
     *,
-    sweep: int,
     size: int,
     blocks: int,
     budget: int,
+    offset: int = 0,
 ) -> list[list[T]]:
     """The chunks one sweep deploys, priority blocks first.
 
     Args:
         priority: rows of the priority line, in query order.
         regular: rows of the regular line, in query order.
-        sweep: sweep number, driving the regular rotation.
         size: rows one chunk may hold (``slots.chunk_size``).
         blocks: chunk blocks the workflow declares (``slots.chunk_count``).
         budget: rows the run job cap allows in total (``slots.available``).
+        offset: how many regular rows to skip before filling the chunks. ``0``
+            (the default) starts at the head of the ranking. A value past the
+            end of the list leaves the regular chunks empty rather than
+            wrapping: an operator who asks for a window beyond the tail gets an
+            empty run, not a silent restart at the top.
 
     Returns:
         at most *blocks* chunks, none larger than *size*, together holding at
@@ -86,6 +76,5 @@ def plan(
     capacity = max(min((blocks - len(priority_chunks)) * size, budget - spent), 0)
     if not capacity or not regular:
         return priority_chunks
-    start = offset(len(regular), capacity, sweep)
-    rotated = [*regular[start:], *regular[:start]]
-    return priority_chunks + slice_chunks(rotated[:capacity], size)
+    start = max(offset, 0)
+    return priority_chunks + slice_chunks(regular[start : start + capacity], size)

cli/meta/ci/matrix.py

@@ -99,21 +99,66 @@ def entries_of(
     )
 
 
-def chunks_of(entries: list[dict[str, str]], sweep: int) -> list[list[dict[str, str]]]:
+def resolve_offset(raw: int | str | None = None) -> str:
+    """The offset a run was given, as written. ``None`` reads
+    ``INFINITO_CI_OFFSET``; :func:`offset_index` decides what it means."""
+    if raw is None:
+        raw = os.environ.get("INFINITO_CI_OFFSET")
+    return "" if raw is None else str(raw).strip()
+
+
+def offset_index(raw: int | str | None, regular: list[dict[str, str]]) -> int:
+    """Where the regular chunks start reading.
+
+    A number is that many rows; anything else is a selection token
+    (:mod:`utils.github.variant.selection`) and the start is the first regular
+    row it names. Resuming at a role beats counting rows by hand: the ranking
+    shifts whenever a variant is added, so the index that meant 'carry on
+    behind nextcloud' yesterday means something else today, while the token
+    still says it.
+
+    Args:
+        raw: the ``offset`` input, a count or a token.
+        regular: the regular line in ranking order.
+
+    Raises:
+        SystemExit: a token that names no regular row. Starting at 0 instead
+            would silently redeploy the head an operator meant to skip.
+    """
+    text = "" if raw is None else str(raw).strip()
+    if not text:
+        return 0
+    if text.lstrip("+-").isdigit():
+        return max(int(text), 0)
+    pin = selection.parse(text)
+    for index, entry in enumerate(regular):
+        if selection.covers(pin, entry):
+            return index
+    raise SystemExit(
+        f"offset {selection.describe(pin)!r} names no row of the regular line; "
+        f"it is on the priority line, outside the run's filters, or gone"
+    )
+
+
+def chunks_of(
+    entries: list[dict[str, str]], offset: int | str | None = 0
+) -> list[list[dict[str, str]]]:
     """Cut the sweep's entries into its chunks, priority blocks first.
 
-    Which rows a chunk holds follows the discovery ranking; the order *inside*
-    a chunk is then sorted (:func:`utils.github.variant.axes.sort_key`), so the
-    job list of a chunk reads by role rather than by the sweep's rotation."""
+    Chunk 0 holds the head of the ranking unless *offset* moves it; the order
+    *inside* a chunk is then sorted
+    (:func:`utils.github.variant.axes.sort_key`), so the job list reads by role
+    rather than by rank."""
+    regular = [entry for entry in entries if entry["priority"] != "true"]
     return [
         sorted(chunk, key=axes.sort_key)
         for chunk in chunks.plan(
             [entry for entry in entries if entry["priority"] == "true"],
-            [entry for entry in entries if entry["priority"] != "true"],
-            sweep=sweep,
+            regular,
             size=slots.chunk_size(),
             blocks=slots.chunk_count(),
             budget=slots.available(),
+            offset=offset_index(offset, regular),
         )
     ]
 
@@ -126,6 +171,7 @@ def build_sweep(
     lifecycles: str,
     sweep: int,
     tor_mode: str,
+    offset: int = 0,
 ) -> list[list[dict[str, str]]]:
     """Every chunk of the sweep, priority blocks first."""
     return chunks_of(
@@ -137,7 +183,7 @@ def build_sweep(
             sweep=sweep,
             tor_mode=tor_mode,
         ),
-        sweep,
+        offset,
     )
 
 
@@ -152,6 +198,7 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--priority", default="")
     parser.add_argument("--lifecycles", default="")
     parser.add_argument("--tor", default=None)
+    parser.add_argument("--offset", default=None)
     args = parser.parse_args(argv)
 
     codec = display_names()
@@ -166,6 +213,7 @@ def main(argv: list[str] | None = None) -> int:
         lifecycles=args.lifecycles,
         sweep=sweep,
         tor_mode=axes.resolve_tor_mode(args.tor),
+        offset=resolve_offset(args.offset),
     )
     chunk = plan[args.index] if 0 <= args.index < len(plan) else []
     dropped = ("priority", "weight", "id", "covered")

cli/meta/ci/plan.py

@@ -11,7 +11,7 @@ diverge. One row per ``role#variant`` candidate in global query order, with
 the chunk it falls into and the axes it was assigned.
 
 Status per row: ⭐ a priority row, ✅ deployed by this sweep, ❌ beyond the
-sweep's budget (the next sweep's offset picks it up). ``--chunk`` marks the
+sweep's budget (raise ``--offset`` to reach it). ``--chunk`` marks the
 block the calling job is running, so a chunk's summary shows its own slice in
 the context of the whole chain. ``--cli`` renders fixed-width terminal tables
 instead of Markdown.
@@ -146,6 +146,7 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--sweep", type=int, default=None)
     parser.add_argument("--chunk", type=int, default=None)
     parser.add_argument("--tor", default=None)
+    parser.add_argument("--offset", default=None)
     parser.add_argument("--cli", action="store_true")
     args = parser.parse_args(argv)
 
@@ -162,7 +163,7 @@ def main(argv: list[str] | None = None) -> int:
         sweep=sweep,
         tor_mode=axes.resolve_tor_mode(args.tor),
     )
-    plan = matrix.chunks_of(entries, sweep)
+    plan = matrix.chunks_of(entries, matrix.resolve_offset(args.offset))
     rows = cells(entries, plan, distros=args.distros, current=args.chunk)
 
     render = render_cli if args.cli else render_markdown

default.env

@@ -84,6 +84,7 @@ INFINITO_CACHE_REGISTRY_PROXY_CONF=./compose/registry-cache/proxy.conf
 
 # Index of the deploy chunk the discovery emits; each chunk block of call-orchestrator.yml passes its own.
 INFINITO_CI_CHUNK=0
+INFINITO_CI_OFFSET=0
 
 # Runner slots a deploy chunk may assume are free; with the job timeout it sizes a chunk in cli.meta.ci.slots.
 INFINITO_CI_CONCURRENCY=20

scripts/meta/resolve/apps.sh

@@ -9,7 +9,8 @@ set -euo pipefail
 #
 # Inputs via env (defaults live in default.env, the single source of truth):
 #   INFINITO_CI_CHUNK              chunk index to emit (required)
-#   INFINITO_CI_SWEEP              sweep number driving rotation and offset
+#   INFINITO_CI_SWEEP              sweep number driving the mode/tor rotation
+#   INFINITO_CI_OFFSET             regular rows to skip before filling the chunks
 #   INFINITO_MODES                 'auto' or a subset of 'host compose swarm'
 #   INFINITO_WHITELIST             optional space-separated app ids to keep
 #   INFINITO_PRIORITY              optional space-separated app ids to lead
@@ -57,6 +58,7 @@ matrix_json="$(
 	run_meta_cli -m cli.meta.ci.matrix \
 		--index "${INFINITO_CI_CHUNK:?INFINITO_CI_CHUNK must be set to the chunk index to emit}" \
 		--sweep "${INFINITO_CI_SWEEP}" \
+		--offset "${INFINITO_CI_OFFSET}" \
 		--modes "${INFINITO_MODES}" \
 		--whitelist "${INFINITO_WHITELIST}" \
 		--priority "${INFINITO_PRIORITY}" \

tests/integration/meta/workflows/test_ci_retrigger_inputs.py

@@ -40,7 +40,7 @@ _VALUES = {
     "mode": "swarm",
     "filesystem": "btrfs",
     "tor": "enforced",
-    "resume_from_chunk": "2",
+    "offset": "40",
     "chunk_gate": "false",
     "sweep": "7",
     "workspace": "true",

tests/unit/cli/meta/ci/test_chunks.py

@@ -18,22 +18,21 @@ class TestSliceChunks(unittest.TestCase):
 
 
 class TestOffset(unittest.TestCase):
-    def test_a_list_that_fits_is_never_rotated(self) -> None:
-        self.assertEqual(chunks.offset(10, 10, 5), 0)
-        self.assertEqual(chunks.offset(10, 20, 5), 0)
+    def test_the_head_leads_when_no_offset_is_asked_for(self) -> None:
+        plan = chunks.plan([], list(range(256)), size=80, blocks=2, budget=256)
+        self.assertEqual(plan[0][0], 0)
 
-    def test_consecutive_sweeps_walk_the_list(self) -> None:
-        self.assertEqual(
-            [chunks.offset(256, 160, sweep) for sweep in range(3)], [0, 160, 64]
-        )
+    def test_an_offset_moves_the_window_by_exactly_that_many_rows(self) -> None:
+        plan = chunks.plan([], list(range(256)), size=80, blocks=2, budget=256, offset=5)
+        self.assertEqual(plan[0][0], 5)
 
-    def test_every_row_is_covered_within_two_sweeps(self) -> None:
-        count, capacity = 256, 160
-        covered = set()
-        for sweep in (0, 1):
-            start = chunks.offset(count, capacity, sweep)
-            covered |= {(start + i) % count for i in range(capacity)}
-        self.assertEqual(covered, set(range(count)))
+    def test_an_offset_past_the_tail_yields_nothing_instead_of_wrapping(self) -> None:
+        plan = chunks.plan([], list(range(10)), size=80, blocks=2, budget=256, offset=99)
+        self.assertEqual(plan, [])
+
+    def test_a_negative_offset_reads_as_the_head(self) -> None:
+        plan = chunks.plan([], list(range(20)), size=80, blocks=2, budget=256, offset=-3)
+        self.assertEqual(plan[0][0], 0)
 
 
 class TestPlan(unittest.TestCase):
@@ -41,7 +40,6 @@ class TestPlan(unittest.TestCase):
         plan = chunks.plan(
             list(range(5)),
             list(range(100, 300)),
-            sweep=0,
             size=80,
             blocks=3,
             budget=211,
@@ -53,7 +51,6 @@ class TestPlan(unittest.TestCase):
         plan = chunks.plan(
             list(range(5)),
             list(range(100, 300)),
-            sweep=0,
             size=80,
             blocks=3,
             budget=211,
@@ -61,47 +58,37 @@ class TestPlan(unittest.TestCase):
         self.assertTrue(all(row < 5 for row in plan[0]))
 
     def test_without_priority_the_first_chunk_is_regular(self) -> None:
-        plan = chunks.plan([], list(range(200)), sweep=0, size=80, blocks=3, budget=211)
+        plan = chunks.plan([], list(range(200)), size=80, blocks=3, budget=211)
         self.assertEqual([len(c) for c in plan], [80, 80, 40])
 
     def test_the_block_count_bounds_the_sweep(self) -> None:
-        plan = chunks.plan([], list(range(500)), sweep=0, size=80, blocks=2, budget=500)
+        plan = chunks.plan([], list(range(500)), size=80, blocks=2, budget=500)
         self.assertEqual(len(plan), 2)
 
     def test_the_job_budget_bounds_the_sweep(self) -> None:
-        plan = chunks.plan([], list(range(500)), sweep=0, size=80, blocks=3, budget=100)
+        plan = chunks.plan([], list(range(500)), size=80, blocks=3, budget=100)
         self.assertEqual(sum(len(c) for c in plan), 100)
 
     def test_priority_alone_may_fill_every_block(self) -> None:
         plan = chunks.plan(
-            list(range(300)), list(range(50)), sweep=0, size=80, blocks=3, budget=300
+            list(range(300)), list(range(50)), size=80, blocks=3, budget=300
         )
         self.assertEqual([len(c) for c in plan], [80, 80, 80])
 
-    def test_priority_never_moves_with_the_sweep_offset(self) -> None:
-        for sweep in range(4):
-            plan = chunks.plan(
-                list(range(5)),
-                list(range(100, 300)),
-                sweep=sweep,
-                size=80,
-                blocks=3,
-                budget=211,
-            )
-            self.assertEqual(plan[0], list(range(5)), f"sweep {sweep}")
-
-    def test_the_regular_tail_rotates_between_sweeps(self) -> None:
-        heads = []
-        for sweep in range(3):
-            plan = chunks.plan(
-                [], list(range(256)), sweep=sweep, size=80, blocks=2, budget=256
-            )
-            heads.append(plan[0][0])
-        self.assertEqual(len(set(heads)), 3)
+    def test_priority_is_never_moved_by_the_offset(self) -> None:
+        plan = chunks.plan(
+            list(range(5)),
+            list(range(100, 300)),
+            size=80,
+            blocks=3,
+            budget=211,
+            offset=40,
+        )
+        self.assertEqual(plan[0], list(range(5)))
 
     def test_an_empty_selection_yields_no_chunk(self) -> None:
         self.assertEqual(
-            chunks.plan([], [], sweep=0, size=80, blocks=3, budget=211), []
+            chunks.plan([], [], size=80, blocks=3, budget=211), []
         )
 
 

tests/unit/cli/meta/ci/test_matrix.py

@@ -60,3 +60,36 @@ class TestChunksOf(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+_REGULAR = [
+    _entry("web-app-a", "0", "compose"),
+    _entry("web-app-b", "1", "swarm"),
+    _entry("web-app-b", "2", "compose"),
+]
+
+
+class TestOffsetIndex(unittest.TestCase):
+    def test_nothing_given_starts_at_the_head(self) -> None:
+        for raw in (None, "", 0, "0"):
+            with self.subTest(raw=raw):
+                self.assertEqual(matrix.offset_index(raw, _REGULAR), 0)
+
+    def test_a_number_is_a_row_count(self) -> None:
+        self.assertEqual(matrix.offset_index("2", _REGULAR), 2)
+
+    def test_a_negative_count_reads_as_the_head(self) -> None:
+        self.assertEqual(matrix.offset_index("-5", _REGULAR), 0)
+
+    def test_a_role_token_starts_at_its_first_row(self) -> None:
+        self.assertEqual(matrix.offset_index("web-app-b", _REGULAR), 1)
+
+    def test_a_pinned_variant_starts_at_that_variant(self) -> None:
+        self.assertEqual(matrix.offset_index("web-app-b#2", _REGULAR), 2)
+
+    def test_a_pinned_mode_picks_the_row_of_that_mode(self) -> None:
+        self.assertEqual(matrix.offset_index("web-app-b#1@swarm", _REGULAR), 1)
+
+    def test_a_token_naming_no_regular_row_aborts(self) -> None:
+        with self.assertRaises(SystemExit):
+            matrix.offset_index("web-app-gone#3", _REGULAR)

utils/github/variant/selection.py

@@ -160,6 +160,25 @@ def parse(token: str) -> Pin:
     )
 
 
+def covers(pin: Pin, entry: Mapping[str, Any]) -> bool:
+    """Whether *pin* names this matrix entry.
+
+    An axis the token leaves open matches anything, so ``web-app-x`` covers
+    every row of that role and ``web-app-x#2@swarm`` only its swarm deploy of
+    variant 2. Reads the strings a matrix entry carries, not a discovery row's
+    native types.
+    """
+    if entry.get("apps") != pin.app:
+        return False
+    if pin.variants and str(entry.get("variant", "")) not in {
+        str(index) for index in pin.variants
+    }:
+        return False
+    if pin.mode is not None and entry.get("mode") != pin.mode:
+        return False
+    return pin.tor is None or (entry.get("tor") == "true") == pin.tor
+
+
 def parse_list(tokens: str) -> list[Pin]:
     """Every token of a space-separated ``whitelist``/``priority`` input."""
     return [parse(token) for token in tokens.split()]