Deepen all 20 media-streaming modules with expanded APIs and tests.

Add batch ingest, tree depth/path, peer scoring helpers, scheduler peek, FEC and ABR controls, telemetry sessions, and richer production api.md for bandwidth-aggregator and media-tree-orchestrator.

Co-authored-by: Cursor <[email protected]>

media-streaming/README.md

@@ -22,6 +22,19 @@ Shared types: [`../_shared/media-streaming-base.js`](../_shared/media-streaming-
 | **production** | `media-tree-orchestrator`, `peer-selector-streaming`, `bandwidth-aggregator`, `chunk-scheduler-media` |
 | **scaffold** | remaining 16 (full API + tests; deepen in later waves) |
 
+## Recent API expansions
+
+- **media-chunker:** `keyframeIndices`, `totalBytes`, `pruneBefore`
+- **bandwidth-aggregator:** `ingestBatch`, `chunkAt`, `topSource`, `clearBefore`
+- **media-tree-orchestrator:** `depth`, `pathToRoot`, `hasNode`, `nodeCount`
+- **peer-selector-streaming:** `updatePeer`, `filterByRegion`, `peersAboveScore`
+- **helper-swarm-coordinator:** `unassignViewer`, `totalCapacity`, `helperUtilization`
+- **chunk-scheduler-media:** `peek`, `dequeueSeq`, `boostType`
+- **adaptive-streaming-engine:** `canSustain`, `downgrade`, `upgrade`
+- **buffer-health-predictor:** `recommendedFetchCount`, `averageBufferMs`
+- **live-edge-manager:** `segmentsBehind`, `advanceEdge`
+- **stream-telemetry:** `sessionsForStream`, `listSessionIds`
+
 ## Composition
 
 Typical stack:

media-streaming/hyper-p2p-adaptive-streaming-engine/index.js

@@ -58,6 +58,22 @@ class HyperP2PAdaptiveStreamingEngine extends EventEmitter {
     return { ...this._recommendation, bufferMs: this._bufferMs, swarmHealth: this._swarmHealth }
   }
 
+  canSustain (targetBitrate) {
+    return this._recommendation.bitrate >= (targetBitrate | 0)
+  }
+
+  downgrade () {
+    this._swarmHealth = Math.max(0.2, this._swarmHealth - 0.15)
+    this._recompute()
+    return this._recommendation
+  }
+
+  upgrade () {
+    this._swarmHealth = Math.min(1, this._swarmHealth + 0.1)
+    this._recompute()
+    return this._recommendation
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { recommendation: this._recommendation })
   }

media-streaming/hyper-p2p-adaptive-streaming-engine/test/test.js

@@ -9,5 +9,6 @@ test('recommend', async (t) => {
   m.reportBuffer(3000)
   m.reportSwarmHealth(0.9)
   t.ok(m.currentRecommendation().bitrate > 0)
+  t.ok(m.canSustain(1_000_000))
   await m.close()
 })

media-streaming/hyper-p2p-bandwidth-aggregator/docs/api.md

@@ -6,13 +6,34 @@
 
 ## Overview
 
-`HyperP2PBandwidthAggregator` — see [`index.js`](../index.js) for methods, events, and `getStats()`.
+Merges media chunks from multiple peers into **one logical ordered stream** for the viewer. Tracks per-source contribution and gap detection.
+
+## Constructor
+
+```js
+const agg = new HyperP2PBandwidthAggregator({ streamId: 'live-1' })
+```
+
+## Methods
+
+| Method | Description |
+|--------|-------------|
+| `ingest(seq, peerId, data)` | Store chunk if seq not present; returns `false` on duplicate |
+| `ingestBatch(entries)` | Batch ingest `{ seq, peerId, data }[]`; returns count accepted |
+| `hasSeq(seq)` | Whether seq is present |
+| `chunkAt(seq)` | Copy of chunk record or `null` |
+| `missingRanges()` | Gap ranges in sequence space |
+| `logicalView(maxSeq?)` | Concatenated buffer for seq `0..maxSeq` |
+| `sourceStats()` | Per-peer `{ peerId, chunks, bytes }` |
+| `topSource()` | Peer id with most bytes contributed |
+| `coverage()` | `chunks / nextSeq` ratio |
+| `clearBefore(seq)` | Drop chunks with seq < limit |
 
 ## Lifecycle
 
-- `async ready()` — optional Hyperswarm join when `topic` is set
-- `async close()` — teardown
-- `getStats()` — metrics + `protocol: 'bandwidth-aggregator/v1'`
+- `async ready()` — no-op (local merge)
+- `async close()` — clear state
+- `getStats()` — `{ ingested, duplicates, bytes, streamId, chunks, coverage, sources, protocol }`
 
 ## Testing
 

media-streaming/hyper-p2p-bandwidth-aggregator/index.js

@@ -69,6 +69,38 @@ class HyperP2PBandwidthAggregator extends EventEmitter {
     return this._chunks.size / this._nextSeq
   }
 
+  ingestBatch (entries = []) {
+    let n = 0
+    for (const e of entries) {
+      if (e && this.ingest(e.seq, e.peerId, e.data)) n++
+    }
+    return n
+  }
+
+  chunkAt (seq) {
+    const c = this._chunks.get(chunkKey(this.streamId, seq))
+    return c ? { ...c, data: b4a.from(c.data) } : null
+  }
+
+  topSource () {
+    let best = null
+    for (const s of this._sources.values()) {
+      if (!best || s.bytes > best.bytes) best = s
+    }
+    return best ? best.peerId : null
+  }
+
+  clearBefore (seq) {
+    let n = 0
+    for (const [key, c] of this._chunks) {
+      if (c.seq < seq) {
+        this._chunks.delete(key)
+        n++
+      }
+    }
+    return n
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, {
       streamId: this.streamId,

media-streaming/hyper-p2p-bandwidth-aggregator/test/test.js

@@ -10,5 +10,7 @@ test('aggregate chunks', async (t) => {
   t.ok(m.ingest(1, 'p2', Buffer.from('bb')))
   const view = m.logicalView()
   t.is(view.toString(), 'aabb')
+  t.is(m.ingestBatch([{ seq: 2, peerId: 'p3', data: Buffer.from('cc') }]), 1)
+  t.is(m.topSource(), 'p1')
   await m.close()
 })

media-streaming/hyper-p2p-buffer-health-predictor/index.js

@@ -51,6 +51,18 @@ class HyperP2PBufferHealthPredictor extends EventEmitter {
     return [{ type: 'keyframe', count: risk > 0.7 ? 3 : 1 }, { type: 'audio', count: 1 }]
   }
 
+  recommendedFetchCount () {
+    const risk = this.predictStallRisk()
+    if (risk < 0.3) return 4
+    if (risk < 0.6) return 12
+    return 24
+  }
+
+  averageBufferMs () {
+    if (!this._samples.length) return 0
+    return this._samples.reduce((s, x) => s + x.levelMs, 0) / this._samples.length
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, {
       stallRisk: this.predictStallRisk(),

media-streaming/hyper-p2p-buffer-health-predictor/test/test.js

@@ -8,5 +8,6 @@ test('stall risk', async (t) => {
   const m = new HyperP2PBufferHealthPredictor()
   m.reportBuffer(400)
   t.ok(m.predictStallRisk() > 0.5)
+  t.ok(m.recommendedFetchCount() >= 12)
   await m.close()
 })

media-streaming/hyper-p2p-chunk-scheduler-media/index.js

@@ -64,6 +64,26 @@ class HyperP2PChunkSchedulerMedia extends EventEmitter {
 
   pendingCount () { return this._queue.length }
 
+  peek (n = 5) {
+    return this._queue.slice(0, Math.max(0, n | 0)).map((e) => e.chunk)
+  }
+
+  dequeueSeq (seq) {
+    const idx = this._queue.findIndex((e) => e.chunk.seq === seq)
+    if (idx < 0) return null
+    const [entry] = this._queue.splice(idx, 1)
+    this._stats.scheduled++
+    return entry.chunk
+  }
+
+  boostType (type, amount = 20) {
+    for (const e of this._queue) {
+      if (e.chunk.type === type) e.priority += amount
+    }
+    this._sort()
+    return this._queue.length
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { pending: this._queue.length })
   }

media-streaming/hyper-p2p-chunk-scheduler-media/test/test.js

@@ -10,5 +10,7 @@ test('priority schedule', async (t) => {
   m.enqueue({ seq: 0, type: 'keyframe', keyframe: true })
   const next = m.nextChunks(1)
   t.is(next[0].type, 'keyframe')
+  m.enqueue({ seq: 1, type: 'audio' })
+  t.is(m.peek(1)[0].type, 'audio')
   await m.close()
 })

media-streaming/hyper-p2p-content-protection/index.js

@@ -55,6 +55,14 @@ class HyperP2PContentProtection extends EventEmitter {
     return { streamId: sid, keyId: kid }
   }
 
+  listKeyIds () {
+    return [...this._keys.keys()]
+  }
+
+  activeKeyId () {
+    return this._activeKeyId
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { activeKeyId: this._activeKeyId, keys: this._keys.size })
   }

media-streaming/hyper-p2p-content-protection/test/test.js

@@ -9,5 +9,6 @@ test('encrypt decrypt', async (t) => {
   const enc = m.encryptSegment(Buffer.from('secret'))
   const dec = m.decryptSegment(enc.data, enc.keyId)
   t.is(dec.toString(), 'secret')
+  t.ok(m.listKeyIds().length >= 1)
   await m.close()
 })

media-streaming/hyper-p2p-contribution-ledger/index.js

@@ -43,6 +43,21 @@ class HyperP2PContributionLedger extends EventEmitter {
       .map((r) => ({ peerId: r.peerId, bytes: r.bytes }))
   }
 
+  totalUploaded () {
+    let sum = 0
+    for (const r of this._ledger.values()) sum += r.bytes
+    return sum
+  }
+
+  streamsForPeer (peerId) {
+    const cur = this._ledger.get(assertPeerId(peerId))
+    return cur ? [...cur.streams] : []
+  }
+
+  resetPeer (peerId) {
+    return this._ledger.delete(assertPeerId(peerId))
+  }
+
   _gossip (payload) {
     if (this._peerMsgs) {
       gossipSend(this, payload)

media-streaming/hyper-p2p-enterprise-orchestrator/index.js

@@ -66,6 +66,21 @@ class HyperP2PEnterpriseOrchestrator extends EventEmitter {
     return [...this._regions.values()]
   }
 
+  reportViolation (region, detail = {}) {
+    const r = this._regions.get(String(region))
+    if (!r) return false
+    r.lastViolation = { ...detail, at: Date.now() }
+    this._stats.violations++
+    this.emit('violation', { region, detail })
+    return true
+  }
+
+  totalViewers () {
+    let n = 0
+    for (const r of this._regions.values()) n += r.viewers || 0
+    return n
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { regions: this._regions.size })
   }

media-streaming/hyper-p2p-fec-video/index.js

@@ -56,6 +56,22 @@ class HyperP2PFecVideo extends EventEmitter {
     return this.decodeGroup(groupId, { parity: [1] }) != null
   }
 
+  listGroupIds () {
+    return [...this._groups.keys()]
+  }
+
+  redundancyFor (groupId) {
+    const g = this._groups.get(String(groupId))
+    if (!g || !g.dataCount) return 0
+    return g.parityCount / g.dataCount
+  }
+
+  setRedundancy (ratio) {
+    if (ratio < 0 || ratio > 2) throw new Error('redundancy must be 0-2')
+    this.redundancy = ratio
+    return this.redundancy
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { groups: this._groups.size })
   }

media-streaming/hyper-p2p-helper-swarm-coordinator/index.js

@@ -57,6 +57,22 @@ class HyperP2PHelperSwarmCoordinator extends EventEmitter {
     return [...this._helpers.values()].filter((h) => h.usedBps < h.capacityBps)
   }
 
+  unassignViewer (viewerId) {
+    return this._assignments.delete(assertPeerId(viewerId))
+  }
+
+  totalCapacity () {
+    let sum = 0
+    for (const h of this._helpers.values()) sum += h.capacityBps
+    return sum
+  }
+
+  helperUtilization (peerId) {
+    const h = this._helpers.get(assertPeerId(peerId))
+    if (!h || !h.capacityBps) return 0
+    return Math.min(1, h.usedBps / h.capacityBps)
+  }
+
   _gossip (payload) {
     if (this._peerMsgs) gossipSend(this, payload)
   }

media-streaming/hyper-p2p-helper-swarm-coordinator/test/test.js

@@ -9,5 +9,7 @@ test('assign helpers', async (t) => {
   m.registerHelper('h1', 50_000_000)
   const a = m.assignViewer('viewer-1', ['h1'])
   t.is(a.helpers[0], 'h1')
+  t.is(m.totalCapacity(), 50_000_000)
+  t.ok(m.unassignViewer('viewer-1'))
   await m.close()
 })

media-streaming/hyper-p2p-latency-optimizer/index.js

@@ -45,6 +45,18 @@ class HyperP2PLatencyOptimizer extends EventEmitter {
     return candidates[0] || plan.root
   }
 
+  shortestPath () {
+    let best = null
+    for (const p of this._paths.values()) {
+      if (!best || p.rttMs < best.rttMs) best = p
+    }
+    return best
+  }
+
+  listPathIds () {
+    return [...this._paths.keys()]
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { paths: this._paths.size })
   }

media-streaming/hyper-p2p-live-edge-manager/index.js

@@ -49,6 +49,18 @@ class HyperP2PLiveEdgeManager extends EventEmitter {
     return out
   }
 
+  segmentsBehind (streamId, seq) {
+    const e = this.getLiveEdge(streamId)
+    if (!e) return null
+    return Math.max(0, e.seq - (seq | 0))
+  }
+
+  advanceEdge (streamId, delta = 1) {
+    const e = this.getLiveEdge(streamId)
+    if (!e) return null
+    return this.setLiveEdge(streamId, e.seq + (delta | 0), Date.now())
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { streams: this._edges.size })
   }

media-streaming/hyper-p2p-media-chunker/index.js

@@ -62,6 +62,29 @@ class HyperP2PMediaChunker extends EventEmitter {
     return b4a.toString(crypto.hash(c.data), 'hex')
   }
 
+  keyframeIndices (streamId) {
+    return this.listChunks(streamId)
+      .filter((c) => c.keyframe || c.type === CHUNK_TYPES.KEYFRAME)
+      .map((c) => c.seq)
+  }
+
+  totalBytes (streamId) {
+    return this.listChunks(streamId).reduce((s, c) => s + c.byteLength, 0)
+  }
+
+  pruneBefore (streamId, seq) {
+    const sid = assertStreamId(streamId)
+    let n = 0
+    for (const [id, c] of this._chunks) {
+      if (c.streamId === sid && c.seq < seq) {
+        this._chunks.delete(id)
+        n++
+      }
+    }
+    if (n) this.emit('prune', { streamId: sid, before: seq, count: n })
+    return n
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { stored: this._chunks.size })
   }

media-streaming/hyper-p2p-media-chunker/test/test.js

@@ -12,5 +12,7 @@ test('segment and list', async (t) => {
   const chunks = m.segment(Buffer.from('abcdefghij'), { streamId: 's1', keyframe: true })
   t.ok(chunks.length >= 2)
   t.is(m.listChunks('s1').length, chunks.length)
+  t.ok(m.keyframeIndices('s1').length)
+  t.is(m.totalBytes('s1'), 10)
   await m.close()
 })

media-streaming/hyper-p2p-media-tree-orchestrator/docs/api.md

@@ -6,13 +6,26 @@
 
 ## Overview
 
-`HyperP2PMediaTreeOrchestrator` — see [`index.js`](../index.js) for methods, events, and `getStats()`.
+Builds and maintains **crazy-tree** media distribution overlays: parent selection, fanout limits, churn heal, and gossip sync.
 
-## Lifecycle
+## Methods
 
-- `async ready()` — optional Hyperswarm join when `topic` is set
-- `async close()` — teardown
-- `getStats()` — metrics + `protocol: 'media-tree-orchestrator/v1'`
+| Method | Description |
+|--------|-------------|
+| `addNode(peerId, parentId?, role?)` | Join tree; throws if parent fanout exceeded |
+| `selectParent(peerId, candidates)` | Pick parent with most spare fanout |
+| `heal()` | Reattach orphans to root; returns count healed |
+| `treeSnapshot()` | `{ root, maxFanout, nodes[] }` |
+| `optimizeFanout()` | Spill overloaded children to new parents |
+| `removeNode(peerId)` | Detach node |
+| `hasNode(peerId)` | Membership check |
+| `nodeCount()` | Total nodes |
+| `depth(peerId)` | Hops to root |
+| `pathToRoot(peerId)` | Peer id path root → peer |
+
+## P2P
+
+Gossip `tree-join` when `topic` is set via `ready()`.
 
 ## Testing
 

media-streaming/hyper-p2p-media-tree-orchestrator/index.js

@@ -111,6 +111,34 @@ class HyperP2PMediaTreeOrchestrator extends EventEmitter {
     return true
   }
 
+  hasNode (peerId) {
+    return this._nodes.has(assertPeerId(peerId))
+  }
+
+  nodeCount () {
+    return this._nodes.size
+  }
+
+  depth (peerId) {
+    let d = 0
+    let cur = this._nodes.get(assertPeerId(peerId))
+    while (cur && cur.parentId) {
+      d++
+      cur = this._nodes.get(cur.parentId)
+    }
+    return d
+  }
+
+  pathToRoot (peerId) {
+    const path = []
+    let cur = this._nodes.get(assertPeerId(peerId))
+    while (cur) {
+      path.unshift(cur.peerId)
+      cur = cur.parentId ? this._nodes.get(cur.parentId) : null
+    }
+    return path
+  }
+
   _gossip (payload) {
     if (!this._peerMsgs) return
     gossipSend(this, payload)

media-streaming/hyper-p2p-media-tree-orchestrator/test/test.js

@@ -10,7 +10,9 @@ test('tree build heal', async (t) => {
   m.addNode('relay', 'root')
   m.addNode('leaf', 'relay')
   t.is(m.treeSnapshot().nodes.length, 3)
+  t.is(m.depth('leaf'), 2)
   m.removeNode('relay')
   t.ok(m.heal() >= 0)
+  t.alike(m.pathToRoot('leaf'), ['root', 'leaf'])
   await m.close()
 })

media-streaming/hyper-p2p-origin-hybrid-bridge/index.js

@@ -44,6 +44,14 @@ class HyperP2POriginHybridBridge extends EventEmitter {
     return coverage < 0.85
   }
 
+  listOrigins (streamId) {
+    return [...(this._origins.get(assertStreamId(streamId)) || [])]
+  }
+
+  totalBytes () {
+    return this._stats.originBytes + this._stats.p2pBytes
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, {
       offloadRatio: this.p2pOffloadRatio(),

media-streaming/hyper-p2p-peer-selector-streaming/index.js

@@ -63,6 +63,22 @@ class HyperP2PPeerSelectorStreaming extends EventEmitter {
     return out
   }
 
+  updatePeer (peerId, patch = {}) {
+    const p = this._peers.get(assertPeerId(peerId))
+    if (!p) return null
+    Object.assign(p, patch, { updatedAt: Date.now() })
+    return p
+  }
+
+  filterByRegion (region) {
+    const r = String(region)
+    return this.rankPeers().filter((p) => p.region === r)
+  }
+
+  peersAboveScore (minScore = 0.5) {
+    return this.rankPeers().filter((p) => p.score >= minScore)
+  }
+
   removePeer (peerId) {
     return this._peers.delete(assertPeerId(peerId))
   }

media-streaming/hyper-p2p-peer-selector-streaming/test/test.js

@@ -9,5 +9,7 @@ test('pick top peers', async (t) => {
   m.registerPeer('a', { uploadBps: 1_000_000, rttMs: 30, stability: 0.9 })
   m.registerPeer('b', { uploadBps: 10_000_000, rttMs: 20, stability: 0.95 })
   t.is(m.pickTop(1)[0].peerId, 'b')
+  m.updatePeer('a', { uploadBps: 20_000_000 })
+  t.ok(m.peersAboveScore(0.5).length >= 1)
   await m.close()
 })

media-streaming/hyper-p2p-quality-ladder/index.js

@@ -57,6 +57,22 @@ class HyperP2PQualityLadder extends EventEmitter {
     return layers.length ? layers[layers.length - 1] : null
   }
 
+  removeLayer (streamId, repId) {
+    const ladder = this._ladders.get(assertStreamId(streamId))
+    if (!ladder) return false
+    const before = ladder.layers.length
+    ladder.layers = ladder.layers.filter((l) => l.repId !== repId)
+    return ladder.layers.length < before
+  }
+
+  layersAboveBitrate (streamId, minBps) {
+    return this.listLayers(streamId).filter((l) => l.bitrate >= minBps)
+  }
+
+  layersForFps (streamId, minFps) {
+    return this.listLayers(streamId).filter((l) => l.fps >= minFps)
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { ladders: this._ladders.size })
   }

media-streaming/hyper-p2p-retransmission-media/index.js

@@ -48,7 +48,28 @@ class HyperP2PRetransmissionMedia extends EventEmitter {
       const n = this._nacks.get(key)
       if (n) n.priority = 100
     }
-    return this.pendingNacks()
+    return this.pendingNacks().sort((a, b) => (b.priority || 0) - (a.priority || 0))
+  }
+
+  bumpRetry (streamId, seq) {
+    const key = chunkKey(streamId || this.streamId, seq)
+    const n = this._nacks.get(key)
+    if (!n) return false
+    n.retries++
+    n.at = Date.now()
+    return true
+  }
+
+  expireStale (maxAgeMs = 5000) {
+    const now = Date.now()
+    let n = 0
+    for (const [key, entry] of this._nacks) {
+      if (now - entry.at > maxAgeMs) {
+        this._nacks.delete(key)
+        n++
+      }
+    }
+    return n
   }
 
   getStats () {

media-streaming/hyper-p2p-stream-access-control/index.js

@@ -55,6 +55,17 @@ class HyperP2PStreamAccessControl extends EventEmitter {
     return [...this._tokens.values()].filter((c) => c.expiresAt > now)
   }
 
+  extendTtl (token, extraMs = 3600000) {
+    const cap = this._tokens.get(token)
+    if (!cap) return false
+    cap.expiresAt += extraMs
+    return cap.expiresAt
+  }
+
+  hasAccess (streamId, token) {
+    return this.verify(token, streamId)
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { active: this.listActive().length })
   }

media-streaming/hyper-p2p-stream-access-control/test/test.js

@@ -8,5 +8,6 @@ test('grant verify', async (t) => {
   const m = new HyperP2PStreamAccessControl()
   const cap = m.grantCapability('premium-live')
   t.ok(m.verify(cap.token, 'premium-live'))
+  t.ok(m.hasAccess('premium-live', cap.token))
   await m.close()
 })

media-streaming/hyper-p2p-stream-manifest/index.js

@@ -66,6 +66,25 @@ class HyperP2PStreamManifest extends EventEmitter {
 
   listStreamIds () { return [...this._manifests.keys()] }
 
+  removeLayer (streamId, repId) {
+    const m = this._manifests.get(assertStreamId(streamId))
+    if (!m) return false
+    const before = m.layers.length
+    m.layers = m.layers.filter((l) => l.repId !== repId)
+    m.updatedAt = Date.now()
+    return m.layers.length < before
+  }
+
+  peerSources (streamId) {
+    const m = this.getManifest(streamId)
+    return m ? m.sources.filter((s) => s.type === 'peer' || s.peerId) : []
+  }
+
+  layerCount (streamId) {
+    const m = this._manifests.get(assertStreamId(streamId))
+    return m ? m.layers.length : 0
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { streams: this._manifests.size })
   }

media-streaming/hyper-p2p-stream-telemetry/index.js

@@ -63,6 +63,20 @@ class HyperP2PStreamTelemetry extends EventEmitter {
     }
   }
 
+  sessionsForStream (streamId) {
+    const sid = assertStreamId(streamId)
+    const out = []
+    for (const [sessionId, hist] of this._sessions) {
+      const last = hist[hist.length - 1]
+      if (last && last.streamId === sid) out.push(sessionId)
+    }
+    return out
+  }
+
+  listSessionIds () {
+    return [...this._sessions.keys()]
+  }
+
   getStats () {
     return mediaStats(this._stats, PROTOCOL, { sessions: this._sessions.size })
   }