From a587895956fc19a9e28ca4494fe34e0cff0e866a Mon Sep 17 00:00:00 2001 From: Ciprian Goea Date: Wed, 12 Aug 2026 17:13:06 +0300 Subject: [PATCH 1/5] prevent cross-architectural leakeage in matrix_job_re --- .../tests/therock_update_status_json_test.py | 46 +++++++++ .../therock_update_status_json.py | 96 +++++++++++++++---- 2 files changed, 125 insertions(+), 17 deletions(-) diff --git a/scripts/receive_therock/tests/therock_update_status_json_test.py b/scripts/receive_therock/tests/therock_update_status_json_test.py index c1f4b920e..f5ae5c1db 100644 --- a/scripts/receive_therock/tests/therock_update_status_json_test.py +++ b/scripts/receive_therock/tests/therock_update_status_json_test.py @@ -1599,3 +1599,49 @@ def test_completed_fanout_build_refreshes_same_run_test_leaves() -> None: assert leaf.completed_at == "2026-06-19T15:18:00Z" assert leaf.variants is not None assert leaf.variants[0].status is Status.success + + +def test_completed_fanout_build_does_not_leak_status_across_architectures() -> None: + # Two GPUs are tested under the *same* (py, torch) build cell. The + # matrix-cell key parsed from job names carries no arch, so refreshing + # same-run test leaves must not broadcast one architecture's outcome onto + # another's: gfx1101 failing must not drag down gfx942's passing leaf. + doc = StatusDocument() + for arch in ("gfx942", "gfx1101"): + stale_test = _variant_run( + pipeline_type="pytorch", + pipeline_phase="test", + architectures=[arch], + run_id=901, + conclusion=None, + jobs=[ + _job("Build | py 3.12 | torch release/2.10 / Build"), + _job( + f"Build | py 3.12 | torch release/2.10 / Test | {arch}", + conclusion=None, + completed=None, + ), + ], + ) + doc.upsert_leaf("linux", arch, "pytorch", "test", tusj._create_leaf(stale_test)) + + completed_build = _variant_run( + pipeline_type="pytorch", + pipeline_phase="build", + run_id=901, + conclusion="failure", + jobs=[ + _job("Build | py 3.12 | torch release/2.10 / Build"), + _job("Build | py 3.12 | torch release/2.10 / Test | gfx942"), + _job( + "Build | py 3.12 | torch release/2.10 / Test | gfx1101", + conclusion="failure", + ), + ], + ) + tusj._merge_run_into_document( + doc, completed_build, tusj._create_leaf(completed_build) + ) + + assert doc.pipelines.pytorch.test["linux"]["gfx942"].status is Status.success + assert doc.pipelines.pytorch.test["linux"]["gfx1101"].status is Status.failure diff --git a/scripts/receive_therock/therock_update_status_json.py b/scripts/receive_therock/therock_update_status_json.py index 1e004eb2c..1866b7409 100644 --- a/scripts/receive_therock/therock_update_status_json.py +++ b/scripts/receive_therock/therock_update_status_json.py @@ -329,6 +329,15 @@ def _update_document_metadata( # `.search` (not fullmatch) so a reusable-workflow prefix/suffix still matches. _MATRIX_JOB_RE = re.compile(r"py\s+(?P\S+)\s*\|\s*(?:torch|jax)\s+(?P\S+)") +# One (py, ref) build cell can nest per-arch test jobs, e.g. +# "Build | py 3.12 | torch release/2.10 / Test | gfx942" +# Extracts the arch a job's own "Test | " segment names, if any, so +# jobs from different architectures nested under the same cell are never +# grouped together as if they were one architecture's result. +_TEST_ARCH_JOB_RE = re.compile( + r"Test\s*\|\s*(?Pgfx[0-9A-Za-z]+(?:-[0-9A-Za-z]+)?)" +) + # pipeline_type -> the matrix axis key used in the variant (reference schema: # pytorch cells key the ref as "torch", jax cells as "jax_ref"). _VARIANT_AXIS_KEY: dict[str, str] = {"pytorch": "torch", "jax": "jax_ref"} @@ -354,15 +363,30 @@ def _run_status(workflow_run: WorkflowRunRecord) -> Status: return Status.in_progress +def _job_matches_arch(job_name: str, arch: str) -> bool: + """True if `job_name` is arch-agnostic (no "Test | " segment of its + own, e.g. the cell's shared build step) or explicitly names `arch`.""" + named = _TEST_ARCH_JOB_RE.findall(job_name) + return not named or arch in named + + def _variants_from_jobs( - workflow_run: WorkflowRunRecord, axis_key: str + workflow_run: WorkflowRunRecord, axis_key: str, *, arch: str | None = None ) -> list[Variant]: - """One variant per (py, ref) matrix cell parsed from job names.""" + """One variant per (py, ref) matrix cell parsed from job names. + + A single (py, ref) build cell can nest test jobs for several + architectures (see `_TEST_ARCH_JOB_RE`). Passing `arch` scopes the cell + to that architecture's own jobs plus any arch-agnostic job, so one + architecture's result can never roll up into another's variant. + """ jobs = ( workflow_run.api_jobs if workflow_run.api_jobs is not None else workflow_run.jobs ) + if arch is not None: + jobs = [j for j in jobs if _job_matches_arch(j.name, arch)] cells: dict[tuple[str, str], list[WorkflowJobRecord]] = {} order: list[tuple[str, str]] = [] for j in jobs: @@ -437,19 +461,30 @@ def _variants_from_inputs( ] -def _derive_variants(workflow_run: WorkflowRunRecord) -> list[Variant] | None: - """Matrix-cell variants for fan-out pipelines (pytorch/jax py x ref).""" +def _derive_variants( + workflow_run: WorkflowRunRecord, *, arch: str | None = None +) -> list[Variant] | None: + """Matrix-cell variants for fan-out pipelines (pytorch/jax py x ref). + + See `_variants_from_jobs` for what `arch` scopes. + """ axis_key = _VARIANT_AXIS_KEY.get(workflow_run.classification.pipeline_type) if axis_key is None: return None - variants = _variants_from_jobs(workflow_run, axis_key) + variants = _variants_from_jobs(workflow_run, axis_key, arch=arch) if not variants: variants = _variants_from_inputs(workflow_run, axis_key) return variants or None -def _create_leaf(workflow_run: WorkflowRunRecord) -> RunLeaf: - """Map the enriched WorkflowRunRecord to a v2 RunLeaf.""" +def _create_leaf( + workflow_run: WorkflowRunRecord, *, arch: str | None = None +) -> RunLeaf: + """Map the enriched WorkflowRunRecord to a v2 RunLeaf. + + `arch`, when given, scopes matrix-cell variants (and the leaf's own + rolled-up status) to that architecture -- see `_variants_from_jobs`. + """ ts_start = workflow_run.run_started_at or workflow_run.created_at started_at = _datetime_to_z(ts_start) if ts_start is not None else None @@ -457,13 +492,18 @@ def _create_leaf(workflow_run: WorkflowRunRecord) -> RunLeaf: if workflow_run.conclusion and workflow_run.updated_at is not None: completed_at = _datetime_to_z(workflow_run.updated_at) + variants = _derive_variants(workflow_run, arch=arch) + status = _run_status(workflow_run) + if arch is not None and variants: + status = Variant.rollup_status(variants, status) + return RunLeaf( run_id=workflow_run.workflow_run_id, run_attempt=workflow_run.run_attempt, - status=_run_status(workflow_run), + status=status, started_at=started_at, completed_at=completed_at, - variants=_derive_variants(workflow_run), + variants=variants, ) @@ -479,9 +519,10 @@ def _refresh_same_run_fanout_tests( those same-run test leaves stale. """ cls = workflow_run.classification + axis_key = _VARIANT_AXIS_KEY.get(cls.pipeline_type) if ( cls.pipeline_phase != "build" - or cls.pipeline_type not in _VARIANT_AXIS_KEY + or axis_key is None or not workflow_run.conclusion or not leaf.variants or leaf.run_id is None @@ -492,16 +533,29 @@ def _refresh_same_run_fanout_tests( wrote = False for phase_map in (pipeline.test, pipeline.test_full): for arch_map in phase_map.values(): - for existing in arch_map.values(): + for arch, existing in arch_map.items(): if existing.run_id != leaf.run_id: continue if (existing.run_attempt or 0) != (leaf.run_attempt or 0): continue - if not existing.should_replace(leaf): + # One build cell can nest test jobs for several architectures + # (see `_variants_from_jobs`); re-derive this arch's own + # variants instead of broadcasting `leaf.variants`, which may + # have rolled every architecture's outcome together. + arch_variants = _variants_from_jobs(workflow_run, axis_key, arch=arch) + if not arch_variants: continue - existing.status = leaf.status - existing.completed_at = leaf.completed_at - existing.variants = leaf.variants + candidate = leaf.model_copy( + update={ + "status": Variant.rollup_status(arch_variants, leaf.status), + "variants": arch_variants, + } + ) + if not existing.should_replace(candidate): + continue + existing.status = candidate.status + existing.completed_at = candidate.completed_at + existing.variants = candidate.variants wrote = True return wrote @@ -735,14 +789,22 @@ def _merge_run_into_document( list(cls.architectures) if cls.pipeline_phase in ("test", "test-full") else [""] ) + # A single event reporting more than one architecture (multi-arch test + # dispatch) would otherwise upsert the *same* leaf object -- variants + # derived from the run's full, arch-blind job list -- into every target + # arch's slot. Re-derive a leaf scoped to each arch so one architecture's + # result can never be attributed to another's. + multi_arch = len(targets) > 1 and cls.pipeline_type in _VARIANT_AXIS_KEY + leaf_accepted = False for arch in targets: + arch_leaf = _create_leaf(workflow_run, arch=arch) if multi_arch else leaf wrote = doc.upsert_leaf( platform=cls.platform, arch=arch, pipeline_type=cls.pipeline_type, pipeline_phase=cls.pipeline_phase, - leaf=leaf, + leaf=arch_leaf, ) leaf_accepted = leaf_accepted or wrote if not wrote: @@ -755,7 +817,7 @@ def _merge_run_into_document( cls.pipeline_phase, workflow_run.workflow_run_id, workflow_run.run_attempt, - leaf.status, + arch_leaf.status, ) # URLs are updated only after the leaf upsert, gated on acceptance: a stale From c4b3705e21f93610e5b149a01a03f85ae3bea4c5 Mon Sep 17 00:00:00 2001 From: Ciprian Goea Date: Fri, 14 Aug 2026 12:54:55 +0300 Subject: [PATCH 2/5] address review: clarify why _create_leaf must not fold run status into arch rollup --- .../receive_therock/therock_update_status_json.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/scripts/receive_therock/therock_update_status_json.py b/scripts/receive_therock/therock_update_status_json.py index f1c788aff..55d1feb2d 100644 --- a/scripts/receive_therock/therock_update_status_json.py +++ b/scripts/receive_therock/therock_update_status_json.py @@ -526,6 +526,14 @@ def _create_leaf( `arch`, when given, scopes matrix-cell variants (and the leaf's own rolled-up status) to that architecture -- see `_variants_from_jobs`. + `arch` is only ever set when this run reports *multiple* architectures + (see `_merge_run_into_document`), so `workflow_run`'s own conclusion is a + whole-run aggregate across all of them, not this one arch's outcome. + Unlike `_refresh_same_run_fanout_tests`'s single-arch case, it must not be + folded into the rollup as a vote here: doing so would broadcast one + shared status onto every architecture -- exactly the leakage `arch` + scoping exists to prevent. It is used only as the fallback when this + arch has no variants of its own to roll up. """ ts_start = workflow_run.run_started_at or workflow_run.created_at started_at = _datetime_to_z(ts_start) if ts_start is not None else None @@ -535,9 +543,10 @@ def _create_leaf( completed_at = _datetime_to_z(workflow_run.updated_at) variants = _derive_variants(workflow_run, arch=arch) - status = _run_status(workflow_run) if arch is not None and variants: - status = Variant.rollup_status(variants, status) + status = Variant.rollup_status(variants, Status.in_progress) + else: + status = _run_status(workflow_run) return RunLeaf( run_id=workflow_run.workflow_run_id, From 9da7cf193a931a472fee686df92f7ab40df2a13f Mon Sep 17 00:00:00 2001 From: Ciprian Goea Date: Fri, 14 Aug 2026 13:59:10 +0300 Subject: [PATCH 3/5] address review: same dead-fallback pattern in _refresh_same_run_fanout_tests --- scripts/receive_therock/therock_update_status_json.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/scripts/receive_therock/therock_update_status_json.py b/scripts/receive_therock/therock_update_status_json.py index 55d1feb2d..76781bcc8 100644 --- a/scripts/receive_therock/therock_update_status_json.py +++ b/scripts/receive_therock/therock_update_status_json.py @@ -629,7 +629,12 @@ def _refresh_same_run_fanout_tests( statuses.append(leaf.status) candidate = leaf.model_copy( update={ - "status": rollup_statuses(statuses, leaf.status), + # `statuses` is never empty here (guarded by the + # `if not arch_variants: continue` above), so this + # fallback can never fire; `leaf.status` only ever + # affects the result via the `single_arch` vote above, + # never as an unconditional broadcast to every arch. + "status": rollup_statuses(statuses, Status.in_progress), "variants": arch_variants, } ) From 072538533da4bc8d1adb34a8213d2bb4380457c2 Mon Sep 17 00:00:00 2001 From: Ciprian Goea Date: Fri, 14 Aug 2026 14:04:10 +0300 Subject: [PATCH 4/5] address review: share GPU family token shape with therock_classify --- scripts/receive_therock/therock_classify.py | 8 +++++++- .../receive_therock/therock_update_status_json.py | 15 ++++++++++----- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/scripts/receive_therock/therock_classify.py b/scripts/receive_therock/therock_classify.py index b1ad81ab6..402318cb7 100644 --- a/scripts/receive_therock/therock_classify.py +++ b/scripts/receive_therock/therock_classify.py @@ -127,7 +127,13 @@ def _platform_from_test_runs_on(wr: WorkflowRunRecord) -> str: return "windows" if "windows" in runs_on else "linux" -_GPU_FAMILY_RE: Final = re.compile(r"\b(gfx[0-9A-Za-z]+(?:-[0-9A-Za-z]+)?)\b") +# Shape of one GPU family token, e.g. "gfx942" or "gfx94X-dcgpu". Exposed +# (unlike the compiled regexes below) so other modules that need the same +# token shape in a different context -- e.g. therock_update_status_json's +# `_TEST_ARCH_JOB_RE`, which anchors it to a job's own "Test | " +# segment -- stay in sync with this one instead of drifting independently. +GPU_FAMILY_TOKEN: Final = r"gfx[0-9A-Za-z]+(?:-[0-9A-Za-z]+)?" +_GPU_FAMILY_RE: Final = re.compile(rf"\b({GPU_FAMILY_TOKEN})\b") def _split_families(val: object) -> list[str]: diff --git a/scripts/receive_therock/therock_update_status_json.py b/scripts/receive_therock/therock_update_status_json.py index 76781bcc8..26427cb46 100644 --- a/scripts/receive_therock/therock_update_status_json.py +++ b/scripts/receive_therock/therock_update_status_json.py @@ -47,6 +47,7 @@ from therock_classify import ( FINALIZING_PHASES, + GPU_FAMILY_TOKEN, RELEASE_CDN_PHASES, is_top_level_orchestrator, ) @@ -336,13 +337,17 @@ def _update_document_metadata( ) # One (py, ref) build cell can nest per-arch test jobs, e.g. -# "Build | py 3.12 | torch release/2.10 / Test | gfx942" +# "Build | py 3.12 | torch release/2.10 / Test | gfx942 | linux-gfx942-1gpu..." # Extracts the arch a job's own "Test | " segment names, if any, so # jobs from different architectures nested under the same cell are never -# grouped together as if they were one architecture's result. -_TEST_ARCH_JOB_RE = re.compile( - r"Test\s*\|\s*(?Pgfx[0-9A-Za-z]+(?:-[0-9A-Za-z]+)?)" -) +# grouped together as if they were one architecture's result. Deliberately +# anchored to the "Test | " segment rather than reusing therock_classify's +# bare `_GPU_FAMILY_RE` (though it shares the same GPU_FAMILY_TOKEN shape): +# an unanchored scan would also match the runner-label segment that often +# follows in the same job name (e.g. "linux-gfx942-1gpu-..." above, which +# names a *different* family string than the job's own "Test | gfx94X-dcgpu" +# segment) and reintroduce the cross-arch conflation this exists to prevent. +_TEST_ARCH_JOB_RE = re.compile(rf"Test\s*\|\s*(?P{GPU_FAMILY_TOKEN})") # pipeline_type -> the matrix axis key used in the variant (reference schema: # pytorch cells key the ref as "torch", jax cells as "jax_ref"). From 8ca3e9e3543ff12cedd1f05247443424471963d7 Mon Sep 17 00:00:00 2001 From: Ciprian Goea Date: Fri, 14 Aug 2026 15:00:20 +0300 Subject: [PATCH 5/5] address review: pin _merge_run_into_document's multi_arch branch --- .../tests/therock_update_status_json_test.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/scripts/receive_therock/tests/therock_update_status_json_test.py b/scripts/receive_therock/tests/therock_update_status_json_test.py index 371c297ad..ffa42f7fb 100644 --- a/scripts/receive_therock/tests/therock_update_status_json_test.py +++ b/scripts/receive_therock/tests/therock_update_status_json_test.py @@ -1836,6 +1836,41 @@ def test_completed_fanout_build_does_not_leak_status_across_architectures() -> N assert doc.pipelines.pytorch.test["linux"]["gfx1101"].status is Status.failure +def test_multi_arch_test_event_does_not_leak_or_alias_across_architectures() -> None: + # Distinct from the fanout-build case above (a *build* run whose nested + # test jobs get refreshed into already-existing per-arch test leaves): + # this pins _merge_run_into_document's own multi_arch branch, for a + # single *test*-phase event that itself reports more than one + # architecture in cls.architectures. No dispatcher today fans a single + # test-phase run across several different GPUs (each test run reports + # exactly one arch), but nothing prevents one from doing so in the + # future -- if it did, the two architectures' leaves must be genuinely + # independent objects, not aliases sharing one leaf/variants list. + run = _variant_run( + pipeline_type="pytorch", + pipeline_phase="test", + architectures=["gfx942", "gfx1101"], + run_id=904, + conclusion="failure", + jobs=[ + _job("py 3.12 | torch release/2.10 / Test | gfx942"), + _job( + "py 3.12 | torch release/2.10 / Test | gfx1101", + conclusion="failure", + ), + ], + ) + doc = StatusDocument() + tusj._merge_run_into_document(doc, run, tusj._create_leaf(run)) + + gfx942 = doc.pipelines.pytorch.test["linux"]["gfx942"] + gfx1101 = doc.pipelines.pytorch.test["linux"]["gfx1101"] + assert gfx942.status is Status.success + assert gfx1101.status is Status.failure + assert gfx942 is not gfx1101 + assert gfx942.variants is not gfx1101.variants + + def test_fanout_projection_uses_variant_rollup_not_raw_run_conclusion() -> None: # The build run's own top-level GitHub conclusion is not necessarily the # worst-of its matrix cells (e.g. a cell whose nested test job failed does