seydd/
main.rs

1// Copyright 2026 Anton Gravestam
2// SPDX-License-Identifier: Apache-2.0
3//! seydd — the Seyd robot daemon. See docs/protocol/seydd.md.
4//!
5//! The daemon is a *host* of [`seyd_core::Agent`] (ADR 0004): the agent owns
6//! discovery, transport, sessions, signaling and certificate maintenance, and
7//! seydd supplies the parts that are specific to running as a daemon — TOML
8//! configuration, RTSP/RTP and UDP inputs, UDP command sinks, and the UDP
9//! publisher-control channel. `seyd-ffi` hosts the same agent differently.
10
11mod config;
12mod enrol;
13mod input;
14mod publisher_control;
15
16use clap::Parser;
17use config::{ChannelKind, Config};
18use seyd_core::channels::{ChannelKind as CoreKind, ChannelSpec};
19use seyd_core::{Agent, AgentConfig, AgentEvent, VideoFrame};
20use std::collections::HashMap;
21use std::path::PathBuf;
22use std::time::Duration;
23
24#[derive(Parser)]
25#[command(version, about)]
26struct Cli {
27    /// Path to seydd.toml
28    #[arg(long, default_value = "/etc/seyd/seydd.toml")]
29    config: PathBuf,
30    /// Open the video inputs, print frame statistics for N seconds, exit.
31    #[arg(long, value_name = "SECONDS")]
32    probe_input: Option<u64>,
33    #[command(subcommand)]
34    command: Option<Command>,
35}
36
37#[derive(clap::Subcommand)]
38enum Command {
39    /// Redeem an enrolment token from the console, registering this robot.
40    Enrol {
41        /// The token shown once when the token was created.
42        #[arg(long)]
43        token: String,
44        /// Override the signal server from the config file.
45        #[arg(long)]
46        signal_url: Option<String>,
47    },
48}
49
50#[tokio::main]
51async fn main() -> anyhow::Result<()> {
52    tracing_subscriber::fmt()
53        .with_env_filter(
54            tracing_subscriber::EnvFilter::from_default_env().add_directive("info".parse()?),
55        )
56        .init();
57    let cli = Cli::parse();
58    let cfg = Config::load(&cli.config)?;
59
60    if let Some(Command::Enrol { token, signal_url }) = &cli.command {
61        let url = signal_url.as_deref().unwrap_or(&cfg.agent.signal_url);
62        return enrol::run(&cfg.agent.robot_id, url, &cfg.agent.credential_path, token).await;
63    }
64
65    tracing::info!(robot_id = %cfg.agent.robot_id, channels = cfg.channels.len(), version = env!("CARGO_PKG_VERSION"), "seydd starting");
66
67    // A robot with no channels connects, announces nothing and looks healthy
68    // in the console — there is no media or telemetry for a pilot to receive.
69    // The usual cause is the section being spelled `[[channels]]`: the key is
70    // `channel`, and serde's `default` turns the typo into an empty list
71    // rather than an error.
72    if cfg.channels.is_empty() {
73        tracing::warn!(
74            config = %cli.config.display(),
75            "no channels configured — this robot will announce nothing and a pilot will \
76             receive no media. Channel sections are spelled [[channel]], not [[channels]]."
77        );
78    }
79
80    if let Some(secs) = cli.probe_input {
81        return probe_input(&cfg, secs).await;
82    }
83    run(cfg).await
84}
85
86/// Redeem an enrolment token on start, so provisioning a robot is one step
87/// rather than two.
88///
89/// This deliberately also runs when a credential already exists. Enrolment
90/// registers the *public* half of whatever key the robot holds, so a robot that
91/// predates enrolment — one that joined a dev server by trust-on-first-use —
92/// keeps its identity and simply becomes known. Requiring a missing credential
93/// stranded exactly those robots: they connected fine and the server answered
94/// `unknown-robot` with nothing to do about it.
95///
96/// With a credential already present the attempt is best-effort: a single-use
97/// token left in the config or environment will fail on the second start, and
98/// that must not stop a robot that is already enrolled from running.
99async fn enrol_if_needed(cfg: &Config) -> anyhow::Result<()> {
100    let Some(token) = cfg.agent.enrolment_token() else {
101        return Ok(());
102    };
103    let first_run = !cfg.agent.credential_path.exists();
104    let result = enrol::run(
105        &cfg.agent.robot_id,
106        &cfg.agent.signal_url,
107        &cfg.agent.credential_path,
108        &token,
109    )
110    .await;
111
112    match result {
113        Ok(()) => Ok(()),
114        // No credential yet means this robot cannot connect at all without
115        // enrolling, so the failure is fatal and worth stopping for.
116        Err(e) if first_run => Err(e),
117        Err(e) => {
118            tracing::warn!(
119                error = %e,
120                "enrolment did not succeed, continuing with the existing credential — \
121                 expected if this robot is already enrolled and the token was spent"
122            );
123            Ok(())
124        }
125    }
126}
127
128async fn run(cfg: Config) -> anyhow::Result<()> {
129    enrol_if_needed(&cfg).await?;
130
131    let specs: Vec<ChannelSpec> = cfg
132        .channels
133        .iter()
134        .enumerate()
135        .map(|(i, c)| ChannelSpec {
136            id: (i + 1) as u8,
137            kind: match c.kind {
138                ChannelKind::Video => CoreKind::Video,
139                ChannelKind::Sensor => CoreKind::Sensor,
140                ChannelKind::Command => CoreKind::Command,
141            },
142            name: c.name.clone(),
143            codec: c.codec.clone(),
144            fps: c.fps,
145            layers: c.layer_specs(),
146            max_bitrate_kbps: c.max_bitrate_kbps,
147        })
148        .collect();
149
150    let (agent, mut events) = Agent::start(AgentConfig {
151        recovery_ladder: cfg.agent.recovery_ladder,
152        robot_id: cfg.agent.robot_id.clone(),
153        signal_url: cfg.agent.signal_url.clone(),
154        credential_path: cfg.agent.credential_path.clone(),
155        quic_port: cfg.agent.quic_port,
156        ipv6: cfg.agent.ipv6,
157        port_mapping: cfg.agent.port_mapping,
158        host_override: cfg.agent.host_override.clone(),
159        qos_profile: cfg.agent.qos_profile.clone(),
160        max_sessions: cfg.agent.max_sessions,
161        channels: specs.clone(),
162        agent_version: format!("seydd/{}", env!("CARGO_PKG_VERSION")),
163        relay: cfg.agent.relay,
164    })
165    .await?;
166
167    // ── inputs ───────────────────────────────────────────────────────────
168    for spec in &specs {
169        let c = &cfg.channels[(spec.id - 1) as usize];
170        match spec.kind {
171            CoreKind::Video => {
172                // One input per simulcast layer (ADR 0008); a single-`input`
173                // channel yields exactly one at layer 0. Every layer's frames
174                // are offered and the engine relays one, so the alternative is
175                // always encoded and one keyframe away.
176                let codec = input::Codec::from_codec_string(&c.codec)?;
177                for (layer, url) in c.inputs() {
178                    let (tx, mut rx) = tokio::sync::mpsc::channel(4);
179                    input::spawn_video(input::rtsp::resolve_credentials(&url), codec, tx)?;
180                    let agent = agent.clone();
181                    let id = spec.id;
182                    tokio::spawn(async move {
183                        while let Some(au) = rx.recv().await {
184                            agent.push_video_layer(
185                                id,
186                                layer,
187                                VideoFrame {
188                                    data: au.data,
189                                    keyframe: au.keyframe,
190                                    capture_ts_us: au.capture_ts_us,
191                                },
192                            );
193                        }
194                    });
195                }
196            }
197            CoreKind::Sensor => {
198                let (tx, mut rx) = tokio::sync::mpsc::channel(64);
199                tokio::spawn(input::udp::run(
200                    input::udp::addr_from_url(c.input.as_deref().unwrap())?,
201                    tx,
202                ));
203                let agent = agent.clone();
204                let id = spec.id;
205                tokio::spawn(async move {
206                    while let Some(m) = rx.recv().await {
207                        agent.push_message(id, &m);
208                    }
209                });
210            }
211            CoreKind::Command => {}
212        }
213    }
214
215    // ── outputs: command UDP sinks and publisher control ─────────────────
216    let out_sock = std::net::UdpSocket::bind("0.0.0.0:0")?;
217    out_sock.set_nonblocking(true)?;
218    let mut command_targets: HashMap<u8, String> = HashMap::new();
219    for spec in specs.iter().filter(|s| s.kind == CoreKind::Command) {
220        let c = &cfg.channels[(spec.id - 1) as usize];
221        command_targets.insert(
222            spec.id,
223            input::udp::addr_from_url(c.output.as_deref().unwrap())?,
224        );
225    }
226    let publisher = publisher_control::PublisherControl::new(
227        cfg.publisher_control.as_ref().map(|p| p.udp.as_str()),
228    );
229    if publisher.enabled() {
230        for spec in specs.iter().filter(|s| s.kind == CoreKind::Video) {
231            publisher.send(&agent.publisher_config(spec.id, "profile"));
232        }
233    }
234
235    // ── event loop ───────────────────────────────────────────────────────
236    // Everything here is daemon policy: where a command goes, what the
237    // publisher is told. The agent has already done the protocol work.
238    loop {
239        tokio::select! {
240            ev = events.recv() => {
241                let Some(ev) = ev else { break };
242                match ev {
243                    AgentEvent::SessionStarted { signal_id, role, .. } => {
244                        publisher.send(&serde_json::json!({"type":"session","state":"started","session_id":signal_id,"role":role.as_str(),"sessions":agent.session_count()}));
245                    }
246                    AgentEvent::SessionEnded { signal_id, .. } => {
247                        publisher.send(&serde_json::json!({"type":"session","state":"ended","session_id":signal_id,"sessions":agent.session_count()}));
248                    }
249                    AgentEvent::Command { channel, payload } => {
250                        if let Some(t) = command_targets.get(&channel) {
251                            let _ = out_sock.send_to(&payload, t);
252                        }
253                    }
254                    AgentEvent::RequestedConfig(v) => { publisher.send(&v); }
255                    AgentEvent::RecoveryRequest { channel, kind, reason } => {
256                        publisher.send(&serde_json::json!({"type":"recovery-request","channel":channel,"kind":kind,"reason":reason}));
257                    }
258                    // The engine switches on the new layer's next keyframe. A
259                    // publisher that can force an IDR on that stream turns "next
260                    // keyframe" from up to a GOP into right now.
261                    AgentEvent::LayerChanged { channel, layer, name, reason } => {
262                        tracing::info!(channel, layer, %name, reason, "layer");
263                        publisher.send(&serde_json::json!({"type":"layer","channel":channel,"layer":layer,"name":name,"reason":reason}));
264                    }
265                    AgentEvent::SignalDenied { .. }
266                    | AgentEvent::SignalConnected
267                    | AgentEvent::SignalDisconnected
268                    | AgentEvent::NatReport(_) => {}
269                }
270            }
271            _ = tokio::signal::ctrl_c() => { tracing::info!("shutting down"); break; }
272        }
273    }
274    agent.stop();
275    Ok(())
276}
277
278async fn probe_input(cfg: &Config, secs: u64) -> anyhow::Result<()> {
279    let (tx, mut rx) = tokio::sync::mpsc::channel(8);
280    for c in cfg.channels.iter().filter(|c| c.kind == ChannelKind::Video) {
281        let url = input::rtsp::resolve_credentials(c.input.as_deref().unwrap());
282        input::spawn_video(url, input::Codec::from_codec_string(&c.codec)?, tx.clone())?;
283    }
284    drop(tx);
285    let deadline = tokio::time::Instant::now() + Duration::from_secs(secs);
286    let (mut frames, mut keyframes, mut bytes, mut loss) = (0u64, 0u64, 0u64, 0u64);
287    let mut last_report = tokio::time::Instant::now();
288    loop {
289        tokio::select! {
290            _ = tokio::time::sleep_until(deadline) => break,
291            au = rx.recv() => {
292                let Some(au) = au else { break };
293                frames += 1;
294                bytes += au.data.len() as u64;
295                loss += au.input_loss as u64;
296                if au.keyframe { keyframes += 1; }
297                if last_report.elapsed().as_secs() >= 1 {
298                    tracing::info!(frames, keyframes, kbps = bytes * 8 / 1000 / last_report.elapsed().as_secs().max(1), input_loss = loss, "probe");
299                    last_report = tokio::time::Instant::now();
300                    bytes = 0;
301                }
302            }
303        }
304    }
305    println!(
306        "probe: {frames} frames, {keyframes} keyframes, {loss} input packets lost over {secs}s"
307    );
308    if frames == 0 {
309        anyhow::bail!("no frames received");
310    }
311    Ok(())
312}
seydd/main.rs

seydd/
main.rs
