Fuse group node loads at scheduler-time — exploit new `_add_jobs_for_group` for bulk-load paths

🤖 Written by Claude

Follow-up to #346. Contingent on the new event-driven scheduler landing — this issue describes a group-level optimisation that becomes natural once `_add_jobs_for_group` is the canonical choke-point for "what runs together".

## The seam

Per `claude/analysis_node_scheduling_plan.md`, each scheduler run toposorts the schedulable set into layers and calls `_add_jobs_for_group(nodes_to_update, grp, groups, all_cache_jobs)` per layer. Within a single layer:

- All nodes have all-READY parents
- They have no dependencies on each other — can run in any order, in parallel
- Today (post-#346) they're dispatched as N independent `update_node_task` Celery tasks
- They could equally be dispatched as **one fused-load task** that issues a single CTE-shared SQL and distributes results back to per-node DB updates

This is a different shape from cross-node query optimisation (which would require building a planner). Here the scheduler already knows which nodes are ready *together*, and the new architecture explicitly groups them — the optimisation is just making one existing primitive smarter.

## Where the win is

The bulk-load paths are where this pays off:

- **Create-from-template** (extremely common): every node is DIRTY, the first scheduler run sees the entire leaf layer (typically 6–10 source/filter nodes). Currently dispatched as 6–10 parallel Celery tasks → 6–10 concurrent PG queries with massive shared substrate (cohort scan, AF lookups). Fusing into one CTE statement = the biggest possible DB-load win on the most common bulk path.
- **Force reload**: same shape — bulk DIRTY across many layers, but each layer's reschedule pass sees a wide group.
- **Steady-state user edits**: small groups (often N=1), fusion has little material to work with. No regression; falls back to single-node dispatch trivially.

Concrete profile evidence: in benchmarking analysis 11270, six leaf nodes (Cohort, Population, GeneList ×2, Phenotype, Damage ×2) each independently include the cohort filter chain. The cohort scan operator costs ~7 s per node × 6 = ~42 s of duplicated CPU per analysis load. Fusion collapses this to one ~7 s scan + ~1 s per downstream branch.

## Concrete integration

Inside `_add_jobs_for_group` (per the plan, around line 163):

```python
fusable, non_fusable = _partition_fusable(grp_nodes_to_update)
if len(fusable) >= 2:
jobs.append(fused_update_task.signature(args=[]))
for node in non_fusable:
jobs.append(node.get_update_task())
```

`_partition_fusable` decides which nodes can be load-expressed as a `SELECT variant_id WHERE ` projection — count + PK list, no per-node annotations or ordering required. Most filter nodes qualify; output/grid nodes don't.

`fused_update_task` runs one statement:

```sql
WITH common_substrate AS MATERIALIZED (
-- shared filter substrate, built from group's longest common Q prefix
-- e.g. cohort filter scan, shared AF/annotation joins
)
SELECT 'node_a' AS node, variant_id FROM common_substrate WHERE
UNION ALL
SELECT 'node_b' AS node, variant_id FROM common_substrate WHERE
UNION ALL ...
```

Reads back, splits by `node` label, runs each node's post-load (set count, mark READY, fire `_trigger_rescheduling`).

## Side benefits

1. **NodeCache populated as a byproduct.** While materialising `common_substrate` and per-node PKs anyway, write the per-node PK lists into `VariantCollection`s. Downstream layers' first scheduler run gets free cache hits — exactly the #1551 / #546 family of wins, but populated as a byproduct of fused execution rather than its own opt-in mechanism.
2. **Counts come for free.** The "N separate `SELECT COUNT(*)`" pattern collapses: each fused branch knows its row count from the result set. No separate count query needed.

## Practical caveats

- `_partition_fusable` must be conservative: any node that requires annotation aliases the fused projection doesn't carry, or distinct/order, falls out to per-node dispatch. The fallback is exactly today's behaviour.
- Result-set size matters. For a layer of 6 nodes whose outputs are 100 k variants each, the fused result is 600 k labelled rows. Worth a row-count cap that pushes huge results back to per-node dispatch.
- The `disable_cache=True` semantics from MergeNode (commit ad35a7fb1, see #240 / #546) still apply — the fused query must assemble each node's residual Q correctly, bypassing parent NodeCache substitution where needed.
- Distributed lock still on `NodeTask.unique_together`. The fused task locks all participating nodes' NodeTasks atomically (or it falls back to per-node).
- The "longest common Q prefix" detection wants careful spec — naive Q-hash equality works for trivial cases, but for partial overlap (e.g. nodes A and B share filters X, Y; node C shares only X) the substrate decomposition needs thought.

## Why now is the right time to scope this

Earlier discussion in #546 floated cross-node query fusion as a general optimisation; the practical answer was "the gain is real but the cost of building a query optimiser on top of Django's ORM is too high". The new scheduler architecture from #346 changes the picture — group decisions are already being made at a single named choke-point, and bulk-load paths (template create / force reload) are the dominant case where shared substrate is most pronounced.

This is contingent on #346 landing first; the seam doesn't exist in the current scheduler.

## Acceptance criteria (rough)

- [ ] `_partition_fusable` returns the safe-to-fuse subset of a group, with a documented predicate
- [ ] `fused_update_task` exists and handles result splitting + per-node post-load
- [ ] At least the create-from-template path measurably reduces total CPU/DB-load (target: ~5× reduction in cohort-scan operator time on benchmark analyses with shared cohort substrate)
- [ ] Side-effect: per-node `VariantCollection`s populated from the fused result, picked up by downstream layers as warm-cache hits
- [ ] Steady-state single-node load path is unchanged in code paths and timing
- [ ] Row-count cap and per-node-fallback for huge groups