diff --git a/cli/ffrwd/lower.py b/cli/ffrwd/lower.py
index 143c3f6..1000218 100644
--- a/cli/ffrwd/lower.py
+++ b/cli/ffrwd/lower.py
@@ -16629,7 +16629,15 @@ def _node_instance(
                 if number not in numbers_given:
                     numbers_given.append(number)
                 feeds.append((number, one, conformed))
-            ports[hold.port_param] = numbers_given if listed else numbers_given[0]
+            if listed:
+                # Every many input of a group lists the group's connections
+                # in one param, in the order they were first written.
+                given = ports.get(hold.port_param)
+                listing = list(given) if isinstance(given, list) else []
+                listing += [number for number in numbers_given if number not in listing]
+                ports[hold.port_param] = listing
+            else:
+                ports[hold.port_param] = numbers_given[0]
         if ports:
             # The port is one of the call's params, and the shape is asked
             # for the params the call ends up with.
@@ -17236,7 +17244,10 @@ def _check_feeder_groups(self) -> None:
         group split across sources is one whole set of calls per source. A
         source that feeds only some of them, such as the picture from one
         lateral and the sound from another, would leave each call without
-        its partner, and is refused.
+        its partner, and is refused. A group whose every connection reaches
+        one and the same call is that call's many inputs, a stream of each
+        per connection, and a connection may bring some of them and not
+        others.
         """
         by_group: dict[str, list[tuple[frozenset[str], int, exp.Expr]]] = {}
         for (group, sources), (port, argument) in self._feeder_groups.items():
@@ -17250,10 +17261,16 @@ def reached(port: int) -> list[str]:
                     {f"{call.function}({call.param})" for call in self._feeds[port].calls}
                 )
 
+            def nodes(port: int) -> set[str]:
+                return {call.node for call in self._feeds[port].calls}
+
             first_sources, first_port, _ = connections[0]
+            one_call = len(nodes(first_port)) == 1
             for sources, port, argument in connections[1:]:
                 if reached(port) == reached(first_port):
                     continue
+                if one_call and nodes(port) == nodes(first_port):
+                    continue
                 raise _error(
                     ErrorCode.UNSUPPORTED_SQL,
                     f"the group '{group}' reads {_listed_sources(first_sources)} in "
diff --git a/cli/ffrwd/registry.py b/cli/ffrwd/registry.py
index adf4f92..8b24557 100644
--- a/cli/ffrwd/registry.py
+++ b/cli/ffrwd/registry.py
@@ -154,12 +154,12 @@
 import re
 import subprocess
 import tempfile
+import threading
 from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Literal
 
 from ffrwd import binaries
-from ffrwd.errors import FfrwdError
 from ffrwd.ir import StreamType
 
 _TIMEOUT_SECONDS = 10.0
@@ -743,6 +743,9 @@ def __init__(self) -> None:
         self._ffmpeg: str | None = None
         self._version_line: str | None = None
         self._loaded = False
+        # Run-time lateral instances compile on threads of their own, and
+        # the first to read the registry loads it while the others wait.
+        self._loading = threading.Lock()
         self._filters: dict[str, DynamicFilter] = {}
         self._sources: dict[str, SourceFilter] = {}
         self._options: dict[str, dict[str, FilterOption]] = {}
@@ -753,15 +756,16 @@ def __init__(self) -> None:
     def _ensure_loaded(self) -> None:
         if self._loaded:
             return
-        self._loaded = True
-
-        try:
-            ffmpeg = binaries.ffmpeg_path()
-        except FfrwdError:
-            # FFRWD_OFFLINE refused the provider's download: asked again,
-            # it says so again.
-            self._loaded = False
-            raise
+        with self._loading:
+            if self._loaded:
+                return
+            # FFRWD_OFFLINE refusing the provider's download raises, and
+            # leaves it unloaded: asked again, it says so again.
+            self._load()
+            self._loaded = True
+
+    def _load(self) -> None:
+        ffmpeg = binaries.ffmpeg_path()
         if ffmpeg is None:
             return
         version_line = _get_version_line(ffmpeg)
@@ -893,14 +897,16 @@ def to_snapshot_payload(self) -> dict[str, object] | None:
 
 
 _registry: Registry | None = None
+_making = threading.Lock()
 
 
 def load() -> Registry:
     """Return the process-wide Registry singleton (memoized)."""
     global _registry
-    if _registry is None:
-        _registry = Registry()
-    return _registry
+    with _making:
+        if _registry is None:
+            _registry = Registry()
+        return _registry
 
 
 def load_reference(path: str | Path) -> Registry:
diff --git a/cli/tests/exec/test_exec_laterals.py b/cli/tests/exec/test_exec_laterals.py
index 7f12b1d..4a15771 100644
--- a/cli/tests/exec/test_exec_laterals.py
+++ b/cli/tests/exec/test_exec_laterals.py
@@ -48,6 +48,7 @@
 _PROBE = _RELEASE / "feed_probe.wasm"
 _HEAR = _RELEASE / "feed_probe_audio.wasm"
 _STAMP = _RELEASE / "data_stamp.wasm"
+_GROUP = _RELEASE / "shape_group.wasm"
 _LAUNCH = _CLI_ROOT / "tests" / "data" / "launch.nut"
 _TIMEOUT = 120.0
 
@@ -75,6 +76,24 @@
 """
 
 
+# play with both columns; an instance whose sound no feeder reads binds its
+# sound's values by DEFAULT.
+_BOTH_PLAY = """CREATE FUNCTION play(launch data_stream, url text, start_pts number,
+                     duration number, width number, height number, fps number,
+                     pix_fmt text, rate number DEFAULT 48000,
+                     channels number DEFAULT 2, sample_fmt text DEFAULT 'f32')
+RETURNS TABLE(video video_stream, audio audio_stream) AS $$
+  SELECT setpts(ffmpeg.format(fps(scale(m.video[1], width, height), fps),
+                              pix_fmts => pix_fmt),
+                'PTS+' || start_pts::text || '/TB') AS video,
+         asetpts(ffmpeg.aformat(aresample(m.audio[1], rate), sample_fmts => sample_fmt,
+                                channel_layouts => channels::text || 'c'),
+                 'PTS+' || start_pts::text || '/TB') AS audio
+  FROM input(url) m
+$$ LANGUAGE sql;
+"""
+
+
 @pytest.fixture(autouse=True)
 def _require_everything() -> None:
     if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None:
@@ -242,3 +261,57 @@ def test_a_data_filters_output_reaches_the_lateral_and_the_file_both(
     written = _data_messages(out)
     assert [at for at, _ in written] == [0.1, 0.2, 0.3, 0.5]
     assert {message["node"] for _, message in written} == {"leaf"}
+
+
+def test_two_laterals_on_one_many_hold_group_are_a_connection_and_an_offset_each(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The fleet's ``shape-group`` takes many pictures and many sounds on one
+    hold group. `m` brings its picture and its sound, `c` its picture alone,
+    stamped a second later: each is a connection of its own, live beside the
+    other on its own origin, and `m`'s picture and sound share each
+    instance's offset."""
+    if not _GROUP.exists():
+        pytest.skip(f"module missing: {_GROUP}")
+    programme = _media(tmp_path)
+    monkeypatch.chdir(tmp_path)
+    fed = tmp_path / "fed.ndjson"
+    late = _BOTH_PLAY.replace("play(", "late(", 1).replace(
+        "'PTS+' || start_pts::text", "'PTS+' || (start_pts + 1)::text"
+    )
+    sql = (
+        "CREATE FUNCTION grp(c video_stream, a audio_stream, v video_stream[] DEFAULT NULL, "
+        "s audio_stream[] DEFAULT NULL, ports number DEFAULT NULL)\nRETURNS STRUCT("
+        "c video_stream, feeds STRUCT(pts number, member text, k number, live boolean, "
+        "at_pts number, first_pts number)[])\n"
+        f"AS '{_GROUP.as_posix()}', 'shape_group' LANGUAGE wasm;\n" + _BOTH_PLAY + late
+        + "COPY (SELECT grp(p.video[1], p.audio[1], ARRAY[m.video, c.video], "
+        "ARRAY[m.audio]).feeds FROM "
+        f"input('{programme.as_posix()}', realtime => true) p, "
+        f"input('{_LAUNCH.as_posix()}') f, LATERAL play(f.data[1]) m, "
+        f"LATERAL late(f.data[1]) c) TO '{fed.as_posix()}'"
+    )
+    rows = _run(sql, capsys)
+    feeders = [row for row in rows if row.get("event") == "feeder"]
+    assert len([row for row in feeders if row.get("exit") == 0]) == 4, feeders
+    said = [json.loads(line) for line in fed.read_text().splitlines() if line]
+
+    def starts(member: str, k: int, field: str = "at_pts") -> list[int]:
+        return [
+            int(row[field]) for row in said
+            if (row["member"], row["k"], row["live"]) == (member, k, True)
+        ]
+
+    # Two instances of each lateral, each anchored as it connects.
+    assert len(starts("v", 0)) == 2, said
+    assert len(starts("v", 1)) == 2, said
+    assert starts("s", 0) == starts("v", 0), "m's sound shares its picture's offset"
+    assert starts("s", 1) == [], "c brings no sound"
+    firsts = list(zip(starts("v", 0, "first_pts"), starts("v", 1, "first_pts"), strict=True))
+    assert all(mine < its for mine, its in firsts), f"c's feeds start on c's pts: {firsts}"
+    live: dict[tuple[str, int], bool] = {}
+    both = False
+    for row in said:
+        live[(str(row["member"]), int(row["k"]))] = bool(row["live"])
+        both = both or (live.get(("v", 0), False) and live.get(("v", 1), False))
+    assert both, f"each connection is live while the other is: {said}"
diff --git a/cli/tests/test_laterals.py b/cli/tests/test_laterals.py
index 57a1a12..28bc9f9 100644
--- a/cli/tests/test_laterals.py
+++ b/cli/tests/test_laterals.py
@@ -1025,6 +1025,86 @@ def test_a_port_written_by_hand_stays_one_integer_where_a_list_also_fits() -> No
     assert port == 9100 and isinstance(port, int)
 
 
+# compose taking each lateral's sound beside its picture, on the same group.
+_COMPOSE_AUDIO = (
+    "CREATE FUNCTION compose(v video_stream, inputs video_stream[] DEFAULT NULL, "
+    "audio audio_stream[] DEFAULT NULL, port number DEFAULT NULL) RETURNS video_stream "
+    f"AS '{COMPOSE}', 'compose' LANGUAGE wasm;"
+)
+
+
+def _compose_audio_shape(
+    module: str, params: str, bound: Sequence[shapes.Binding], grants: Sequence[str] = ()
+) -> shapes.NodeShape:
+    """compose with `audio`, many sounds held on the pictures' port."""
+    held = {
+        "kind": "hold",
+        "anchor": {"kind": "tagged", "tag": "smart_timed"},
+        "lead": 0.3,
+        "port_param": "port",
+    }
+
+    def port(
+        name: str, kind: str, pairing: dict[str, object], many: bool
+    ) -> dict[str, object]:
+        return {
+            "name": name, "kind": kind, "required": name == "v", "many": many,
+            "pairing": pairing, "rows": "ignore", "window": 1, "stride": 1, "accepts": {},
+        }
+
+    return shapes.node_shape(module, {
+        "inputs": [
+            port("v", "video", {"kind": "lockstep"}, False),
+            port("inputs", "video", held, True),
+            port("audio", "audio", held, True),
+        ],
+        "outputs": [{"name": "v", "kind": "video", "latency": 0,
+                     "format": {"kind": "like", "port": "v"}}],
+        "clock": {"kind": "input", "port": "v"},
+        "pure": True, "one_to_one": True, "bounded": True, "relation": [],
+    })
+
+
+def test_a_lateral_bringing_a_picture_alone_to_a_call_taking_sound_too_is_its_connection() -> None:
+    """ad brings compose its picture and its sound and lbar its picture alone:
+    both connections reach the one call, so the group is not split, and the
+    port list names each connection once, in the order first written."""
+    query = _TWO_ADS.replace(
+        "ARRAY[ad.video, lbar.video])", "ARRAY[ad.video, lbar.video], audio => ARRAY[ad.audio])"
+    )
+    plan = compile_all(
+        _declared(query).replace("COPY (", _COMPOSE_AUDIO + "\nCOPY (", 1),
+        describe=lambda path: _compose_modules({"type": "array", "items": {"type": "integer"}})[
+            path
+        ],
+        shape=_compose_audio_shape,
+    ).plan
+    assert plan is not None
+    assert [len(lateral.connections) for lateral in plan.laterals] == [1, 1]
+    ports = [lateral.connections[0].port for lateral in plan.laterals]
+    assert len(set(ports)) == 2
+    assert _compose_node(plan).args["port"] == ports
+
+
+def test_a_group_whose_connections_reach_different_calls_is_still_refused() -> None:
+    """The picture from one lateral into the picture switch and the sound from
+    another into the sound switch: neither switch finds its partner."""
+    error = _refused(
+        """COPY (
+  WITH prog AS (SELECT s.video[1] AS v, s.audio[1] AS a, s.data[1] AS d
+                FROM input('leaf.nut') s)
+  SELECT video(prog.v, ad.video), audio(prog.a, lbar.audio)
+  FROM prog, LATERAL play(prog.d) ad, LATERAL play(prog.d) lbar
+) TO 'out.nut'"""
+    )
+    assert error.code == ErrorCode.UNSUPPORTED_SQL
+    assert error.message.startswith("the group 'switch' reads ")
+    assert error.message.endswith(
+        "each source a group reads is a connection of its own, and each has to reach "
+        "the same calls"
+    )
+
+
 # -- a node's data, read by several and by a lateral, with no copying ffmpeg ---
 
 SELL = "modules/sell_node.wasm"
diff --git a/cli/tests/test_registry.py b/cli/tests/test_registry.py
index 832d58f..32caecd 100644
--- a/cli/tests/test_registry.py
+++ b/cli/tests/test_registry.py
@@ -26,6 +26,8 @@
 
 import json
 import subprocess
+import threading
+import time
 from collections.abc import Iterator
 from pathlib import Path
 
@@ -1567,3 +1569,30 @@ def counting_run(argv: list[str], **kwargs: object) -> subprocess.CompletedProce
     help_calls = [c for c in calls if any(a.startswith("filter=") for a in c)]
     assert len(help_calls) == 2  # exactly gblur + overlay, never all ~460+
     assert sum(1 for c in calls if "-filters" in c) == 1
+
+
+def test_a_registry_read_on_two_threads_at_once_is_loaded_once_and_whole(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Two lateral instances compiling at once: the second waits for the
+    first's load rather than finding no filters."""
+    _fake_ffmpeg_present(monkeypatch)
+    calls = _fake_run(monkeypatch)
+    parse = registry_mod._parse_filters_list
+    started = threading.Event()
+
+    def slow_parse(ffmpeg: str) -> object:
+        started.set()
+        time.sleep(0.2)
+        return parse(ffmpeg)
+
+    monkeypatch.setattr(registry_mod, "_parse_filters_list", slow_parse)
+    reg = load()
+    found: list[object] = []
+    first = threading.Thread(target=lambda: found.append(reg.get("gblur")))
+    first.start()
+    assert started.wait(5)
+    found.append(reg.get("gblur"))
+    first.join()
+    assert found[0] is not None and found[0] == found[1]
+    assert sum("-filters" in argv for argv in calls) == 1
diff --git a/docs/dialect.md b/docs/dialect.md
index 95fa201..19d14f8 100644
--- a/docs/dialect.md
+++ b/docs/dialect.md
@@ -634,6 +634,12 @@ dest    := 'path' | STDOUT | ( value-expression ) | sink(value, ...)
     `port_param`) is given a stream, or a port number: written in its
     place, or by the param's name, `inset(v, port => 9100)`. Given a
     number it binds nothing, and whatever connects to that port is shown.
+  - A `[]` input held on a port and given run-time laterals takes one
+    connection per lateral, every port in one list in its param. Inputs
+    of one group share that list, so a lateral's picture and sound arrive
+    on one connection with one offset, and a lateral may bring some of a
+    group's inputs and not others:
+    `compose(k.v, ARRAY[m.video, c.video], audio => ARRAY[m.audio])`.
   - `RETURNS` a stream; rows alone (`STRUCT(...)[]`, `cue[]`), a data
     stream the query reads while it runs; `STRUCT(<name> <type>, ...)`,
     one field per output the module makes, read off the call
diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md
index 9eb561a..4e13910 100644
--- a/sidecar/NODE-CLI.md
+++ b/sidecar/NODE-CLI.md
@@ -272,9 +272,13 @@ it is conformed ends `on the GPU`. A coded picture for such an input is
 decoded on the process's device where its decoder shares it (`d3d12va` on
 Windows), as `decode` does for a reader on the GPU, so it never crosses
 host memory; elsewhere it is decoded in software and uploaded once. A
-group of
-hold inputs (`hold.group`) shares one listener and one connection, its
-picture to the video port and its sound to the audio port.
+group of hold inputs (`hold.group`) is one per connection: the inputs of a
+group share each port's listener and connection, its picture to the video
+input and its sound to the audio input. Many inputs (`many`) given a list
+of ports take a stream each per port, in the list's order, so the k-th
+stream of each input arrives on the k-th port. Each connection has its
+own source, feed and offset: a second port's feeder connecting, dropping
+or rejoining leaves the first's feed as it was.
 
 One connection at a time: a second while one is on is accepted and closed
 at once, and a connection that has sent nothing is dropped for one that
@@ -285,7 +289,7 @@ tags and time base from its start on; a source whose pts jump backwards or
 more than a second forwards is a new feed on the same connection. A
 connection may carry any of a group's members, and streams beyond them are
 left on the wire: a feeder of sound alone feeds the sound. Every member of
-a group is told the lead's offset, the picture's where the connection
+a connection is told its lead's offset, the picture's where the connection
 brings one and otherwise its first member's: its `feed-start.at` is the
 lead's, and its `first-pts` the point of its own stream that offset puts
 there, so `(at, first-pts)` maps any member's pts onto the clock. A member
@@ -293,7 +297,8 @@ the connection does not bring (not sent, or its sound refused) gets no feed
 at all on any tick, where one not arrived yet gets a feed and no frames.
 
 A data input whose `interval` names the group (`interval.group`) arrives
-on the same connection, unbound by any pad, as the hold inputs do. What a
+on the group's first port, unbound by any pad, and is placed with that
+connection's offset. What a
 feeder writes so its rows land there, in one NUT:
 
 - the picture (raw video, rgba or yuv420p) and the sound (PCM) as above,
@@ -317,6 +322,8 @@ it. A connection's messages never outlive its feed: those queued when it
 ends are dropped with it. A group data input bound by a pad is refused;
 with its group's hold inputs bound to streams rather than a port, it is
 bound to a stream like any data input and placed with that group's offset.
+A group bound to streams pairs its inputs' streams by position: the first
+stream of each input is one group with one offset, the second another.
 
 The host says what it did on stderr, as lines and as rows behind
 `ffrwd:row `: `{"kind":"listen",...}` once the port is bound,
diff --git a/sidecar/ffrwd-wasm/src/hold.rs b/sidecar/ffrwd-wasm/src/hold.rs
index 7a1fc76..242eb42 100644
--- a/sidecar/ffrwd-wasm/src/hold.rs
+++ b/sidecar/ffrwd-wasm/src/hold.rs
@@ -1,7 +1,7 @@
 //! Hold pairing: a source's frames paired by time onto a node's clock.
 //!
-//! One [`Group`] serves one hold input, or every hold input of one `group`
-//! (a feeder's picture and sound on one connection). Frames arrive as the
+//! One [`Group`] serves one hold input, or the hold inputs of one `group`
+//! on one connection (a feeder's picture and sound). Frames arrive as the
 //! source delivers them and queue per member; each tick, a video member
 //! hands the newest frame at or before the tick, repeating the last one
 //! while the source is behind and skipping forward when it catches up, and
diff --git a/sidecar/ffrwd-wasm/src/node_graph.rs b/sidecar/ffrwd-wasm/src/node_graph.rs
index 4345849..198a650 100644
--- a/sidecar/ffrwd-wasm/src/node_graph.rs
+++ b/sidecar/ffrwd-wasm/src/node_graph.rs
@@ -1619,9 +1619,9 @@ fn open_module(
         &schema,
         defs,
     )?;
-    let port_fed: Vec<u32> = feeds
+    let port_fed: Vec<(u32, u16)> = feeds
         .iter()
-        .flat_map(|f| f.members.iter().map(|m| m.id))
+        .flat_map(|f| f.members.iter().map(|m| (m.id, f.port)))
         .collect();
     let latched: Vec<String> = call
         .outputs
diff --git a/sidecar/ffrwd-wasm/src/tick.rs b/sidecar/ffrwd-wasm/src/tick.rs
index e4f8465..f3b668b 100644
--- a/sidecar/ffrwd-wasm/src/tick.rs
+++ b/sidecar/ffrwd-wasm/src/tick.rs
@@ -591,14 +591,28 @@ pub struct Assembler {
     report: Report,
 }
 
+/// A hold group's name and connection; none for a hold input of no group,
+/// which is a group of its own.
+type GroupKey = Option<(String, Connection)>;
+
+/// The connection a stream of a hold group arrives on: the port of a port
+/// feed, or for a stream bound in the plan its place among its input's
+/// streams. A group is one per name and connection.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Connection {
+    Port(u16),
+    Bound(usize),
+}
+
 impl Assembler {
     /// The ticks of `name`, a node of `shape` bound to `bound`. `port_fed`
-    /// names the hold inputs a port feed serves, which never hold a tick.
+    /// names the streams a port feed serves, each with its port; their hold
+    /// inputs never hold a tick.
     pub fn new(
         shape: &NodeShape,
         bound: &[BoundStream],
         name: &str,
-        port_fed: &[u32],
+        port_fed: &[(u32, u16)],
     ) -> Result<Assembler> {
         let clock_base = match &shape.clock {
             Clock::Input(name) => bound.iter().find(|b| &b.port == name).map(|b| b.time_base),
@@ -610,9 +624,23 @@ impl Assembler {
         };
         let mut inputs = Vec::with_capacity(bound.len());
         let mut holds: Vec<Group> = Vec::new();
-        let mut on_groups: Vec<(usize, String, Member, bool)> = Vec::new();
-        let mut grouped: Vec<(Option<String>, Hold, Vec<Member>, bool)> = Vec::new();
-        for stream in bound {
+        let mut on_groups: Vec<(usize, String, Connection, Member, bool)> = Vec::new();
+        let mut grouped: Vec<(GroupKey, Hold, Vec<Member>, bool)> = Vec::new();
+        for (index, stream) in bound.iter().enumerate() {
+            let fed_on = port_fed
+                .iter()
+                .find(|(id, _)| *id == stream.id)
+                .map(|(_, port)| *port);
+            let connection = match fed_on {
+                Some(port) => Connection::Port(port),
+                None => Connection::Bound(
+                    bound[..index]
+                        .iter()
+                        .filter(|b| b.port == stream.port)
+                        .count(),
+                ),
+            };
+            let fed = fed_on.is_some();
             let port = shape
                 .input(&stream.port)
                 .ok_or_else(|| anyhow!("no input port '{}'", stream.port))?;
@@ -637,8 +665,7 @@ impl Assembler {
                         audio,
                         ahead: 0.0,
                     };
-                    let fed = port_fed.contains(&stream.id);
-                    let key = hold.group.clone();
+                    let key = hold.group.clone().map(|g| (g, connection));
                     let group = match key
                         .as_ref()
                         .and_then(|k| grouped.iter().position(|(g, ..)| g.as_ref() == Some(k)))
@@ -661,6 +688,7 @@ impl Assembler {
                     on_groups.push((
                         inputs.len(),
                         group.clone(),
+                        connection,
                         Member {
                             name: port.name.clone(),
                             kind: port.kind,
@@ -669,7 +697,7 @@ impl Assembler {
                             audio: None,
                             ahead: *ahead,
                         },
-                        port_fed.contains(&stream.id),
+                        fed,
                     ));
                     None
                 }
@@ -700,11 +728,19 @@ impl Assembler {
                 restamp,
             });
         }
-        for (index, key, member, fed) in on_groups {
-            let Some(group) = grouped
-                .iter()
-                .position(|(g, ..)| g.as_deref() == Some(key.as_str()))
-            else {
+        for (index, key, connection, member, fed) in on_groups {
+            let named = |g: &GroupKey| g.as_ref().is_some_and(|(g, _)| *g == key);
+            let Some(group) = grouped.iter().position(|(g, ..)| {
+                g.as_ref()
+                    .is_some_and(|(g, c)| *g == key && *c == connection)
+            }) else {
+                if grouped.iter().any(|(g, ..)| named(g)) {
+                    bail!(
+                        "input '{}' arrives on group '{key}', and none of that group's hold \
+                         inputs arrives on its connection",
+                        member.name
+                    );
+                }
                 bail!(
                     "input '{}' arrives on group '{key}', and this call binds none of that \
                      group's hold inputs",
@@ -1961,7 +1997,7 @@ mod tests {
                 bound("feed_audio", 2, KHZ48, audio()),
             ],
             "m",
-            &[1, 2],
+            &[(1, 9000), (2, 9000)],
         )
         .expect("assembler");
         let source = |base| crate::hold::SourceInfo {
@@ -1995,6 +2031,64 @@ mod tests {
         assert_eq!(sound.info.as_ref().map(|(_, base)| *base), Some(KHZ48));
     }
 
+    #[test]
+    fn a_group_bound_in_the_plan_pairs_its_many_inputs_stream_by_stream() {
+        let hold = Pairing::Hold(ffrwd_wasm_runtime::node::Hold {
+            anchor: ffrwd_wasm_runtime::node::Anchor::FirstFrame,
+            lead: 0.0,
+            linger: None,
+            timeout: None,
+            group: Some("g".to_string()),
+            port_param: None,
+        });
+        let many = |name: &str, kind: PortKind| InputPort {
+            many: true,
+            ..port(name, kind, hold.clone(), RowsUse::Ignore)
+        };
+        let s = shape(
+            vec![
+                port("v", PortKind::Video, Pairing::Lockstep, RowsUse::Ignore),
+                many("feed", PortKind::Video),
+                many("feed_audio", PortKind::Audio),
+            ],
+            Clock::Input("v".into()),
+        );
+        let tb = TimeBase { num: 1, den: 30 };
+        let mut a = Assembler::new(
+            &s,
+            &[
+                bound("v", 0, tb, video()),
+                bound("feed", 1, tb, video()),
+                bound("feed", 2, tb, video()),
+                bound("feed_audio", 3, KHZ48, audio()),
+                bound("feed_audio", 4, KHZ48, audio()),
+            ],
+            "m",
+            &[],
+        )
+        .expect("assembler");
+        for k in 0..3 {
+            a.arrive(0, frame(k, &[])).expect("video");
+            a.arrive(1, frame(k, &[])).expect("feed");
+            a.arrive(2, frame(100 + k, &[])).expect("feed");
+            a.arrive(3, samples(k * 1600, 1600)).expect("sound");
+            a.arrive(4, samples((100 + k) * 1600, 1600)).expect("sound");
+        }
+        let ticks = drain(&mut a);
+        let start = |id: usize| {
+            let feed = ticks[0].streams[id].feed.as_ref().expect("a feed");
+            (feed.start.at, feed.start.first_pts)
+        };
+        assert_eq!(start(1), (0, 0));
+        assert_eq!(
+            start(2),
+            (0, 100),
+            "the second picture is a feed of its own, anchored on its own first frame"
+        );
+        assert_eq!(start(3), (0, 0), "the first sound beside the first picture");
+        assert_eq!(start(4), (0, 160_000), "the second sound beside the second");
+    }
+
     #[test]
     fn a_data_clock_ticks_once_per_pts_and_a_packets_clock_on_its_running_dts() {
         let s = shape(
@@ -2162,8 +2256,8 @@ mod tests {
             Clock::Rate(Rational { num: 30, den: 1 }),
         );
         let tb = TimeBase { num: 1, den: 30 };
-        let mut a =
-            Assembler::new(&s, &[bound("feed", 4, tb, video())], "m", &[4]).expect("assembler");
+        let mut a = Assembler::new(&s, &[bound("feed", 4, tb, video())], "m", &[(4, 9000)])
+            .expect("assembler");
         a.arrive(4, frame(0, &[])).expect("a frame");
         let mut heard: Vec<Vec<(u64, usize, i64)>> = vec![Vec::new(); 3];
         for _ in 0..10 {
diff --git a/sidecar/ffrwd-wasm/tests/live.rs b/sidecar/ffrwd-wasm/tests/live.rs
index 374c5ff..95f8e20 100644
--- a/sidecar/ffrwd-wasm/tests/live.rs
+++ b/sidecar/ffrwd-wasm/tests/live.rs
@@ -32,6 +32,7 @@ const MODULES: &[&str] = &[
     "shape-window",
     "shape-hold",
     "shape-probe",
+    "shape-group",
 ];
 
 const RATE: u32 = 48_000;
@@ -676,7 +677,8 @@ fn start_padded(
     if chain.contains("[feeds=f]") || chain.contains("[spots=f]") {
         command.args(["-map", "[f]", "-f", "ndjson", &feeds]);
     }
-    if chain.contains("[clock=c]") {
+    // A second rows output of the node rides the same file as `clock`.
+    if chain.contains("[clock=c]") || chain.contains("[cues=c]") {
         command.args(["-map", "[c]", "-f", "ndjson", &clock]);
     }
     command.stdin(Stdio::piped());
@@ -1637,3 +1639,137 @@ fn rows_a_feeder_writes_beside_its_picture_and_sound_land_with_the_groups_offset
         out.stderr
     );
 }
+
+/// The feeds `shape_group` said one stream of `input` had, in order: each
+/// start as (at, first_pts), or None where its feed went off.
+fn group_feeds(out: &Output, input: &str, k: u64) -> Vec<(i64, Option<(i64, i64)>)> {
+    out.feeds
+        .iter()
+        .filter(|r| r["member"] == input && r["k"].as_u64() == Some(k))
+        .map(|r| {
+            let start = (
+                r["at_pts"].as_i64().unwrap(),
+                r["first_pts"].as_i64().unwrap(),
+            );
+            (
+                r["pts"].as_i64().unwrap(),
+                r["live"].as_bool().unwrap().then_some(start),
+            )
+        })
+        .collect()
+}
+
+#[test]
+fn each_connection_of_a_many_hold_group_is_a_feed_of_its_own_with_its_sound_beside_its_picture() {
+    let _serial = serial();
+    let program = Program::new(30, 9.0);
+    let (first, second) = (free_port(), free_port());
+    let dir = scratch("group-many-params");
+    let params = dir.join("params.json");
+    fs::write(&params, format!(r#"{{"ports":[{first},{second}]}}"#)).expect("write the params");
+    let from = format!("shape_group={}", params.display());
+    let (mut run, _) = start_padded(
+        "group-many",
+        &program,
+        "1",
+        &["shape_group"],
+        "[c=0:v][a=0:a]shape_group[c=o][feeds=f][cues=c]",
+        &["-params-from", &from],
+    );
+    wait_for_port(&mut run, first);
+    wait_for_port(&mut run, second);
+    // The first connection stays on throughout; the second comes, goes and
+    // comes back on another origin while it is.
+    let mut main = Feeder::new(30, 6.0);
+    main.start = 10.0;
+    main.rows = vec![
+        (10_200, r#"{"text":"a"}"#.to_string()),
+        (12_000, r#"{"text":"b"}"#.to_string()),
+    ];
+    let mut again = Feeder::new(30, 1.5);
+    again.start = 20.0;
+    let fed = [
+        feed(first, main, Duration::from_millis(500)),
+        feed(second, Feeder::new(30, 1.5), Duration::from_millis(1500)),
+        feed(second, again, Duration::from_millis(4500)),
+    ];
+    let out = finish(run);
+    for one in fed {
+        one.join().expect("a feeder");
+    }
+    let _ = fs::remove_dir_all(&dir);
+
+    let main_feeds = group_feeds(&out, "v", 0);
+    let second_feeds = group_feeds(&out, "v", 1);
+    assert!(
+        matches!(
+            main_feeds.as_slice(),
+            [(_, Some(_)), (_, None)] | [(_, Some(_))]
+        ),
+        "the first connection is one feed, on until it ends: {main_feeds:?}\n{}",
+        out.stderr
+    );
+    let (main_at, main_first) = main_feeds[0].1.unwrap();
+    assert_eq!(main_first, 300, "its own origin, ten seconds in");
+    let starts: Vec<i64> = second_feeds
+        .iter()
+        .filter_map(|(_, start)| start.map(|s| s.1))
+        .collect();
+    assert_eq!(
+        starts,
+        vec![0, 600],
+        "the second connection is anchored again when it rejoins: {second_feeds:?}\n{}",
+        out.stderr
+    );
+    let rejoined = second_feeds
+        .iter()
+        .find(|(_, start)| start.is_some_and(|s| s.1 == 600))
+        .unwrap()
+        .0;
+    assert!(
+        main_feeds.get(1).is_none_or(|(off, _)| *off > rejoined),
+        "the first is still on when the second rejoins: {main_feeds:?} {second_feeds:?}"
+    );
+    for k in 0..2 {
+        let picture = group_feeds(&out, "v", k);
+        let sound = group_feeds(&out, "s", k);
+        let on = |feeds: &[(i64, Option<(i64, i64)>)]| -> Vec<(i64, i64)> {
+            feeds.iter().filter_map(|(_, start)| *start).collect()
+        };
+        let (picture, sound) = (on(&picture), on(&sound));
+        assert_eq!(
+            picture.iter().map(|s| s.0).collect::<Vec<_>>(),
+            sound.iter().map(|s| s.0).collect::<Vec<_>>(),
+            "connection {k}'s picture and sound share each feed's offset\n{}",
+            out.stderr
+        );
+        for (p, s) in picture.iter().zip(&sound) {
+            let apart = p.1 as f64 / 30.0 - s.1 as f64 / f64::from(RATE);
+            assert!(
+                apart.abs() <= 1.0 / 30.0,
+                "connection {k}'s sound starts where its picture does: {p:?} {s:?}"
+            );
+        }
+    }
+    let cues: Vec<(i64, i64, String)> = out
+        .clock
+        .iter()
+        .map(|r| {
+            let text: Value = serde_json::from_str(r["text"].as_str().unwrap()).unwrap();
+            (
+                r["pts"].as_i64().unwrap(),
+                r["cue_pts"].as_i64().unwrap(),
+                text["text"].as_str().unwrap().to_string(),
+            )
+        })
+        .collect();
+    assert_eq!(
+        cues,
+        vec![
+            (main_at + 6, main_at + 6, "a".to_string()),
+            (main_at + 60, main_at + 60, "b".to_string()),
+        ],
+        "the first connection's rows land with its own offset, the second on or not:\n{}",
+        out.stderr
+    );
+}
diff --git a/sidecar/modules/Cargo.lock b/sidecar/modules/Cargo.lock
index 19e7577..05ac294 100644
--- a/sidecar/modules/Cargo.lock
+++ b/sidecar/modules/Cargo.lock
@@ -1955,6 +1955,15 @@ dependencies = [
  "wit-bindgen 0.57.1",
 ]
 
+[[package]]
+name = "shape-group"
+version = "0.1.0"
+dependencies = [
+ "ffrwd-node",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "shape-hold"
 version = "0.1.0"
diff --git a/sidecar/modules/Cargo.toml b/sidecar/modules/Cargo.toml
index f5db833..328d99f 100644
--- a/sidecar/modules/Cargo.toml
+++ b/sidecar/modules/Cargo.toml
@@ -100,6 +100,8 @@ members = [
     # What the live switch did, as a node: the host's hold pairing, feeds by
     # port and both anchors, with none of the switch's own code.
     "shape-switch",
+    # Many pictures and many sounds on one hold group, a connection each.
+    "shape-group",
 ]
 resolver = "2"
 
diff --git a/sidecar/modules/shape-group/Cargo.toml b/sidecar/modules/shape-group/Cargo.toml
new file mode 100644
index 0000000..6d60484
--- /dev/null
+++ b/sidecar/modules/shape-group/Cargo.toml
@@ -0,0 +1,13 @@
+[package]
+name = "shape-group"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+name = "shape_group"
+crate-type = ["cdylib"]
+
+[dependencies]
+ffrwd-node = { git = "https://github.com/imbcmdth/ffrwd-node", tag = "v0.2.0" }
+serde = { version = "1.0.229", features = ["derive"] }
+serde_json = "1.0.151"
diff --git a/sidecar/modules/shape-group/src/lib.rs b/sidecar/modules/shape-group/src/lib.rs
new file mode 100644
index 0000000..b404052
--- /dev/null
+++ b/sidecar/modules/shape-group/src/lib.rs
@@ -0,0 +1,148 @@
+//! Many pictures and many sounds held on one group, as a compositor takes
+//! each lateral's picture and sound: `v` and `s` take a stream per port
+//! `ports` lists, the picture and the sound on one port arriving on one
+//! connection. `cues` takes rows on the group's first connection.
+//!
+//! It passes the clock `c` through. `feeds` has a row whenever what the
+//! host says of one stream's feed changes: its input (`member`), its place
+//! among the input's streams, whether a feed is live and, while one is,
+//! its start. `cues` has a row per message `cues` is handed, on the tick it is.
+
+use ffrwd_node::{Anchor, Bound, Init, Input, Node, Out, Output, Result, Shape, Tick};
+use serde::{Deserialize, Serialize};
+
+#[derive(Deserialize)]
+struct Params {
+    lead: f64,
+    timeout: f64,
+}
+
+#[derive(Serialize)]
+struct FeedRow<'a> {
+    pts: i64,
+    member: &'a str,
+    k: usize,
+    live: bool,
+    at_pts: i64,
+    first_pts: i64,
+}
+
+#[derive(Serialize)]
+struct CueRow {
+    pts: i64,
+    cue_pts: i64,
+    text: String,
+}
+
+/// One stream of `v` or `s`: its input, its place, its id and the start
+/// said last.
+struct Held {
+    input: &'static str,
+    k: usize,
+    id: u32,
+    said: Option<(i64, i64)>,
+}
+
+struct Group {
+    c: u32,
+    held: Vec<Held>,
+    cues: Option<u32>,
+}
+
+impl Node for Group {
+    const NAME: &'static str = "shape_group";
+    const VERSION: &'static str = "0.1.0";
+    const PARAMS_SCHEMA: &'static str = r#"{"type":"object","properties":{"ports":{"type":"array","items":{"type":"integer","minimum":1,"maximum":65535}},"lead":{"type":"number","minimum":0,"default":0.3},"timeout":{"type":"number","exclusiveMinimum":0,"default":1}},"additionalProperties":false}"#;
+    type Params = Params;
+
+    fn shape(params: &Params, _: &Bound) -> Result<Shape> {
+        let held = |input: Input| {
+            input
+                .optional()
+                .many()
+                .hold()
+                .anchor(Anchor::FirstFrame)
+                .lead(params.lead)
+                .timeout(params.timeout)
+                .group("lateral")
+                .port_param("ports")
+        };
+        Ok(Shape::new()
+            .input(Input::video("c").clock().pixel_formats(&["rgba"]))
+            .input(Input::audio("a").optional().sample_formats(&["f32"]))
+            .input(held(Input::video("v")).like("c").pixel_formats(&["rgba"]))
+            .input(held(Input::audio("s")).like("a").sample_formats(&["f32"]))
+            .input(Input::rows("cues").optional().group("lateral").per_frame())
+            .output(Output::like("c"))
+            .output(Output::rows("feeds").schema_json(
+                r#"{"type":"object","properties":{"pts":{"type":"integer"},"member":{"type":"string"},"k":{"type":"integer"},"live":{"type":"boolean"},"at_pts":{"type":"integer"},"first_pts":{"type":"integer"}}}"#,
+            ))
+            .output(Output::rows("cues").schema_json(
+                r#"{"type":"object","properties":{"pts":{"type":"integer"},"cue_pts":{"type":"integer"},"text":{"type":"string"}}}"#,
+            ))
+            .one_to_one())
+    }
+
+    fn init(_: Params, init: &Init) -> Result<Group> {
+        let mut held = Vec::new();
+        for input in ["v", "s"] {
+            for (k, stream) in init.streams(input).iter().enumerate() {
+                held.push(Held {
+                    input,
+                    k,
+                    id: stream.id,
+                    said: None,
+                });
+            }
+        }
+        Ok(Group {
+            c: init.stream("c")?.id,
+            held,
+            cues: init.optional("cues").map(|b| b.id),
+        })
+    }
+
+    fn process(&mut self, tick: &Tick, out: &mut Out) -> Result<()> {
+        let pts = tick.pts();
+        let Some(frame) = tick.frame(self.c) else {
+            return Ok(());
+        };
+        out.pass("c", self.c, &frame)?;
+        for held in &mut self.held {
+            let start = tick
+                .feed(held.id)
+                .map(|feed| (feed.start.at, feed.start.first_pts));
+            if start == held.said {
+                continue;
+            }
+            held.said = start;
+            let (at, first_pts) = start.unwrap_or_default();
+            out.row(
+                "feeds",
+                pts,
+                &FeedRow {
+                    pts,
+                    member: held.input,
+                    k: held.k,
+                    live: start.is_some(),
+                    at_pts: at,
+                    first_pts,
+                },
+            )?;
+        }
+        for message in self.cues.map(|id| tick.messages(id)).unwrap_or_default() {
+            out.row(
+                "cues",
+                pts,
+                &CueRow {
+                    pts,
+                    cue_pts: message.pts,
+                    text: String::from_utf8_lossy(&message.data).into_owned(),
+                },
+            )?;
+        }
+        Ok(())
+    }
+}
+
+ffrwd_node::export!(Group);