feat(graph): streaming audio ring buffer + dynamic-port gate fixes

[ryanontheinside]

Stacked on #856. See commit message for details.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 83aa0d62-f80f-424b-947b-43053d575577

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch ryanontheinside/acestep

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[Copilot]

Pull request overview

Adds unified backend “node” support alongside pipelines, enabling audio-only/custom-node graphs to stream audio reliably and exposing a node-definition catalog for the frontend to render custom nodes.

Changes:

• Backend: introduce NodeRegistry + NodeProcessor, extend graph schema/executor to run custom nodes and support audio edges/backpressure.
• Backend/RTC: improve audio detection for node-only graphs; adjust audio track buffering; add node-definition discovery endpoint and plugin registration for nodes.
• Frontend: add custom_node rendering/hydration, typed port validation for custom nodes, and improve audio-only playback via a dedicated <audio> element + sink audio handle.

Reviewed changes

Copilot reviewed 32 out of 32 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
src/scope/server/webrtc.py	Detect audio-producing graphs that use custom nodes; add optional inbound param tracing.
src/scope/server/pipeline_processor.py	Add audio edge fan-out; introduce (currently unused) audio-input port bookkeeping.
src/scope/server/mcp_router.py	Allow node-only graphs and set expect_audio based on custom-node outputs.
src/scope/server/graph_schema.py	Add type="node" graph nodes with node_type_id + params; validate structure.
src/scope/server/graph_executor.py	Build/run graphs with NodeProcessor; dynamic-port gating; sink audio edge handling.
src/scope/server/frame_processor.py	Allow node-only graphs to start; add inbound param tracing.
src/scope/server/audio_track.py	Implement lazy audio draining/backpressure; add global playhead lookup helper.
src/scope/server/app.py	Derive pipeline schemas via NodeRegistry; add /api/v1/nodes/definitions.
src/scope/core/plugins/manager.py	Add unified plugin node registration (nodes + pipelines) with compat aliases.
src/scope/core/plugins/hookspecs.py	Add register_nodes hookspec.
src/scope/core/plugins/init.py	Export register_plugin_nodes.
src/scope/core/pipelines/registry.py	Make PipelineRegistry a filtered view over NodeRegistry; init built-in nodes.
src/scope/core/pipelines/interface.py	Make Pipeline a BaseNode; provide get_definition() projection.
src/scope/core/nodes/*	Add node base types, registry, processor, and built-in AudioSourceNode.
frontend/src/pages/StreamPage.tsx	Skip pipeline-load / pipeline_ids for node-only graphs.
frontend/src/lib/graphUtils.ts	Add custom-node serialization + flow conversion support.
frontend/src/lib/api.ts	Add DTOs + fetcher for node definitions; extend GraphNode for type="node".
frontend/src/components/graph/utils/nodeEnrichment.ts	Wire parameter-change handler into custom nodes.
frontend/src/components/graph/utils/connectionValidation.ts	Enforce port-type matching for custom-node stream connections.
frontend/src/components/graph/nodes/SinkNode.tsx	Add audio element for audio-only playback; add stream:audio sink handle.
frontend/src/components/graph/nodes/PipelineNode.tsx	Rename some UI labels (“Pipeline” → “Model”); display name tweak.
frontend/src/components/graph/nodes/CustomNode.tsx	New custom node renderer with typed ports and simple param widgets.
frontend/src/components/graph/hooks/node/useNodeFactories.ts	Add factory support for creating custom_node nodes.
frontend/src/components/graph/hooks/graph/useGraphPersistence.ts	Hydrate custom nodes via /api/v1/nodes/definitions with abort support.
frontend/src/components/graph/contextMenuItems.tsx	Rename “Pipeline” menu item to “Node”.
frontend/src/components/graph/GraphEditor.tsx	Register custom_node type.
frontend/src/components/graph/AddNodeModal.tsx	Populate catalog with plugin custom nodes from /api/v1/nodes/definitions.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

get_current_playhead_seconds() returns the playhead from the first live AudioProcessingTrack in a global WeakSet. The server supports multiple concurrent WebRTC sessions, so this can return a playhead from a different user/session and make any playhead-driven gating nondeterministic under concurrency.

To avoid cross-session leakage, scope playhead lookup to a specific session/FrameProcessor (e.g. store the track on the Session / FrameProcessor and query via that), or key the registry by session/connection id and require callers to pass that key.

Suggested change

	# Registry of live audio tracks so graph nodes that need the playhead (e.g.
	# ACE-Step's StreamVAEDecode skip gate, which mirrors the realtime demo's
	# ``audio_eng.position / SAMPLE_RATE`` read in pipeline.py) can query it
	# without a hard dependency on FrameProcessor. Weak refs so closed tracks
	# drop out automatically.
	_PLAYHEAD_LOCK = threading.Lock()
	_PLAYHEAD_TRACKS: "weakref.WeakSet[AudioProcessingTrack]" = weakref.WeakSet()
	def get_current_playhead_seconds() -> float | None:
	"""Return the playhead position of the first live audio track, in seconds.
	Returns None if no track is registered yet or none are live. Callers
	should treat None as "skip gate disabled this tick".
	"""
	with _PLAYHEAD_LOCK:
	for track in _PLAYHEAD_TRACKS:
	if track.readyState == "live":
	return track.playhead_seconds
	# Registry of live audio tracks. Weak refs ensure closed tracks drop out
	# automatically. Playhead reads must be scoped to a specific track; callers
	# must not query an arbitrary live track from this process-global registry.
	_PLAYHEAD_LOCK = threading.Lock()
	_PLAYHEAD_TRACKS: "weakref.WeakSet[AudioProcessingTrack]" = weakref.WeakSet()
	def get_current_playhead_seconds(
	track: "AudioProcessingTrack | None",
	) -> float | None:
	"""Return the playhead position for the specified live audio track.
	Returns None if ``track`` is None or not live. Callers should pass the
	session-specific ``AudioProcessingTrack`` instance they own and treat
	None as "skip gate disabled this tick".
	"""
	if track is None:
	return None
	with _PLAYHEAD_LOCK:
	if track.readyState == "live" and track in _PLAYHEAD_TRACKS:
	return track.playhead_seconds

[Copilot]

In the sink-wiring loop, encountering an audio edge triggers a break, which stops processing any other stream edges into the same sink. If a sink has both video and audio edges (or if the audio edge appears first in edges_to() order), this will skip creating the dedicated video sink queue and can break WebRTC preview / hardware sinks for that sink.

Instead of breaking out of the edge loop on audio edges, skip only the queue-allocation for that edge (e.g. continue) and still allow other (video) edges to be processed; only break once the sink’s video queue(s) have been successfully attached.

Suggested change

	break
	continue

[Copilot]

audio_input_ports is introduced as a way to mark ports that carry (audio_tensor, sample_rate) tuples, but PipelineProcessor never uses it. As a result, if a graph wires a stream edge carrying audio into a pipeline input port (e.g. to_port == "audio"), process_chunk() will still treat that queue as a video frame queue (qsize >= chunk_size + ensure_video_packet(...)), which will misbehave or crash when it receives audio tuples.

Either implement audio-port handling in PipelineProcessor (exclude audio_input_ports from chunk accumulation and pass audio values through separately), or explicitly disallow audio stream edges into pipeline nodes at validation time until support is complete.

[Copilot]

The numeric input uses value={Number(val)} where val falls back to "" when unset. Number("") becomes 0, so an unset numeric param will render as 0 and can get persisted/sent back as an explicit 0, overriding the backend/defaults unintentionally.

Consider keeping the input controlled with an empty string when the value is unset/invalid (and only coercing to a number on commit), or initialize customNodeParams with numeric defaults so val is never "" for number params.

Suggested change

	value={Number(val)}
	min={(p.ui?.min as number | undefined) ?? undefined}
	max={(p.ui?.max as number | undefined) ?? undefined}
	step={(p.ui?.step as number | undefined) ?? undefined}
	onChange={e => setParam(p.name, Number(e.target.value))}
	value={
	val === "" ||
	val == null ||
	(typeof val === "string" && val.trim() === "") ||
	Number.isNaN(Number(val))
	? ""
	: String(val)
	}
	min={(p.ui?.min as number | undefined) ?? undefined}
	max={(p.ui?.max as number | undefined) ?? undefined}
	step={(p.ui?.step as number | undefined) ?? undefined}
	onChange={e =>
	setParam(
	p.name,
	e.target.value === "" ? "" : e.target.valueAsNumber,
	)
	}