Automatic phase-2 hand-off: create super cohort and self-submit via analysis-runner

[joshuamschmidt]

Purpose

Stacked on #44. Removes the manual break between the two pipeline phases so a single launch runs end to end.

Previously the analyst ran phase 1, created the super cohort between phases, then launched phase 2 as a new run. cpg-flow can't create or validate a cohort mid-run, but the cohort only needs to exist before the phase-2 driver builds its DAG. A batch job at the end of phase 1 can therefore create the cohort and submit phase 2 itself. Self-submission is possible since analysis-runner 3.3.0 (populationgenomics/analysis-runner#764), which skips the full-access confirmation prompt when stdin is not a TTY. Same approach as populationgenomics/cpg-flow-seqr-loader#62, plus cohort creation via CohortApi.create_cohort_from_criteria.

Changes

• jobs/submit_phase2_job.py: PythonJob in the driver image, run off spot so a preemption cannot repeat a completed submission. In order, the job:

  1. Refuses to run if the submission record already exists, so a forced re-run (check_expected_outputs = false) or a rescheduled attempt cannot submit phase 2 twice.
  2. Waits until every plate cohort's array_cohort_bed and array_bafregress analyses are registered in Metamist: cpg-flow's registration jobs are not stage dependencies, and phase 2 resolves these analyses at driver startup.
  3. Computes super-cohort membership (previous aggregate cohort SGs plus this run's plate SGs) and creates or reuses the cohort.
  4. Rewrites the run config (input_cohorts = [super], strips ar-guid and any stage selection), writes the submission record, then POSTs the phase-2 submission to the analysis-runner server. An HTTP failure fails the job (the run_analysis_runner helper would log and swallow it). The POST carries no repo or commit, so the pinned driver image alone supplies the code.

  The record is written before the POST, so the worst crash leaves a record without a submission; the recovery is in the error message (delete the sentinel, re-run phase 1).

• stages.py: new SubmitPhase2 MultiCohortStage with required_stages=[BafRegress, CohortBcfToPlink], so it runs once, after the plate compute jobs. Reads the same previous_aggregate_cohort_id key as MergeCohortPlink, so the two phases cannot drift.

• metamist_utils.py: resolve_super_cohort_membership, find_cohort_by_membership, create_custom_cohort, wait_for_cohort_analyses, and a shared metamist_project helper replacing three copies of the -test suffix logic. Membership resolution raises when every plate SG is already in the previous aggregate (nothing new to aggregate). A re-run reuses an existing cohort with identical membership (latest by numeric ID if several match) instead of creating a duplicate; a name collision with different membership raises. Creation passes sg_ids_internal as the sole criterion, then verifies the created cohort's membership against the request and raises on any shortfall (e.g. SGs Metamist excluded as ineligible), rather than shipping a silently smaller cohort.

• previous_aggregate_cohort_id is now required in both phases: a cohort ID, or the literal 'bootstrap' for a from-scratch build. Previously the key was optional and absence meant bootstrap, so a forgotten entry would silently build a new-plates-only aggregate and register it as the latest. This is a config-breaking change for existing configs.

• Entry points: run_workflow replaced by first_workflow (per-plate stages plus hand-off) and second_workflow (merge, export, QC). The split replaces Two phase pipeline #44's submission-time phase check: each entry point pins its own stage list and rejects an only_stages selection naming stages outside its phase, and second_workflow still requires exactly one input cohort. Accumulating plates over several phase-1 runs before one aggregation still works via workflow.last_stages.

• Configs: config_phase1.toml no longer pins only_stages (it would skip SubmitPhase2) and gains the required super_cohort_name key. In both phases driver_image must be this repo's own image pinned to an exact tag, never :latest: the submit job imports the package and re-submits with that image string, which phase 2 resolves at its own start, so a floating tag can run the two phases on different code.

• pyproject.toml: require analysis-runner>=3.3.0; declare requests and toml as direct dependencies of the submit job.

• README and DAG diagram updated.

For reviewers: this changes when an aggregate goes live

Under the agreed design (Harper/Josh/Silk meeting, 13 Jul 2026), running phase 2 was a deliberate act with a manual cohort-creation step in between. With this PR, launching phase 1 produces a new registered array_aggregate_pgen hours later, unattended, and downstream consumers (single_sample_qc, the atlas) pick up the latest aggregate. The merge-plan printout becomes a post-hoc audit rather than a gate; the hard check is the merge job's --keep count assert, which fails the run if the merged fileset disagrees with the super cohort. If we want a human gate back, the cheap option is a config flag that creates the cohort but prints the phase-2 command instead of submitting it. Please confirm we are happy with unattended release before this merges.

Tests

27 new unit tests (11 of #44's run_workflow phase-check tests are replaced by entry-point equivalents); 89 total pass.