Add runtime add/remove for sources/outputs, with web UI menus

Sources and outputs are no longer fixed at startup by the TOML config —
crates/router/src/management.rs adds POST/DELETE /api/manage/sources
and /outputs, and the web UI gets Add source / Add destination forms
plus a remove control on every row/column. Config-loaded and
API-added entries go through the identical code path
(crates/router/src/registry.rs), so a config-declared source is exactly
as manageable as one added later.

Removal is real, not just bookkeeping: srt-io's spawn_input/spawn_output
now return a tokio_util::sync::CancellationToken, threaded through
relay_in/relay_out's select! loops so a removed input/output actually
stops promptly and frees its socket — verified with lsof (add cam3,
confirm the port binds, remove it, confirm the port frees), and again
live in a real browser (found and fixed a real bug along the way: the
add-source form's <input name="id"> collided with the DOM's own
form.id property, a genuine HTML forms quirk — renamed to entry_id).
crosspoint-core gained deregister_source/deregister_output and
has_source/has_output to support this, with a new unit test.

The add-source/add-destination transport dropdown also lists NDI,
Scaler, and Media player as disabled options (per user request) —
honest placeholders pointing at docs/roadmap.md rather than silently
accepting a request that would fail, since only SRT is wired into
management.rs today.

Docs updated throughout (README, architecture.md, roadmap.md) to
describe the runtime API and NDI's status, plus a new diagram
(docs/diagrams/multi-transport-envelope.svg) illustrating how NDI's
envelope encoding lets it share crosspoint-core's plain Bytes channel
with SRT's direct relay. Updated README screenshot shows the new UI.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: allan sargeant

Cargo.lock

@@ -3,9 +3,9 @@
 > **AI-assisted project.** This codebase was created with [Claude](https://claude.com/claude-code)
 > (Anthropic), directed and reviewed by a human author. The relay/crosspoint
 > engine and web UI have been exercised locally — including integration
-> tests that relay real SRT protocol traffic end-to-end and a live
+> tests that relay real SRT and NDI protocol traffic end-to-end and a live
 > crosspoint switch via the web UI, see [Status](#status) — but **not yet
-> run against real-world third-party SRT encoders/decoders or over an
+> run against real-world third-party SRT/NDI encoders/decoders or over an
 > actual (non-loopback) network path**. Review before relying on it for
 > anything live.
 
@@ -17,11 +17,11 @@ SDI/HDMI.
 
 ![srt-router architecture: SRT inputs feeding crosspoint-core's per-source broadcast channels, a routing table mapping each output to one source, out to SRT outputs](docs/diagrams/architecture.svg)
 
-<img src="docs/screenshots/crosspoint-grid.png" alt="The crosspoint web UI: a grid of outputs (program, preview) by sources (remote-feed, cam1, cam2), with the currently-routed cell highlighted green in each row" width="480">
+<img src="docs/screenshots/crosspoint-grid.png" alt="The crosspoint web UI: a grid of outputs (program, preview) by sources (cam1, cam2, remote-feed), each labeled with its transport kind and a remove control, plus Add source / Add destination buttons below" width="560">
 
 *The web UI above is a real screenshot of the router running locally against
 [config/example.toml](config/example.toml), not a mockup — captured while
-verifying the routing/persistence/websocket behavior described in
+verifying the routing/persistence/websocket/add-remove behavior described in
 [Status](#status) below.*
 
 ## What it does
@@ -39,14 +39,24 @@ crosspoint as a "source" without the engine caring that they're not relayed
 SRT. **Not built yet** — Phase 1 is relay-only; see
 [docs/roadmap.md](docs/roadmap.md).
 
-Control is a local web UI (a crosspoint grid: click a cell to route that
-output from that source) backed by a small REST API. No auth/TLS — this is
-meant to run on a trusted operations network, the same trust model as a
-hardware router's control port.
+Control is a local web UI backed by a small REST API: a crosspoint grid
+(click a cell to route that output from that source), plus **Add
+source**/**Add destination** menus and a remove control on every row/column
+for adding or tearing down SRT inputs/outputs at runtime — not just what was
+in the TOML config at startup. No auth/TLS — this is meant to run on a
+trusted operations network, the same trust model as a hardware router's
+control port.
+
+**Transports beyond SRT:** `crates/ndi-io` is a real, tested NDI transport
+(see [Status](#status)) but isn't wired into the router binary/web UI yet —
+the add-source/add-destination menus show NDI as a disabled option for now.
+`crates/omt-io` is a placeholder for the equivalent OMT transport (open,
+MIT-licensed) — not yet implemented, see [docs/roadmap.md](docs/roadmap.md).
 
 ## Status
 
-**Phase 1 (current): relay-only crosspoint + web UI.** Working:
+**Phase 1 (current): relay-only crosspoint + web UI, dynamic add/remove.**
+Working:
 
 - SRT input/output as either `listener` (this router waits for a
   connection) or `caller` (this router dials out), each reconnecting on its
@@ -56,27 +66,46 @@ hardware router's control port.
 - A local web UI (`crates/web`) — grid of outputs x sources, click to route,
   updated live over a websocket (`GET /ws`) with a REST poll (`GET
   /api/state`) as first paint / fallback.
+- **Runtime add/remove**: `POST`/`DELETE /api/manage/sources` and
+  `/api/manage/outputs` (`crates/router/src/management.rs`) spawn or tear
+  down an SRT input/output on the fly — the same code path the static TOML
+  config uses at startup, so a config-declared source is exactly as
+  removable as one added later. Backed by
+  [`tokio_util::sync::CancellationToken`](https://docs.rs/tokio-util) per
+  task (added to `srt-io`) so removal actually stops the task and frees the
+  socket, not just forgets about it. The web UI exposes this as **Add
+  source**/**Add destination** forms plus a remove control per row/column.
 - Routing changes optionally persist to disk (`[state]` in the config) and
   reload on restart, overriding each output's `default_source`.
+- `crates/ndi-io`: a real NDI transport using
+  [grafton-ndi](https://github.com/GrantSparks/grafton-ndi) (Apache-2.0)
+  against the actual NDI SDK — not wired into the router binary yet (see
+  [What it does](#what-it-does)), but its own integration test drives a real
+  NDI sender and receiver against it, consistently passing.
 - CI (GitHub Actions) runs `fmt --check`, `clippy -D warnings`, and the full
-  test suite on every push/PR.
+  test suite on every push/PR — SRT-only (`ndi-io`/`omt-io` need real
+  SDKs CI can't install, so they're real workspace members but excluded
+  from `default-members`, see [docs/architecture.md](docs/architecture.md)).
 - Verified locally, not just compiled: `cargo test` passes, including
-  integration tests (`crates/srt-io/tests/relay.rs`) that relay real SRT
-  protocol traffic end-to-end through the crosspoint using `srt-tokio`
-  clients as the encoder/decoder — one test also exercises a **live
-  re-route mid-stream over an already-established SRT connection**.
-  Separately confirmed by hand: running the binary against
-  `config/example.toml` binds real UDP/SRT listener sockets (via `lsof`),
-  the REST API and a real browser click both drive live crosspoint changes,
-  the websocket push updates the grid with no client-side polling, and a
-  persisted route survives a real process restart.
-
-**Not yet done:** no test against a real third-party SRT
-encoder/decoder or over a real (non-loopback) network path — only local
-testing so far, still the main open gap. Also missing: special-purpose
-sources (stills/media player/scaler), auth on the web UI/API, external
-control API/Companion integration. See [docs/roadmap.md](docs/roadmap.md)
-for the full phased plan.
+  integration tests (`crates/srt-io/tests/relay.rs`,
+  `crates/ndi-io/tests/relay.rs`) that relay real protocol traffic
+  end-to-end through the crosspoint — one SRT test also exercises a **live
+  re-route mid-stream over an already-established connection**. Separately
+  confirmed by hand: running the binary against `config/example.toml` binds
+  real UDP/SRT listener sockets (via `lsof`), adding a source through the
+  web UI binds a new one live and removing it frees the port (also via
+  `lsof`), the REST API and a real browser click both drive live crosspoint
+  changes, the websocket push updates the grid with no client-side polling,
+  and a persisted route survives a real process restart.
+
+**Not yet done:** no test against a real third-party SRT/NDI encoder or
+decoder, or over a real (non-loopback) network path — only local testing so
+far, still the main open gap. Also missing: NDI wired into the router's
+config/web UI, OMT (placeholder crate only), special-purpose sources
+(stills/media player/scaler — the add-source menu shows them as disabled
+options), auth on the web UI/API, external control API/Companion
+integration. See [docs/roadmap.md](docs/roadmap.md) for the full phased
+plan.
 
 ## Quick start
 

README.md

@@ -8,3 +8,6 @@ license.workspace = true
 bytes = "1"
 tokio = { version = "1", features = ["sync"] }
 parking_lot = "0.12"
+
+[dev-dependencies]
+tokio = { version = "1", features = ["sync", "macros", "rt"] }

crates/core/Cargo.toml

@@ -62,6 +62,29 @@ impl Crosspoint {
         self.sources.read().get(source).map(|tx| tx.subscribe())
     }
 
+    /// Remove a source. Any output currently routed to it keeps its `watch`
+    /// value pointing at the now-gone id (routing is a separate concern,
+    /// see [`route`](Self::route)) but its `subscribe` will start returning
+    /// `None`/closed — the caller (a `*-io` crate's output task) is expected
+    /// to treat that the same as "nothing routed yet" until re-routed.
+    /// Returns `true` if a source with this id existed.
+    pub fn deregister_source(&self, id: &str) -> bool {
+        self.sources.write().remove(id).is_some()
+    }
+
+    /// Remove an output. Returns `true` if an output with this id existed.
+    pub fn deregister_output(&self, id: &str) -> bool {
+        self.outputs.write().remove(id).is_some()
+    }
+
+    pub fn has_source(&self, id: &str) -> bool {
+        self.sources.read().contains_key(id)
+    }
+
+    pub fn has_output(&self, id: &str) -> bool {
+        self.outputs.read().contains_key(id)
+    }
+
     /// Point an output at a different source. No-ops (returns `false`) if
     /// either id is unknown, so callers (e.g. the web API) can distinguish
     /// a bad request from a successful re-route.
@@ -126,4 +149,23 @@ mod tests {
         assert!(!xp.route("out1", "nonexistent"));
         assert!(!xp.route("nonexistent", "a"));
     }
+
+    #[test]
+    fn deregister_removes_and_reports_prior_existence() {
+        let xp = Crosspoint::new();
+        xp.register_source("a");
+        xp.register_output("out1", "a");
+
+        assert!(xp.has_source("a"));
+        assert!(xp.has_output("out1"));
+
+        assert!(xp.deregister_source("a"));
+        assert!(!xp.has_source("a"));
+        assert!(!xp.deregister_source("a"), "second removal reports false");
+
+        assert!(xp.deregister_output("out1"));
+        assert!(!xp.has_output("out1"));
+
+        assert!(xp.subscribe("a").is_none());
+    }
 }

crates/core/src/lib.rs

@@ -12,7 +12,10 @@ path = "src/main.rs"
 crosspoint-core = { path = "../core" }
 srt-io = { path = "../srt-io" }
 crosspoint-web = { path = "../web" }
+axum = "0.7"
 tokio = { version = "1", features = ["full"] }
+tokio-util = "0.7"
+parking_lot = "0.12"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 toml = "0.8"

crates/router/Cargo.toml

@@ -1,4 +1,6 @@
 mod config;
+mod management;
+mod registry;
 mod state;
 
 use std::collections::HashMap;
@@ -8,6 +10,8 @@ use anyhow::{Context, Result};
 use clap::Parser;
 use config::Config;
 use crosspoint_core::Crosspoint;
+use management::ManageState;
+use registry::Registry;
 
 /// Crosspoint-based SRT router.
 #[derive(Parser)]
@@ -31,10 +35,17 @@ async fn main() -> Result<()> {
         .with_context(|| format!("parsing config file {}", args.config.display()))?;
 
     let crosspoint = Crosspoint::new();
+    let registry = Registry::new();
 
+    // Config-defined and management-API-added sources/outputs are
+    // deliberately the same thing once running — both end up in `registry`
+    // via the same spawn_input/spawn_output + insert_* calls, so a
+    // config-declared source is exactly as removable via the API (or
+    // listable in the web UI's add/remove menus) as one added later.
     for input in config.inputs {
         tracing::info!(id = %input.id, "starting SRT input");
-        srt_io::spawn_input(input.id, input.endpoint, crosspoint.clone());
+        let cancel = srt_io::spawn_input(input.id.clone(), input.endpoint, crosspoint.clone());
+        registry.insert_source(input.id, "srt", cancel);
     }
 
     let persisted_routes: HashMap<String, String> = match &config.state {
@@ -58,12 +69,13 @@ async fn main() -> Result<()> {
             .cloned()
             .unwrap_or(output.default_source);
         tracing::info!(id = %output.id, source = %initial_source, "starting SRT output");
-        srt_io::spawn_output(
-            output.id,
+        let cancel = srt_io::spawn_output(
+            output.id.clone(),
             output.endpoint,
             initial_source,
             crosspoint.clone(),
         );
+        registry.insert_output(output.id, "srt", cancel);
     }
 
     if let Some(state_cfg) = config.state {
@@ -76,6 +88,13 @@ async fn main() -> Result<()> {
         .parse()
         .with_context(|| format!("invalid web.bind address {:?}", config.web.bind))?;
 
-    crosspoint_web::serve(bind, crosspoint).await?;
+    let manage_state = ManageState {
+        crosspoint: crosspoint.clone(),
+        registry,
+    };
+    let app = crosspoint_web::app(crosspoint).merge(management::router(manage_state));
+    let listener = tokio::net::TcpListener::bind(bind).await?;
+    tracing::info!(%bind, "crosspoint web UI listening");
+    axum::serve(listener, app).await?;
     Ok(())
 }