test(ci): repair policy payload and probe deadline fixtures

tests/test_ci_resource_policy.py

@@ -20,7 +20,9 @@ def event(name: str, *, pr: int = 101, tag: str = "", backend: bool = True) -> d
         "github": SimpleNamespace(
             event_name=name,
             ref=f"refs/pull/{pr}/merge" if name == "pull_request" else "refs/heads/main",
-            event=SimpleNamespace(pull_request=SimpleNamespace(number=pr), release=SimpleNamespace(tag_name=tag)),
+            # Pull-request webhooks expose number at the payload root as well
+            # as inside pull_request; the shared concurrency owner uses both.
+            event=SimpleNamespace(number=pr, pull_request=SimpleNamespace(number=pr), release=SimpleNamespace(tag_name=tag)),
         ),
         "inputs": SimpleNamespace(tag=tag),
         "needs": SimpleNamespace(changes=SimpleNamespace(outputs=SimpleNamespace(backend_tests=str(backend).lower()))),
@@ -75,10 +77,25 @@ def test_main_supersession_does_not_cancel_nightly_or_manual_full_qualification(
     assert "schedule" in policy[True]
     concurrency = policy["concurrency"]
     contexts = [event(name) for name in ("push", "pull_request", "schedule", "workflow_dispatch", "merge_group")]
+    for context in contexts:
+        if context["github"].event_name != "pull_request":
+            del context["github"].event.number
     assert len({expression(concurrency["group"], context) for context in contexts}) == 5
     assert [expression(concurrency["cancel-in-progress"], context) for context in contexts] == [True, True, False, False, False]
 
 
+@pytest.mark.parametrize("name", ["pull_request", "pull_request_target"])
+def test_python_pr_supersession_uses_the_payload_number(name: str) -> None:
+    concurrency = workflow("python-tests")["concurrency"]
+    first, second = event(name, pr=101), event(name, pr=102)
+    # pull_request_target uses the base ref, so the payload number must keep
+    # distinct PRs apart even when their refs match.
+    first["github"].ref = second["github"].ref = "refs/heads/main"
+    assert expression(concurrency["group"], first) == f"python-tests-{name}-pr-101"
+    assert expression(concurrency["group"], second) == f"python-tests-{name}-pr-102"
+    assert expression(concurrency["cancel-in-progress"], first) is True
+
+
 def test_required_merge_gate_workflow_is_not_suppressed_by_path_filters() -> None:
     # A required workflow omitted by a path filter can leave a PR pending. The
     # stable gate always starts; job-level exemptions handle unrelated changes.

tests/test_zcode_host_diagnostics.py

@@ -336,11 +336,17 @@ def test_probe_deadline_survives_descendant_inheriting_stdout(monkeypatch, capsy
     assert capsys.readouterr().err == ""
 
 
-@pytest.mark.parametrize("mode,expected", [("timeout_default", "timeout"), ("timeout", "timeout"), ("output_limit", "output_limit"), ("exit_pipe", "timeout"), ("exit_closed", "observed")])
-def test_probe_reclaims_persistent_descendants_before_return(tmp_path, monkeypatch, mode, expected):
-    if mode != "timeout_default":
+@pytest.mark.parametrize("mode,expected,child_startup_delay", [
+    ("timeout_default", "timeout", 0), ("timeout", "timeout", 0),
+    ("output_limit", "output_limit", 0), ("exit_pipe", "timeout", 0),
+    ("exit_closed", "observed", 0), ("exit_closed", "observed", .6),
+])
+def test_probe_reclaims_persistent_descendants_before_return(tmp_path, monkeypatch, mode, expected, child_startup_delay):
+    # Successful closure uses the shipped deadline so interpreter and child
+    # startup are not constrained by a deadline for deliberately blocked probes.
+    if mode not in {"timeout_default", "exit_closed"}:
         monkeypatch.setattr(diagnostics, "PROBE_TIMEOUT_SECONDS", .5)
-    child_script = "import time, pathlib, sys, os; p=pathlib.Path(sys.argv[1]); p.write_text(str(os.getpid())); n=0\nwhile True:\n n+=1; p.with_suffix('.pulse').write_text(str(n)); time.sleep(.01)"
+    child_script = f"import time; time.sleep({child_startup_delay!r})\nimport pathlib, sys, os; p=pathlib.Path(sys.argv[1]); p.write_text(str(os.getpid())); n=0\nwhile True:\n n+=1; p.with_suffix('.pulse').write_text(str(n)); time.sleep(.01)"
     path = tmp_path / "child.pid"
     parent = "import subprocess,sys,time,pathlib,os; p=pathlib.Path(sys.argv[1]); c=subprocess.Popen([sys.executable,'-c',sys.argv[2],str(p)],stdout=subprocess.DEVNULL if sys.argv[3]=='exit_closed' else None,stderr=subprocess.DEVNULL if sys.argv[3]=='exit_closed' else None)\nwhile not p.exists(): time.sleep(.01)\nif sys.argv[3]=='output_limit': print('x'*70000,flush=True)\nif sys.argv[3].startswith('exit'): sys.exit(0)\ntime.sleep(30)"
     for _ in range(2):