fix(testmon): run resumable shards with bounded argv

devtools/pytest_progress_plugin.py

@@ -156,10 +156,20 @@ def pytest_collection_modifyitems(session: Any, config: Any, items: list[Any]) -
     _SELECTED_COUNT = len(items)
     limit = _selection_nodeid_limit()
     selected_nodeids = [str(getattr(item, "nodeid", item)) for item in items[:limit]]
+    # Marker metadata is a compact routing index, not a node-id sample. Keep
+    # it complete so seed sharding can isolate load-sensitive/TUI nodes even
+    # when the human-readable node-id sample is capped at 500 entries.
+    selected_node_markers = {
+        str(getattr(item, "nodeid", item)): sorted(
+            {str(mark.name) for mark in getattr(item, "iter_markers", lambda: ())()}
+        )
+        for item in items
+    }
     payload: dict[str, Any] = {
         "selected_count": _SELECTED_COUNT,
         "deselected_count": _DESELECTED_COUNT,
         "selected_nodeids": selected_nodeids,
+        "selected_node_markers": selected_node_markers,
         "selected_nodeids_omitted": max(0, _SELECTED_COUNT - len(selected_nodeids)),
         "deselected_nodeids": list(_DESELECTED_NODEIDS_SAMPLE),
         "deselected_nodeids_omitted": max(0, _DESELECTED_COUNT - len(_DESELECTED_NODEIDS_SAMPLE)),

devtools/testmon_state.py

@@ -81,7 +81,12 @@ class TerminalAuthorization(StrEnum):
 _TERMINAL_NODE_OUTCOMES = frozenset({"passed", "failed", "error", "skipped"})
 
 
-def seed_shard_plan(nodeids: Sequence[str], *, shard_size: int) -> list[dict[str, Any]]:
+def seed_shard_plan(
+    nodeids: Sequence[str],
+    *,
+    shard_size: int,
+    serial_nodeids: Sequence[str] = (),
+) -> list[dict[str, Any]]:
     """Partition a complete node set into stable, contiguous, serial shards."""
     if shard_size <= 0:
         raise ValueError("testmon seed shard_size must be positive")
@@ -90,16 +95,29 @@ def seed_shard_plan(nodeids: Sequence[str], *, shard_size: int) -> list[dict[str
     if len(set(nodeids)) != len(nodeids):
         raise ValueError("testmon seed nodeids must be unique")
     ordered = tuple(sorted(nodeids))
+    serial = set(serial_nodeids)
+    if not serial.issubset(ordered):
+        raise ValueError("testmon serial shard nodes must belong to the seed corpus")
+    chunks: list[tuple[str, list[str]]] = []
+    for offset in range(0, len(ordered), shard_size):
+        chunk = list(ordered[offset : offset + shard_size])
+        parallel = [nodeid for nodeid in chunk if nodeid not in serial]
+        isolated = [nodeid for nodeid in chunk if nodeid in serial]
+        if parallel:
+            chunks.append(("parallel", parallel))
+        if isolated:
+            chunks.append(("serial", isolated))
     return [
         {
             "index": index,
-            "nodeids": list(ordered[offset : offset + shard_size]),
-            "nodeid_count": len(ordered[offset : offset + shard_size]),
-            "nodeid_digest": hashlib.sha256("\n".join(ordered[offset : offset + shard_size]).encode()).hexdigest(),
+            "nodeids": chunk,
+            "nodeid_count": len(chunk),
+            "nodeid_digest": hashlib.sha256("\n".join(chunk).encode()).hexdigest(),
+            "execution_mode": mode,
             "status": SeedShardStatus.PENDING.value,
             "node_outcomes": [],
         }
-        for index, offset in enumerate(range(0, len(ordered), shard_size), start=1)
+        for index, (mode, chunk) in enumerate(chunks, start=1)
     ]
 
 
@@ -120,7 +138,7 @@ def validate_seed_shard_ledger(
     if not expected or len(set(expected)) != len(expected):
         return None
     normalized: list[dict[str, Any]] = []
-    observed: list[str] = []
+    observed: set[str] = set()
     for index, raw in enumerate(shards, start=1):
         if not isinstance(raw, Mapping) or raw.get("index") != index:
             return None
@@ -170,9 +188,11 @@ def validate_seed_shard_ledger(
             and set(outcome_by_node) != set(nodeids)
         ):
             return None
+        if observed.intersection(nodeids):
+            return None
         normalized.append(dict(raw))
-        observed.extend(nodeids)
-    if tuple(observed) != expected:
+        observed.update(nodeids)
+    if observed != set(expected):
         return None
     return normalized
 
@@ -814,6 +834,9 @@ def inspect_testmon_database(path: Path, expected_nodeids: Sequence[str]) -> Gra
                     )
                 execution_ids.add(execution_id)
                 name = test_name
+                if name not in expected:
+                    grouped = [nodeid for nodeid in expected if name.startswith(nodeid + "@")]
+                    name = max(grouped, key=len, default=name)
                 prior = latest.get(name)
                 if prior is None or execution_id > prior[0]:
                     latest[name] = (execution_id, failed == 1)

devtools/verify.py

@@ -2631,7 +2631,19 @@ def _prepare_testmon_seed_shards(
     shards = (
         prior_shards
         if prior_shards is not None
-        else (seed_shard_plan(expected, shard_size=TESTMON_SEED_SHARD_SIZE) if expected else [])
+        else (
+            seed_shard_plan(
+                expected,
+                shard_size=TESTMON_SEED_SHARD_SIZE,
+                serial_nodeids=[
+                    nodeid
+                    for nodeid, markers in (selection or {}).get("selected_node_markers", {}).items()
+                    if "load_sensitive" in markers or "tui" in markers
+                ],
+            )
+            if expected
+            else []
+        )
     )
     payload = {
         **dict(prepared),
@@ -2646,16 +2658,64 @@ def _prepare_testmon_seed_shards(
     return payload
 
 
-def _seed_shard_command(collection_command: Sequence[str], shard: Mapping[str, Any]) -> list[str]:
-    """Build a dynamically balanced explicit-node pytest-testmon invocation."""
+def _seed_shard_command(
+    collection_command: Sequence[str],
+    shard: Mapping[str, Any],
+    *,
+    nodeids_file: Path,
+) -> list[str]:
+    """Build a bounded-argv, dynamically balanced pytest-testmon invocation.
+
+    A full shard's node IDs can exceed the host's ``execve`` argument budget
+    once ``systemd-run`` and the managed environment are included.  Pytest's
+    response-file syntax keeps the authoritative node list in the run
+    artifact while making the child command size independent of shard size.
+    """
     nodeids = shard.get("nodeids")
     if not isinstance(nodeids, list) or not nodeids:
         raise ValueError("testmon seed shard is missing nodeids")
-    command = [argument for argument in collection_command if argument != "--collect-only"]
-    command.extend(["--dist=worksteal", "--testmon", "--testmon-noselect", *nodeids])
+    nodeids_file.parent.mkdir(parents=True, exist_ok=True)
+    nodeids_file.write_text("\n".join(nodeids) + "\n", encoding="utf-8")
+    command: list[str] = []
+    skip_next = False
+    for argument in collection_command:
+        if skip_next:
+            skip_next = False
+            continue
+        if argument == "--collect-only":
+            continue
+        if argument in {"-n", "--numprocesses"}:
+            skip_next = True
+            continue
+        if argument.startswith("--numprocesses=") or (argument.startswith("-n") and len(argument) > 2):
+            continue
+        command.append(argument)
+    # Collection is deliberately serial, but execution is not. pytest-testmon
+    # has an xdist-aware controller database; retaining the managed worker pool
+    # here avoids turning a 20k-node seed into hours of serial fixture setup.
+    if shard.get("execution_mode") == "serial":
+        command.extend(["-n", "0", "--testmon", "--testmon-noselect", f"@{nodeids_file}"])
+    else:
+        command.extend(
+            [
+                "--dist=loadgroup",
+                *_pytest_worker_args(maximum=10),
+                "--testmon",
+                "--testmon-noselect",
+                f"@{nodeids_file}",
+            ]
+        )
     return command
 
 
+def _canonical_seed_nodeid(nodeid: str, expected_nodeids: Sequence[str]) -> str:
+    """Map xdist's ``nodeid@group`` reports back to the collected node ID."""
+    if nodeid in expected_nodeids:
+        return nodeid
+    candidates = [expected for expected in expected_nodeids if nodeid.startswith(expected + "@")]
+    return max(candidates, key=len, default=nodeid)
+
+
 def _seed_shard_outcomes(shards: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
     """Flatten the shard ledger in canonical node order for legacy readers."""
     outcomes: dict[str, dict[str, Any]] = {}
@@ -2684,7 +2744,10 @@ def _checkpoint_testmon_seed_shard(
     nodeids = shard["nodeids"]
     artifact_dir = _safe_testmon_artifact_dir(step.get("artifact_dir"))
     selection = _read_json_artifact(artifact_dir / "selection.json") if artifact_dir is not None else None
-    selected = _seed_selection_nodeids(selection) if isinstance(selection, Mapping) else None
+    selected_raw = _seed_selection_nodeids(selection) if isinstance(selection, Mapping) else None
+    selected = (
+        sorted(_canonical_seed_nodeid(nodeid, nodeids) for nodeid in selected_raw) if selected_raw is not None else None
+    )
     database = _testmon_database_state(nodeids)
     prior = {
         str(item["nodeid"]): item
@@ -2781,6 +2844,7 @@ def _seed_node_outcomes_from_events(
                     nodeid = event.get("nodeid")
                     if not isinstance(nodeid, str) or not nodeid:
                         continue
+                    nodeid = _canonical_seed_nodeid(nodeid, expected_nodeids)
                     if event.get("event") == "test_started":
                         started.add(nodeid)
                     elif event.get("event") == "test_finished":
@@ -3422,7 +3486,32 @@ def main(argv: list[str] | None = None) -> int:
                     continue
                 shard_index = int(shard["index"])
                 shard_label = f"pytest seed-testmon shard {shard_index}/{len(shards)}"
-                shard_cmd = _seed_shard_command(cmd, shard)
+                shard_args_path = verify_run.run_dir / "seed-shards" / f"{shard_index:04d}.args"
+                try:
+                    shard_cmd = _seed_shard_command(cmd, shard, nodeids_file=shard_args_path)
+                except (OSError, PytestResourceError) as exc:
+                    resource_failure_result = {
+                        "name": shard_label,
+                        "duration_s": 0.0,
+                        "exit": 125,
+                        "diagnosis": (
+                            "pytest_resource_refusal"
+                            if isinstance(exc, PytestResourceError)
+                            else "testmon_seed_args_file_write_failed"
+                        ),
+                        "error": str(exc),
+                        "shard_index": shard_index,
+                        "shard_count": len(shards),
+                        "shard_nodeid_count": len(shard["nodeids"]),
+                    }
+                    step_results.append(resource_failure_result)
+                    prepared_seed_attempt = _checkpoint_testmon_seed_shard(
+                        prepared=prepared_seed_attempt,
+                        shard_index=shard_index,
+                        step=resource_failure_result,
+                    )
+                    exit_code = 125
+                    break
                 _warn_low_memory()
                 shard_rc, shard_elapsed, shard_metadata = _run(shard_label, shard_cmd, run=verify_run)
                 shard_result: dict[str, Any] = {

tests/unit/devtools/test_testmon_state.py

@@ -88,6 +88,7 @@ def _attempt(data: Path, *, outcomes: tuple[str, str] = ("passed", "failed")) ->
 def test_failed_complete_graph_is_selection_only_and_rebindable(tmp_path: Path) -> None:
     data = tmp_path / "testmondata"
     _write_graph(data, failed=True)
+
     stamp = stamp_from_attempt(_attempt(data), data, checkout_root=tmp_path, protocol_version=PROTOCOL)
 
     assert stamp is not None
@@ -107,6 +108,35 @@ def test_failed_complete_graph_is_selection_only_and_rebindable(tmp_path: Path)
     assert validate_stamp(stamp_path, data, checkout_root=tmp_path, protocol_version=PROTOCOL) is None
 
 
+def test_seed_shard_ledger_rejects_duplicate_nodes_across_shards() -> None:
+    shard = {
+        "index": 1,
+        "nodeids": [NODEIDS[0]],
+        "nodeid_count": 1,
+        "nodeid_digest": hashlib.sha256(NODEIDS[0].encode()).hexdigest(),
+        "status": "complete",
+        "node_outcomes": [{"nodeid": NODEIDS[0], "outcome": "passed"}],
+    }
+    duplicate = {**shard, "index": 2}
+
+    assert testmon_state.validate_seed_shard_ledger([shard, duplicate], expected_nodeids=[NODEIDS[0]]) is None
+
+
+def test_testmon_database_canonicalizes_xdist_group_names(tmp_path: Path) -> None:
+    data = tmp_path / "testmondata"
+    _write_graph(data)
+    with sqlite3.connect(data) as connection:
+        connection.execute(
+            "UPDATE test_execution SET test_name = ? WHERE test_name = ?",
+            (f"{NODEIDS[0]}@web-reader", NODEIDS[0]),
+        )
+
+    graph = inspect_testmon_database(data, NODEIDS)
+
+    assert graph.missing_nodeids == ()
+    assert graph.recorded_count == len(NODEIDS)
+
+
 def test_omitted_interrupted_and_uncovered_nodes_fail_closed(tmp_path: Path) -> None:
     data = tmp_path / "testmondata"
     _write_graph(data)

tests/unit/devtools/test_verify.py

@@ -322,10 +322,9 @@ def test_seed_testmon_caps_adaptive_workers(monkeypatch: pytest.MonkeyPatch) ->
     assert command[command.index("-n") + 1] == "0"
 
 
-def test_seed_shards_are_deterministic_and_use_one_testmon_writer(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
+def test_seed_shards_are_deterministic_and_use_managed_xdist(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr("devtools.verify.adaptive_pytest_worker_count", lambda _environment: 64)
     expected = sorted(f"tests/test_seed.py::test_{index:03d}" for index in range(TESTMON_SEED_SHARD_SIZE + 2))
     prepared = _prepare_testmon_seed_shards(
         {"resume": False, "expected_nodeids": []},
@@ -340,13 +339,51 @@ def test_seed_shards_are_deterministic_and_use_one_testmon_writer(
     assert [shard["nodeid_count"] for shard in shards] == [TESTMON_SEED_SHARD_SIZE, 2]
     assert shards[0]["nodeids"] == expected[:TESTMON_SEED_SHARD_SIZE]
     assert shards[1]["nodeids"] == expected[TESTMON_SEED_SHARD_SIZE:]
-    command = _seed_shard_command(["pytest", "--collect-only", "-n", "0"], shards[0])
+    nodeids_file = tmp_path / "seed-shard.args"
+    command = _seed_shard_command(["pytest", "--collect-only", "-n", "0"], shards[0], nodeids_file=nodeids_file)
     assert "--collect-only" not in command
-    assert command[command.index("-n") + 1] == "0"
+    assert command[command.index("-n") + 1] == "10"
     assert "--testmon" in command
     assert "--testmon-noselect" in command
-    assert "--dist=worksteal" in command
-    assert command[-TESTMON_SEED_SHARD_SIZE:] == expected[:TESTMON_SEED_SHARD_SIZE]
+    assert "--dist=loadgroup" in command
+    assert command.count("-n") == 1
+    assert command[command.index("-n") + 1] == "10"
+    assert command[-1] == f"@{nodeids_file}"
+    assert nodeids_file.read_text().splitlines() == expected[:TESTMON_SEED_SHARD_SIZE]
+
+
+def test_seed_outcomes_normalize_xdist_group_suffix(tmp_path: Path) -> None:
+    expected = ["tests/test_seed.py::test_grouped"]
+    events = tmp_path / "events.jsonl"
+    events.write_text(
+        json.dumps(
+            {
+                "event": "test_report",
+                "nodeid": f"{expected[0]}@web-reader",
+                "when": "call",
+                "outcome": "passed",
+            }
+        )
+        + "\n"
+    )
+
+    outcomes = _seed_node_outcomes_from_events(
+        events,
+        expected_nodeids=expected,
+        database={"node_outcomes": {}},
+        pytest_step=None,
+    )
+
+    assert outcomes == [
+        {
+            "nodeid": expected[0],
+            "outcome": "passed",
+            "reason": "test call passed",
+            "started": False,
+            "finished": False,
+            "phases": [{"when": "call", "outcome": "passed", "duration_s": None}],
+        }
+    ]
 
 
 def test_seed_shard_checkpoint_preserves_completed_shards_for_resume(
@@ -385,10 +422,24 @@ def test_seed_shard_checkpoint_preserves_completed_shards_for_resume(
     artifact_dir = tmp_path / "shard-1"
     artifact_dir.mkdir()
     (artifact_dir / "selection.json").write_text(
-        json.dumps({"selected_count": 1, "selected_nodeids": [ordered[0]], "selected_nodeids_omitted": 0})
+        json.dumps(
+            {
+                "selected_count": 1,
+                "selected_nodeids": [f"{ordered[0]}@web-reader"],
+                "selected_nodeids_omitted": 0,
+            }
+        )
     )
     (artifact_dir / "events.jsonl").write_text(
-        json.dumps({"event": "test_report", "nodeid": ordered[0], "when": "call", "outcome": "passed"}) + "\n"
+        json.dumps(
+            {
+                "event": "test_report",
+                "nodeid": f"{ordered[0]}@web-reader",
+                "when": "call",
+                "outcome": "passed",
+            }
+        )
+        + "\n"
     )
 
     checkpointed = _checkpoint_testmon_seed_shard(