Skip to main content

scv_server/
components.rs

1//! Server-owned lifecycle for every long-running integration.
2
3use anyhow::{Result, bail};
4use async_trait::async_trait;
5use scv_channels::state::{self, AccountSettings};
6use scv_protocol::{ComponentHealth, ComponentState, DaemonCommand, DaemonStatus, RemoteTools};
7use std::{
8    collections::BTreeMap,
9    path::PathBuf,
10    sync::{Arc, Mutex},
11    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
12};
13use tokio::task::JoinHandle;
14use tokio_util::sync::CancellationToken;
15
16const STOP_GRACE: Duration = Duration::from_secs(5);
17/// How long an operator command waits out a bridge's state commit.
18const BUSY_RETRY: Duration = Duration::from_secs(5);
19
20/// Components must observe cancellation and must not detach child tasks.
21/// Return on failure; the supervisor owns retries and bounded shutdown.
22#[async_trait]
23pub trait Component: Send + Sync + 'static {
24    async fn run(&self, cancellation: CancellationToken, health: HealthReporter) -> Result<()>;
25}
26
27#[derive(Clone)]
28pub struct HealthReporter(Arc<Mutex<ComponentHealth>>);
29
30impl HealthReporter {
31    pub fn contact(&self, connected: bool) {
32        let mut health = self.0.lock().unwrap();
33        if matches!(
34            health.state,
35            ComponentState::Stopping | ComponentState::Stopped
36        ) {
37            return;
38        }
39        health.state = if connected {
40            ComponentState::Connected
41        } else {
42            ComponentState::Disconnected
43        };
44        health.error = (!connected).then(|| "Component contact failed".into());
45        if connected {
46            health.last_success_unix_seconds = Some(
47                SystemTime::now()
48                    .duration_since(UNIX_EPOCH)
49                    .unwrap_or_default()
50                    .as_secs(),
51            );
52        }
53    }
54
55    fn transition(&self, state: ComponentState, error: Option<&str>) {
56        let mut health = self.0.lock().unwrap();
57        health.state = state;
58        health.error = error.map(str::to_owned);
59    }
60
61    fn snapshot(&self) -> ComponentHealth {
62        self.0.lock().unwrap().clone()
63    }
64}
65
66pub struct Supervisor {
67    tasks: BTreeMap<String, RunningComponent>,
68    grace: Duration,
69    initial_backoff: Duration,
70}
71
72struct RunningComponent {
73    cancellation: CancellationToken,
74    task: JoinHandle<()>,
75    health: HealthReporter,
76}
77
78impl Default for Supervisor {
79    fn default() -> Self {
80        Self {
81            tasks: BTreeMap::new(),
82            grace: STOP_GRACE,
83            initial_backoff: Duration::from_secs(1),
84        }
85    }
86}
87
88impl Supervisor {
89    /// Idempotent start: replacement must first stop and join the old instance.
90    pub fn start(&mut self, component: Arc<dyn Component>, health: ComponentHealth) {
91        if self.tasks.contains_key(&health.id) {
92            return;
93        }
94        let id = health.id.clone();
95        let health = HealthReporter(Arc::new(Mutex::new(health)));
96        let cancellation = CancellationToken::new();
97        let cancel = cancellation.clone();
98        let report = health.clone();
99        let initial_backoff = self.initial_backoff;
100        let grace = self.grace;
101        let task = tokio::spawn(async move {
102            let mut delay = initial_backoff;
103            loop {
104                if cancel.is_cancelled() {
105                    break;
106                }
107                report.transition(ComponentState::Starting, None);
108                let started = Instant::now();
109                // Catch task panics without letting them take down the daemon or skip retries.
110                let instance = component.clone();
111                let child_cancel = cancel.clone();
112                let child_report = report.clone();
113                let mut child =
114                    tokio::spawn(async move { instance.run(child_cancel, child_report).await });
115                tokio::select! {
116                    biased;
117                    _ = cancel.cancelled() => {
118                        report.transition(ComponentState::Stopping, None);
119                        if tokio::time::timeout(grace, &mut child).await.is_err() {
120                            child.abort();
121                            let _ = child.await;
122                        }
123                        break;
124                    }
125                    _ = &mut child => {}
126                }
127                report.transition(
128                    ComponentState::Backoff,
129                    Some("Component stopped unexpectedly; retrying"),
130                );
131                report.0.lock().unwrap().restarts += 1;
132                if started.elapsed() >= Duration::from_secs(60) {
133                    delay = initial_backoff;
134                }
135                tokio::select! {
136                    _ = cancel.cancelled() => break,
137                    _ = tokio::time::sleep(delay) => {}
138                }
139                delay = (delay * 2).min(Duration::from_secs(60));
140            }
141            report.transition(ComponentState::Stopped, None);
142        });
143        self.tasks.insert(
144            id,
145            RunningComponent {
146                cancellation,
147                task,
148                health,
149            },
150        );
151    }
152
153    pub fn health(&self) -> Vec<ComponentHealth> {
154        self.tasks
155            .values()
156            .map(|task| task.health.snapshot())
157            .collect()
158    }
159
160    pub async fn stop(&mut self, id: &str) {
161        if let Some(running) = self.tasks.get_mut(id) {
162            running.cancellation.cancel();
163            // The runner owns abort/join of its child, so never abort the runner first.
164            let _ = (&mut running.task).await;
165        }
166        self.tasks.remove(id);
167    }
168
169    pub async fn shutdown(&mut self) {
170        for task in self.tasks.values() {
171            task.cancellation.cancel();
172        }
173        for id in self.tasks.keys().cloned().collect::<Vec<_>>() {
174            self.stop(&id).await;
175        }
176    }
177}
178
179/// A chat channel whose accounts the daemon supervises.
180#[derive(Clone, Copy, Debug, PartialEq, Eq)]
181enum Channel {
182    Wechat,
183    Feishu,
184}
185
186const CHANNELS: [Channel; 2] = [Channel::Wechat, Channel::Feishu];
187
188/// A channel account's saved credentials.
189#[derive(Clone, PartialEq)]
190enum Credentials {
191    Wechat(scv_clawbot::state::Account),
192    Feishu(scv_feishu::state::Account),
193}
194
195impl Channel {
196    fn parse(name: &str) -> Result<Self> {
197        CHANNELS
198            .into_iter()
199            .find(|channel| channel.name() == name)
200            .ok_or_else(|| anyhow::anyhow!("Unknown channel {name:?}"))
201    }
202
203    fn name(self) -> &'static str {
204        match self {
205            Self::Wechat => scv_clawbot::CHANNEL,
206            Self::Feishu => scv_feishu::CHANNEL,
207        }
208    }
209
210    fn title(self) -> &'static str {
211        match self {
212            Self::Wechat => "WeChat",
213            Self::Feishu => "Feishu",
214        }
215    }
216
217    /// Account discovery. WeChat state saved before channels moves first;
218    /// the user resolves a failed move by hand, and `scv channels status`
219    /// names the files.
220    fn account_names(self) -> std::result::Result<Vec<String>, &'static str> {
221        const DISCOVERY: &str =
222            "Account discovery failed; components stopped until configuration is readable";
223        match self {
224            Self::Wechat => scv_clawbot::state::migrate()
225                .map_err(|_| {
226                    "Saved WeChat state could not move to channels/wechat; run `scv channels status` for details"
227                })
228                .and_then(|_| scv_clawbot::state::account_names().map_err(|_| DISCOVERY)),
229            Self::Feishu => scv_feishu::state::account_names().map_err(|_| DISCOVERY),
230        }
231    }
232
233    fn snapshot(self, account: &str) -> Result<(Option<Credentials>, AccountSettings)> {
234        Ok(match self {
235            Self::Wechat => {
236                let (credentials, settings) = scv_clawbot::state::account_snapshot(account)?;
237                (credentials.map(Credentials::Wechat), settings)
238            }
239            Self::Feishu => {
240                let (credentials, settings) = scv_feishu::state::account_snapshot(account)?;
241                (credentials.map(Credentials::Feishu), settings)
242            }
243        })
244    }
245
246    fn signed_in(self, account: &str) -> Result<bool> {
247        Ok(match self {
248            Self::Wechat => scv_clawbot::state::account(account)?.is_some(),
249            Self::Feishu => scv_feishu::state::account(account)?.is_some(),
250        })
251    }
252
253    fn settings(self, account: &str) -> Result<AccountSettings> {
254        match self {
255            Self::Wechat => scv_clawbot::state::settings(account),
256            Self::Feishu => scv_feishu::state::settings(account),
257        }
258    }
259
260    fn save_settings(self, account: &str, settings: &AccountSettings) -> Result<()> {
261        match self {
262            Self::Wechat => scv_clawbot::state::save_settings(account, settings),
263            Self::Feishu => scv_feishu::state::save_settings(account, settings),
264        }
265    }
266
267    fn remove(self, account: &str) -> Result<()> {
268        match self {
269            Self::Wechat => scv_clawbot::state::remove(account),
270            Self::Feishu => scv_feishu::state::remove(account),
271        }
272    }
273}
274
275impl Credentials {
276    /// The authenticated owner, the only sender remote tools may reach.
277    fn owner(&self) -> Option<&str> {
278        match self {
279            Self::Wechat(account) => account.user_id.as_deref(),
280            Self::Feishu(account) => account.owner_open_id.as_deref(),
281        }
282    }
283
284    /// The bot's identity shown in status: the iLink bot or the Feishu app.
285    fn bot_id(&self) -> Option<String> {
286        match self {
287            Self::Wechat(account) => account.bot_id.clone(),
288            Self::Feishu(account) => Some(account.app_id.clone()),
289        }
290    }
291}
292
293struct ChannelAccount {
294    channel: Channel,
295    account: String,
296    credentials: Credentials,
297    workspace: PathBuf,
298    socket: PathBuf,
299    tool_owner: Option<String>,
300}
301
302#[async_trait]
303impl Component for ChannelAccount {
304    async fn run(&self, cancellation: CancellationToken, health: HealthReporter) -> Result<()> {
305        let tool_owner = self.tool_owner.clone().map(|user_id| {
306            let turn_timeout = scv_channels::owner_turn_timeout(max_tool_timeout(&self.workspace));
307            tracing::info!(
308                "{} {} owner turns may run up to {} seconds",
309                self.channel.title(),
310                self.account,
311                turn_timeout.as_secs()
312            );
313            scv_channels::ToolOwner {
314                user_id,
315                turn_timeout,
316            }
317        });
318        let report = Arc::new(move |connected| health.contact(connected));
319        match &self.credentials {
320            Credentials::Wechat(credentials) => {
321                scv_clawbot::run_supervised(
322                    &credentials.token,
323                    &credentials.base_url,
324                    &self.account,
325                    &self.workspace,
326                    &self.socket,
327                    tool_owner.as_ref(),
328                    cancellation,
329                    report,
330                )
331                .await
332            }
333            Credentials::Feishu(credentials) => {
334                scv_feishu::run_supervised(
335                    credentials,
336                    &self.account,
337                    &self.workspace,
338                    &self.socket,
339                    tool_owner.as_ref(),
340                    cancellation,
341                    report,
342                )
343                .await
344            }
345        }
346    }
347}
348
349pub(crate) struct Components {
350    supervisor: Supervisor,
351    /// Running or disabled accounts by component ID, `<channel>:<account>`.
352    desired: BTreeMap<String, (Credentials, AccountSettings)>,
353    inactive: BTreeMap<String, ComponentHealth>,
354    socket: PathBuf,
355    workspace: PathBuf,
356}
357
358impl Components {
359    pub fn new(socket: PathBuf, workspace: PathBuf) -> Self {
360        Self {
361            supervisor: Supervisor::default(),
362            desired: BTreeMap::new(),
363            inactive: BTreeMap::new(),
364            socket,
365            workspace,
366        }
367    }
368
369    pub fn status(&self) -> DaemonStatus {
370        let mut components = self.supervisor.health();
371        components.extend(self.inactive.values().cloned());
372        components.sort_by(|a, b| a.id.cmp(&b.id));
373        DaemonStatus {
374            version: env!("CARGO_PKG_VERSION").into(),
375            pid: std::process::id(),
376            components,
377            delegations: Default::default(),
378        }
379    }
380
381    /// Match running components to every channel's saved accounts. A channel
382    /// whose accounts cannot be discovered stops its own components and
383    /// reports why; other channels keep running.
384    pub async fn reconcile(&mut self) -> Result<()> {
385        let mut failed = false;
386        for channel in CHANNELS {
387            match channel.account_names() {
388                Ok(names) => self.reconcile_channel(channel, names).await,
389                Err(message) => {
390                    failed = true;
391                    for id in self.ids(channel) {
392                        self.supervisor.stop(&id).await;
393                        self.desired.remove(&id);
394                        self.inactive.remove(&id);
395                    }
396                    let mut health = initial_health(channel, "discovery", None, false);
397                    health.id = component_id(channel, "discovery-error");
398                    health.state = ComponentState::Failed;
399                    health.error = Some(message.into());
400                    self.inactive.insert(health.id.clone(), health);
401                }
402            }
403        }
404        if failed {
405            bail!("Account discovery failed");
406        }
407        Ok(())
408    }
409
410    /// Component IDs of one channel, running or not.
411    fn ids(&self, channel: Channel) -> Vec<String> {
412        let prefix = format!("{}:", channel.name());
413        self.desired
414            .keys()
415            .chain(self.inactive.keys())
416            .filter(|id| id.starts_with(&prefix))
417            .cloned()
418            .collect()
419    }
420
421    async fn reconcile_channel(&mut self, channel: Channel, names: Vec<String>) {
422        let wanted: Vec<String> = names
423            .iter()
424            .map(|name| component_id(channel, name))
425            .collect();
426        for id in self.ids(channel) {
427            if !wanted.contains(&id) {
428                self.supervisor.stop(&id).await;
429                self.desired.remove(&id);
430                self.inactive.remove(&id);
431            }
432        }
433        for name in names {
434            let id = component_id(channel, &name);
435            let loaded = (|| -> Result<_> {
436                let (account, settings) = channel.snapshot(&name)?;
437                Ok((
438                    account.ok_or_else(|| anyhow::anyhow!("missing account"))?,
439                    settings,
440                ))
441            })();
442            let (credentials, settings) = match loaded {
443                Ok(value) => value,
444                Err(error) => {
445                    self.account_error(channel, &name, error).await;
446                    continue;
447                }
448            };
449            if self.desired.get(&id) == Some(&(credentials.clone(), settings.clone())) {
450                continue;
451            }
452            self.supervisor.stop(&id).await;
453            self.inactive.remove(&id);
454            let mut health = initial_health(channel, &name, Some(&credentials), settings.enabled);
455            let tool_owner = tool_owner(&credentials, &settings);
456            if tool_owner.is_some() {
457                health.remote_tools = RemoteTools::Owner;
458            }
459            if settings.enabled {
460                let workspace = settings
461                    .workspace
462                    .clone()
463                    .unwrap_or_else(|| self.workspace.clone());
464                if !workspace.is_absolute() || !workspace.is_dir() {
465                    health.state = ComponentState::Failed;
466                    health.error =
467                        Some("Component workspace must be an existing absolute directory".into());
468                    self.inactive.insert(id.clone(), health);
469                    self.desired.remove(&id);
470                    continue;
471                }
472                self.supervisor.start(
473                    Arc::new(ChannelAccount {
474                        channel,
475                        account: name.clone(),
476                        credentials: credentials.clone(),
477                        workspace,
478                        socket: self.socket.clone(),
479                        tool_owner,
480                    }),
481                    health,
482                );
483            } else {
484                health.state = ComponentState::Disabled;
485                self.inactive.insert(id.clone(), health);
486            }
487            self.desired.insert(id, (credentials, settings));
488        }
489    }
490
491    async fn account_error(&mut self, channel: Channel, name: &str, error: anyhow::Error) {
492        // A bridge state commit briefly holds this same lock. Retry next refresh
493        // rather than interrupting healthy work for ordinary lock contention.
494        if is_busy(&error) {
495            return;
496        }
497        let id = component_id(channel, name);
498        self.supervisor.stop(&id).await;
499        self.desired.remove(&id);
500        let mut health = initial_health(channel, name, None, true);
501        health.state = ComponentState::Failed;
502        health.error = Some("Invalid or inaccessible account/settings".into());
503        self.inactive.insert(id, health);
504    }
505
506    pub async fn control(&mut self, command: DaemonCommand) -> Result<DaemonStatus> {
507        match command {
508            // Delegations belong to the connection handler, which adds them.
509            DaemonCommand::Status
510            | DaemonCommand::Delegations { .. }
511            | DaemonCommand::DelegationKill { .. } => return Ok(self.status()),
512            DaemonCommand::Reload => {}
513            DaemonCommand::ChannelSet {
514                channel,
515                account,
516                enabled,
517                workspace,
518                remote_tools,
519            } => {
520                let channel = Channel::parse(&channel)?;
521                state::validate_name(&account)?;
522                let workspace = match workspace {
523                    Some(path) => {
524                        let path = PathBuf::from(path);
525                        if !path.is_absolute() || !path.is_dir() {
526                            bail!("Invalid component workspace");
527                        }
528                        Some(std::fs::canonicalize(path)?)
529                    }
530                    None => None,
531                };
532                retry_while_busy(|| {
533                    if !channel.signed_in(&account)? {
534                        bail!("Account is not logged in");
535                    }
536                    let mut settings = channel.settings(&account)?;
537                    settings.enabled = enabled;
538                    if let Some(path) = &workspace {
539                        settings.workspace = Some(path.clone());
540                    }
541                    if let Some(mode) = remote_tools {
542                        settings.remote_tools = mode;
543                    }
544                    channel.save_settings(&account, &settings)
545                })
546                .await?;
547            }
548            DaemonCommand::ChannelLogout { channel, account } => {
549                let channel = Channel::parse(&channel)?;
550                state::validate_name(&account)?;
551                // Persist disabled and tool-free first, so failed deletion can
552                // neither resurrect a live account nor hand a later login the grant.
553                retry_while_busy(|| {
554                    let mut settings = channel.settings(&account)?;
555                    settings.enabled = false;
556                    settings.remote_tools = RemoteTools::None;
557                    channel.save_settings(&account, &settings)
558                })
559                .await?;
560                let id = component_id(channel, &account);
561                self.supervisor.stop(&id).await;
562                self.desired.remove(&id);
563                self.inactive.remove(&id);
564                retry_while_busy(|| channel.remove(&account)).await?;
565            }
566        }
567        self.reconcile().await?;
568        Ok(self.status())
569    }
570
571    pub async fn shutdown(&mut self) {
572        self.supervisor.shutdown().await;
573    }
574}
575
576/// The longest tool call an owner session in `workspace` may make, from the
577/// configuration its sessions load. Read at each (re)start of the component.
578fn max_tool_timeout(workspace: &std::path::Path) -> std::time::Duration {
579    let seconds = crate::Config::load(workspace, crate::ConfigOverrides::default())
580        .map(|config| config.tools.max_timeout_seconds)
581        .unwrap_or_else(|error| {
582            tracing::warn!("Channel owner turns use the default tool timeout ceiling: {error:#}");
583            crate::config::ToolConfig::default().max_timeout_seconds
584        });
585    std::time::Duration::from_secs(seconds)
586}
587
588/// Account transactions fail fast while their lock is held, and a running
589/// bridge holds it for every state commit. Operator commands retry through
590/// that contention instead of failing whenever they coincide with a commit.
591async fn retry_while_busy<T>(mut operation: impl FnMut() -> Result<T>) -> Result<T> {
592    let deadline = tokio::time::Instant::now() + BUSY_RETRY;
593    loop {
594        match operation() {
595            Err(error) if is_busy(&error) && tokio::time::Instant::now() < deadline => {
596                tokio::time::sleep(Duration::from_millis(10)).await;
597            }
598            result => return result,
599        }
600    }
601}
602
603fn is_busy(error: &anyhow::Error) -> bool {
604    error
605        .downcast_ref::<std::io::Error>()
606        .is_some_and(|error| error.kind() == std::io::ErrorKind::WouldBlock)
607}
608
609/// Only the authenticated account owner may receive tools. Credentials without
610/// a known owner ID grant tools to nobody, even when the setting asks for it.
611fn tool_owner(credentials: &Credentials, settings: &AccountSettings) -> Option<String> {
612    (settings.remote_tools == RemoteTools::Owner)
613        .then(|| credentials.owner().map(str::to_owned))
614        .flatten()
615        .filter(|owner| !owner.is_empty())
616}
617
618fn component_id(channel: Channel, account: &str) -> String {
619    format!("{}:{account}", channel.name())
620}
621
622fn initial_health(
623    channel: Channel,
624    account: &str,
625    credentials: Option<&Credentials>,
626    enabled: bool,
627) -> ComponentHealth {
628    ComponentHealth {
629        id: component_id(channel, account),
630        channel: channel.name().into(),
631        account: account.into(),
632        bot_id: credentials.and_then(Credentials::bot_id),
633        user_id: credentials.and_then(|c| c.owner().map(str::to_owned)),
634        enabled,
635        state: ComponentState::Starting,
636        last_success_unix_seconds: None,
637        error: None,
638        restarts: 0,
639        remote_tools: RemoteTools::None,
640    }
641}
642
643#[cfg(test)]
644mod tests {
645    use super::*;
646    use std::sync::atomic::{AtomicUsize, Ordering};
647
648    struct Fake {
649        starts: Arc<AtomicUsize>,
650        stops: Arc<AtomicUsize>,
651        fail_first: bool,
652    }
653    #[async_trait]
654    impl Component for Fake {
655        async fn run(&self, cancellation: CancellationToken, health: HealthReporter) -> Result<()> {
656            let attempt = self.starts.fetch_add(1, Ordering::SeqCst);
657            if self.fail_first && attempt == 0 {
658                bail!("secret error must never enter status");
659            }
660            health.contact(true);
661            cancellation.cancelled().await;
662            self.stops.fetch_add(1, Ordering::SeqCst);
663            Ok(())
664        }
665    }
666
667    #[tokio::test]
668    async fn starts_once_recovers_reports_contact_and_joins_before_restoration() {
669        let starts = Arc::new(AtomicUsize::new(0));
670        let stops = Arc::new(AtomicUsize::new(0));
671        let fake = Arc::new(Fake {
672            starts: starts.clone(),
673            stops: stops.clone(),
674            fail_first: true,
675        });
676        let mut supervisor = Supervisor {
677            initial_backoff: Duration::from_millis(10),
678            ..Supervisor::default()
679        };
680        supervisor.start(
681            fake.clone(),
682            initial_health(Channel::Wechat, "test", None, true),
683        );
684        supervisor.start(
685            fake.clone(),
686            initial_health(Channel::Wechat, "test", None, true),
687        );
688        tokio::time::timeout(Duration::from_secs(2), async {
689            loop {
690                if supervisor.health()[0].state == ComponentState::Connected {
691                    break;
692                }
693                tokio::time::sleep(Duration::from_millis(1)).await;
694            }
695        })
696        .await
697        .unwrap();
698        let health = &supervisor.health()[0];
699        assert_eq!(starts.load(Ordering::SeqCst), 2);
700        assert_eq!(health.restarts, 1);
701        assert!(health.last_success_unix_seconds.is_some());
702        assert!(health.error.is_none());
703        supervisor.shutdown().await;
704        assert_eq!(stops.load(Ordering::SeqCst), 1);
705        supervisor.start(fake, initial_health(Channel::Wechat, "test", None, true));
706        tokio::time::sleep(Duration::from_millis(20)).await;
707        supervisor.shutdown().await;
708        assert_eq!(starts.load(Ordering::SeqCst), 3);
709        assert_eq!(stops.load(Ordering::SeqCst), 2);
710    }
711
712    #[test]
713    fn credentials_are_not_connection_evidence() {
714        let health = initial_health(Channel::Wechat, "saved", None, true);
715        assert_eq!(health.state, ComponentState::Starting);
716        assert_eq!(health.last_success_unix_seconds, None);
717    }
718
719    #[tokio::test]
720    async fn busy_account_snapshot_preserves_live_work_but_invalid_settings_stop_it() {
721        let starts = Arc::new(AtomicUsize::new(0));
722        let stops = Arc::new(AtomicUsize::new(0));
723        let mut components = Components::new(PathBuf::from("/unused.sock"), PathBuf::from("/"));
724        components.supervisor.start(
725            Arc::new(Fake {
726                starts: starts.clone(),
727                stops: stops.clone(),
728                fail_first: false,
729            }),
730            initial_health(Channel::Wechat, "test", None, true),
731        );
732        tokio::time::timeout(Duration::from_secs(1), async {
733            while starts.load(Ordering::SeqCst) == 0 {
734                tokio::task::yield_now().await;
735            }
736        })
737        .await
738        .unwrap();
739        components
740            .account_error(
741                Channel::Wechat,
742                "test",
743                std::io::Error::from(std::io::ErrorKind::WouldBlock).into(),
744            )
745            .await;
746        assert_eq!(
747            components.status().components[0].state,
748            ComponentState::Connected
749        );
750        assert_eq!(starts.load(Ordering::SeqCst), 1);
751        assert_eq!(stops.load(Ordering::SeqCst), 0);
752        components
753            .account_error(Channel::Wechat, "test", anyhow::anyhow!("invalid settings"))
754            .await;
755        assert_eq!(stops.load(Ordering::SeqCst), 1);
756        assert_eq!(
757            components.status().components[0].state,
758            ComponentState::Failed
759        );
760    }
761
762    struct Stubborn;
763    #[async_trait]
764    impl Component for Stubborn {
765        async fn run(&self, _: CancellationToken, _: HealthReporter) -> Result<()> {
766            std::future::pending().await
767        }
768    }
769
770    #[tokio::test]
771    async fn bounded_stop_aborts_uncooperative_component_and_cancels_backoff() {
772        let mut supervisor = Supervisor {
773            grace: Duration::from_millis(20),
774            ..Supervisor::default()
775        };
776        supervisor.start(
777            Arc::new(Stubborn),
778            initial_health(Channel::Wechat, "stubborn", None, true),
779        );
780        tokio::task::yield_now().await;
781        tokio::time::timeout(Duration::from_secs(1), supervisor.shutdown())
782            .await
783            .unwrap();
784        assert!(supervisor.health().is_empty());
785        let fake = Arc::new(Fake {
786            starts: Arc::new(AtomicUsize::new(0)),
787            stops: Arc::new(AtomicUsize::new(0)),
788            fail_first: true,
789        });
790        supervisor.start(fake, initial_health(Channel::Wechat, "backoff", None, true));
791        tokio::time::sleep(Duration::from_millis(10)).await;
792        assert_eq!(supervisor.health()[0].state, ComponentState::Backoff);
793        assert_eq!(
794            supervisor.health()[0].error.as_deref(),
795            Some("Component stopped unexpectedly; retrying")
796        );
797        tokio::time::timeout(Duration::from_millis(100), supervisor.shutdown())
798            .await
799            .unwrap();
800    }
801
802    #[test]
803    fn remote_tools_require_owner_mode_and_known_owner() {
804        let wechat = |user_id: Option<&str>| {
805            Credentials::Wechat(scv_clawbot::state::Account {
806                token: "token".into(),
807                base_url: "https://example.invalid".into(),
808                bot_id: Some("bot".into()),
809                user_id: user_id.map(Into::into),
810            })
811        };
812        let feishu = |owner: Option<&str>| {
813            Credentials::Feishu(scv_feishu::state::Account {
814                app_id: "cli_a1b2".into(),
815                app_secret: "secret".into(),
816                brand: scv_feishu::state::Brand::Feishu,
817                owner_open_id: owner.map(Into::into),
818            })
819        };
820        let owner = AccountSettings {
821            remote_tools: RemoteTools::Owner,
822            ..Default::default()
823        };
824        assert_eq!(
825            tool_owner(&wechat(Some("owner@im.wechat")), &owner).as_deref(),
826            Some("owner@im.wechat")
827        );
828        assert_eq!(tool_owner(&wechat(None), &owner), None);
829        assert_eq!(tool_owner(&wechat(Some("")), &owner), None);
830        assert_eq!(
831            tool_owner(
832                &wechat(Some("owner@im.wechat")),
833                &AccountSettings::default()
834            ),
835            None
836        );
837        assert_eq!(
838            tool_owner(&feishu(Some("ou_owner")), &owner).as_deref(),
839            Some("ou_owner")
840        );
841        assert_eq!(tool_owner(&feishu(None), &owner), None);
842        assert_eq!(
843            tool_owner(&feishu(Some("ou_owner")), &AccountSettings::default()),
844            None
845        );
846    }
847
848    #[test]
849    fn channels_are_named_and_health_shows_the_app_and_owner() {
850        assert_eq!(Channel::parse("wechat").unwrap(), Channel::Wechat);
851        assert_eq!(Channel::parse("feishu").unwrap(), Channel::Feishu);
852        assert!(Channel::parse("lark").is_err());
853        let credentials = Credentials::Feishu(scv_feishu::state::Account {
854            app_id: "cli_a1b2".into(),
855            app_secret: "secret".into(),
856            brand: scv_feishu::state::Brand::Feishu,
857            owner_open_id: Some("ou_owner".into()),
858        });
859        let health = initial_health(Channel::Feishu, "default", Some(&credentials), true);
860        assert_eq!(health.id, "feishu:default");
861        assert_eq!(health.channel, "feishu");
862        assert_eq!(health.bot_id.as_deref(), Some("cli_a1b2"));
863        assert_eq!(health.user_id.as_deref(), Some("ou_owner"));
864        assert!(!serde_json::to_string(&health).unwrap().contains("secret"));
865    }
866}