Address PR review round 1

Six related changes from the PR review, mapped to the reviewers' points:

- Split the two synthetic-proband stages out of stages.py into a new
  synthetic_proband_stages.py so the pre-workflow's file doesn't have to
  share stages.py with the main workflow's stages (Matt).

- Hash-scope Stage 2's PED and gVCF-manifest into `families-<hash>/`
  subdirectories of self.prefix, driven by the SG set + qualifying family
  set. The family_set_marker sentinel and its Batch job go away - existing
  hash-scoped output paths are now what cpg-flow's REUSE check keys off
  (Matt).

- Collapse each Stage's queue_jobs onto a single public factory call, and
  make the per-artifact builders module-private with a leading underscore.
  Matches the "one factory call per Stage" pattern used everywhere else
  in the repo, and makes the stage-scheduling / stage-doing separation
  cleaner if this ever gets ported to another orchestration language
  (Matt).

- Rewrite Stage 2's per-artifact builders as driver-side writes via
  Path.open('w'), gated on workflow.dry_run. Deletes the two thin Bash
  jobs that only ran `cat > $output <<HEREDOC` and the heredoc-quoting
  hazard that went with them (Eddie + Matt).

- Rewrite scripts/register_synthetic_gvcf_analysis.py to call
  cpg_utils.metamist_registration.create_new instead of hand-rolling the
  AnalysisApi calls. Uses cpg_utils.config.dataset_for_access_level to
  resolve the metamist project name against access level, drops the
  bespoke find_existing_registration + deactivate_analysis idempotency
  layer (cpg-flow's REUSE + our per-family sentinel is enough - matches
  what every other analysis_type-decorated stage in the repo does), and
  picks up the newer `outputs` block shape for free (Matt, Eddie).

- Remove the last inline `ravenscroft-rpl` reference from utils.py
  (Eddie's steer earlier in the review).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: SamBryen

pyproject.toml

@@ -117,6 +117,7 @@ hail = ["hail"]
 # suppress the ARG002 "Unused method argument" warning in the stages.py file
 ## - we don't need generic cpg-flow arguments for every Stage, but need to fit the required method signature
 "src/cpg_seqr_loader/stages.py" = ["ARG002"]
+"src/cpg_seqr_loader/synthetic_proband_stages.py" = ["ARG002"]
 "src/cpg_seqr_loader/scripts/mt_to_es.py" = ["ANN202"]
 "src/cpg_seqr_loader/scripts/annotate_cohort.py" = ["E501"]
 

src/cpg_seqr_loader/jobs/GenerateSyntheticProbandCombinerInputs.py

@@ -1,123 +1,59 @@
 """
-Job factories for the synthetic proband combiner-inputs stage.
+Driver-side writer for the synthetic proband combiner-inputs stage.
 
-Two artifacts, one Batch job each:
+The single public entry point `write_combiner_inputs` produces both of the stage's output
+artifacts from the driver, no Batch jobs required:
 
   - Pedigree PED file: 6-column TSV with three rows per qualifying duo family (mother, father,
-    synthetic proband). Persistent so seqr sync can reuse it across loads.
+    synthetic proband), plus a row for every real SG in the multicohort.
 
   - gVCF manifest: newline-separated list of every gVCF that will go into the synthetic-trio
     combiner run. It includes every real SG in the multicohort that has a gVCF (qualifying or
     not, so we don't silently drop samples from the seqr load), plus every synthetic gVCF
     from Stage 1's outputs.
 
-Content for each file is computed in the driver via the pure builders in `utils`, then dropped
-into a Batch job via a quoted heredoc; `write_output` uploads the produced file to the durable
-gs:// path once the job succeeds. This keeps the writes gated on Batch success and gives the
-framework a real job to depend on rather than a driver-side side effect.
-"""
+Both files are tiny text blobs, so the previous approach of spinning up a Batch job per file
+(only to run `cat > $output <<HEREDOC`) added VM + container startup cost and a heredoc-quoting
+hazard for no benefit. Writing directly to `gs://` via `cpg_utils.Path.open('w')` is the same
+pattern used elsewhere in the repo (see jobs/SubsetMtToDatasetWithHail.py).
 
-from typing import TYPE_CHECKING
+The write is skipped in dry-run mode - unlike Batch jobs, which cpg_flow describes rather than
+executes under `workflow.dry_run`, a driver-side `.open('w')` executes unconditionally, so the
+callers must gate on that config themselves.
+"""
 
 from cpg_flow import targets
-from cpg_utils import Path, config, hail_batch
+from cpg_utils import Path, config
 
 from cpg_seqr_loader.utils import (
     SyntheticProbandFamily,
     build_gvcf_manifest_content,
     build_synthetic_pedigree_content,
 )
 
-if TYPE_CHECKING:
-    from hailtop.batch.job import BashJob
-
-
-def _write_content_job(name: str, content: str, out_path: Path, job_attrs: dict) -> 'BashJob':
-    """Return a Batch job that writes a fixed string to a durable output path.
-
-    Uses a quoted heredoc (`<<'EOF'`) so tabs / newlines pass through literally with no shell
-    expansion. Safe for our alphanumeric family_id / SG_id content; would need escaping for
-    arbitrary user input.
-    """
-    job = hail_batch.get_batch().new_bash_job(name, attributes=job_attrs)
-    job.image(config.config_retrieve(['workflow', 'driver_image']))
-    job.command(f"cat > {job.output} <<'STAGE2_EOF'\n{content}STAGE2_EOF\n")
-    hail_batch.get_batch().write_output(job.output, str(out_path))
-    return job
-
 
-def create_pedigree_job(
+def write_combiner_inputs(
     families: list[SyntheticProbandFamily],
     multicohort: targets.MultiCohort,
-    output_ped: Path,
-    job_attrs: dict,
-) -> 'BashJob':
-    """One Batch job that writes the synthetic-trio PED for the whole multicohort.
-
-    Content includes every real SG in the multicohort plus a synthetic-proband row for each
-    qualifying duo family. See utils.build_synthetic_pedigree_content for the row semantics.
-    """
-    content = build_synthetic_pedigree_content(multicohort, families)
-    return _write_content_job(
-        name='WriteSyntheticPedigree',
-        content=content,
-        out_path=output_ped,
-        job_attrs=job_attrs | {'tool': 'python'},
-    )
-
-
-def create_family_set_marker_job(
-    marker_path: Path,
-    upstream_jobs: list['BashJob'],
-    job_attrs: dict,
-) -> 'BashJob':
-    """Touch an empty file at `marker_path` once the PED and manifest jobs have both succeeded.
-
-    The marker's *filename* encodes a hash of the family set (see stages.py) - when the family
-    set changes, the hash changes, this marker path doesn't exist, and cpg_flow's REUSE check
-    forces Stage 2 to re-run. Without this, Stage 2's fixed-name outputs (PED, manifest) would
-    silently ship stale content whenever qualifying families come and go.
-
-    Depends on the pedigree + manifest jobs so the marker only exists once their outputs land.
-    """
-    job = hail_batch.get_batch().new_bash_job('WriteFamilySetMarker', attributes=job_attrs | {'tool': 'python'})
-    job.image(config.config_retrieve(['workflow', 'driver_image']))
-    for upstream in upstream_jobs:
-        job.depends_on(upstream)
-    job.command(f'touch {job.output}')
-    hail_batch.get_batch().write_output(job.output, str(marker_path))
-    return job
-
-
-def create_manifest_job(
-    families: list[SyntheticProbandFamily],
     synthetic_gvcf_paths: dict[str, Path],
-    multicohort: targets.MultiCohort,
+    output_ped: Path,
     output_manifest: Path,
-    job_attrs: dict,
-) -> 'BashJob':
-    """One Batch job that writes the combiner gVCF manifest.
-
-    Manifest composition (deliberately inclusive so samples aren't silently dropped from seqr):
-      - every real SG in the multicohort whose `.gvcf` is set, whether or not their family
-        qualified for synthetic proband synthesis;
-      - every synthetic gVCF from Stage 1's outputs (indexed by family_id).
+) -> None:
+    """Write the PED and gVCF manifest for the synthetic-trio combiner run.
 
-    Real SGs whose `.gvcf` is None are logged upstream by cpg_flow and simply won't appear here
-    - there's no useful path to write for them.
+    No Batch jobs are queued - both files are written directly from the driver. The stage's
+    `queue_jobs` still returns an empty job list so cpg_flow's REUSE check keys off the output
+    paths existing on disk, exactly as it would with jobs.
     """
-    real_paths: list[str] = []
-    for sg in multicohort.get_sequencing_groups():
-        if sg.gvcf is None:
-            continue
-        real_paths.append(str(sg.gvcf))
-
-    synthetic_paths = [str(synthetic_gvcf_paths[f.family_id]) for f in families]
-
-    content = build_gvcf_manifest_content(real_paths + synthetic_paths)
-    return _write_content_job(
-        name='WriteGvcfManifestWithSynthetics',
-        content=content,
-        out_path=output_manifest,
-        job_attrs=job_attrs | {'tool': 'python'},
-    )
+    if config.config_retrieve(['workflow', 'dry_run'], False):
+        return
+
+    pedigree_content = build_synthetic_pedigree_content(multicohort, families)
+    with output_ped.open('w') as f:
+        f.write(pedigree_content)
+
+    real_paths = [str(sg.gvcf) for sg in multicohort.get_sequencing_groups() if sg.gvcf is not None]
+    synthetic_paths = [str(synthetic_gvcf_paths[family.family_id]) for family in families]
+    manifest_content = build_gvcf_manifest_content(real_paths + synthetic_paths)
+    with output_manifest.open('w') as f:
+        f.write(manifest_content)

src/cpg_seqr_loader/jobs/GenerateSyntheticProbandGvcfs.py

@@ -1,13 +1,14 @@
 """
-Job factories for the synthetic proband gVCF stage.
+Job factory for the synthetic proband gVCF stage.
 
-Two factories work together:
+The single public entry point `create_synthetic_gvcf_jobs` builds every Batch job the stage
+needs and returns them in queue order. Internally it uses two private per-artifact factories:
 
-  - create_synthesis_jobs: one BashJob per duo family that invokes
+  - _create_synthesis_jobs: one BashJob per duo family that invokes
     scripts/create_synthetic_proband_gvcf.py with the family's parental gVCFs, writing the
     output gVCF to the durable per-family path supplied by the stage.
 
-  - create_analysis_registration_jobs: one BashJob per duo family that invokes
+  - _create_analysis_registration_jobs: one BashJob per duo family that invokes
     scripts/register_synthetic_gvcf_analysis.py to record (or refresh) the synthetic gVCF as a
     metamist Analysis of type SYNTHETIC_GVCF_ANALYSIS_TYPE. Each registration job depends_on
     the matching synthesis job so registration only fires once the gVCF actually exists.
@@ -26,7 +27,7 @@
     from hailtop.batch.job import BashJob
 
 
-def create_synthesis_jobs(
+def _create_synthesis_jobs(
     families: list[SyntheticProbandFamily],
     output_paths: dict[str, Path],
     job_attrs: dict,
@@ -78,7 +79,7 @@ def create_synthesis_jobs(
     return jobs
 
 
-def create_analysis_registration_jobs(
+def _create_analysis_registration_jobs(
     families: list[SyntheticProbandFamily],
     gvcf_paths: dict[str, Path],
     marker_paths: dict[str, Path],
@@ -142,3 +143,44 @@ def create_analysis_registration_jobs(
         jobs.append(job)
 
     return jobs
+
+
+def create_synthetic_gvcf_jobs(
+    families: list[SyntheticProbandFamily],
+    gvcf_paths: dict[str, Path],
+    marker_paths: dict[str, Path],
+    job_attrs: dict,
+) -> list['BashJob']:
+    """Build every Batch job the synthetic-gVCF stage needs, in the order they'll be queued.
+
+    Returns the synthesis jobs (one per family) followed by the registration jobs (one per
+    family). Registration jobs depend on their matching synthesis job so they only fire once
+    the gVCF exists.
+
+    Callers (the Stage class) should treat this as the single entry point for the stage - the
+    per-artifact factories are private to this module and shouldn't be invoked directly.
+    """
+    synthesis_jobs = _create_synthesis_jobs(
+        families=families,
+        output_paths=gvcf_paths,
+        job_attrs=job_attrs,
+    )
+
+    # Register per family: each Analysis lives in its parents' metamist project (which is the
+    # dataset owning the mother SG), so we can't batch across families that live in different
+    # datasets. cpg-flow's get_metamist().create_analysis handles the access-level suffix
+    # internally.
+    registration_jobs = []
+    for family in families:
+        registration_jobs.extend(
+            _create_analysis_registration_jobs(
+                families=[family],
+                gvcf_paths=gvcf_paths,
+                marker_paths=marker_paths,
+                synthesis_jobs=synthesis_jobs,
+                project=family.mother_sg.dataset.name,
+                job_attrs=job_attrs,
+            ),
+        )
+
+    return list(synthesis_jobs.values()) + registration_jobs