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 from saved credentials.
218    fn account_names(self) -> std::result::Result<Vec<String>, &'static str> {
219        const DISCOVERY: &str =
220            "Account discovery failed; components stopped until configuration is readable";
221        match self {
222            Self::Wechat => scv_clawbot::state::account_names(),
223            Self::Feishu => scv_feishu::state::account_names(),
224        }
225        .map_err(|_| DISCOVERY)
226    }
227
228    fn snapshot(self, account: &str) -> Result<(Option<Credentials>, AccountSettings)> {
229        Ok(match self {
230            Self::Wechat => {
231                let (credentials, settings) = scv_clawbot::state::account_snapshot(account)?;
232                (credentials.map(Credentials::Wechat), settings)
233            }
234            Self::Feishu => {
235                let (credentials, settings) = scv_feishu::state::account_snapshot(account)?;
236                (credentials.map(Credentials::Feishu), settings)
237            }
238        })
239    }
240
241    fn signed_in(self, account: &str) -> Result<bool> {
242        Ok(match self {
243            Self::Wechat => scv_clawbot::state::account(account)?.is_some(),
244            Self::Feishu => scv_feishu::state::account(account)?.is_some(),
245        })
246    }
247
248    fn settings(self, account: &str) -> Result<AccountSettings> {
249        match self {
250            Self::Wechat => scv_clawbot::state::settings(account),
251            Self::Feishu => scv_feishu::state::settings(account),
252        }
253    }
254
255    fn save_settings(self, account: &str, settings: &AccountSettings) -> Result<()> {
256        match self {
257            Self::Wechat => scv_clawbot::state::save_settings(account, settings),
258            Self::Feishu => scv_feishu::state::save_settings(account, settings),
259        }
260    }
261
262    fn remove(self, account: &str) -> Result<()> {
263        match self {
264            Self::Wechat => scv_clawbot::state::remove(account),
265            Self::Feishu => scv_feishu::state::remove(account),
266        }
267    }
268}
269
270impl Credentials {
271    /// The authenticated owner, the only sender remote tools may reach.
272    fn owner(&self) -> Option<&str> {
273        match self {
274            Self::Wechat(account) => account.user_id.as_deref(),
275            Self::Feishu(account) => account.owner_open_id.as_deref(),
276        }
277    }
278
279    /// The bot's identity shown in status: the iLink bot or the Feishu app.
280    fn bot_id(&self) -> Option<String> {
281        match self {
282            Self::Wechat(account) => account.bot_id.clone(),
283            Self::Feishu(account) => Some(account.app_id.clone()),
284        }
285    }
286}
287
288struct ChannelAccount {
289    channel: Channel,
290    account: String,
291    credentials: Credentials,
292    workspace: PathBuf,
293    socket: PathBuf,
294    tool_owner: Option<String>,
295}
296
297#[async_trait]
298impl Component for ChannelAccount {
299    async fn run(&self, cancellation: CancellationToken, health: HealthReporter) -> Result<()> {
300        let tool_owner = self.tool_owner.clone().map(|user_id| {
301            let turn_timeout = scv_channels::owner_turn_timeout(max_tool_timeout(&self.workspace));
302            tracing::info!(
303                "{} {} owner turns may run up to {} seconds",
304                self.channel.title(),
305                self.account,
306                turn_timeout.as_secs()
307            );
308            scv_channels::ToolOwner {
309                user_id,
310                turn_timeout,
311            }
312        });
313        let report = Arc::new(move |connected| health.contact(connected));
314        match &self.credentials {
315            Credentials::Wechat(credentials) => {
316                scv_clawbot::run_supervised(
317                    &credentials.token,
318                    &credentials.base_url,
319                    &self.account,
320                    &self.workspace,
321                    &self.socket,
322                    tool_owner.as_ref(),
323                    cancellation,
324                    report,
325                )
326                .await
327            }
328            Credentials::Feishu(credentials) => {
329                scv_feishu::run_supervised(
330                    credentials,
331                    &self.account,
332                    &self.workspace,
333                    &self.socket,
334                    tool_owner.as_ref(),
335                    cancellation,
336                    report,
337                )
338                .await
339            }
340        }
341    }
342}
343
344pub(crate) struct Components {
345    supervisor: Supervisor,
346    /// Running or disabled accounts by component ID, `<channel>:<account>`.
347    desired: BTreeMap<String, (Credentials, AccountSettings)>,
348    inactive: BTreeMap<String, ComponentHealth>,
349    socket: PathBuf,
350    workspace: PathBuf,
351}
352
353impl Components {
354    pub fn new(socket: PathBuf, workspace: PathBuf) -> Self {
355        Self {
356            supervisor: Supervisor::default(),
357            desired: BTreeMap::new(),
358            inactive: BTreeMap::new(),
359            socket,
360            workspace,
361        }
362    }
363
364    pub fn status(&self) -> DaemonStatus {
365        let mut components = self.supervisor.health();
366        components.extend(self.inactive.values().cloned());
367        components.sort_by(|a, b| a.id.cmp(&b.id));
368        DaemonStatus {
369            version: env!("CARGO_PKG_VERSION").into(),
370            pid: std::process::id(),
371            components,
372            delegations: Default::default(),
373        }
374    }
375
376    /// Match running components to every channel's saved accounts. A channel
377    /// whose accounts cannot be discovered stops its own components and
378    /// reports why; other channels keep running.
379    pub async fn reconcile(&mut self) -> Result<()> {
380        let mut failed = false;
381        for channel in CHANNELS {
382            match channel.account_names() {
383                Ok(names) => self.reconcile_channel(channel, names).await,
384                Err(message) => {
385                    failed = true;
386                    for id in self.ids(channel) {
387                        self.supervisor.stop(&id).await;
388                        self.desired.remove(&id);
389                        self.inactive.remove(&id);
390                    }
391                    let mut health = initial_health(channel, "discovery", None, false);
392                    health.id = component_id(channel, "discovery-error");
393                    health.state = ComponentState::Failed;
394                    health.error = Some(message.into());
395                    self.inactive.insert(health.id.clone(), health);
396                }
397            }
398        }
399        if failed {
400            bail!("Account discovery failed");
401        }
402        Ok(())
403    }
404
405    /// Component IDs of one channel, running or not.
406    fn ids(&self, channel: Channel) -> Vec<String> {
407        let prefix = format!("{}:", channel.name());
408        self.desired
409            .keys()
410            .chain(self.inactive.keys())
411            .filter(|id| id.starts_with(&prefix))
412            .cloned()
413            .collect()
414    }
415
416    async fn reconcile_channel(&mut self, channel: Channel, names: Vec<String>) {
417        let wanted: Vec<String> = names
418            .iter()
419            .map(|name| component_id(channel, name))
420            .collect();
421        for id in self.ids(channel) {
422            if !wanted.contains(&id) {
423                self.supervisor.stop(&id).await;
424                self.desired.remove(&id);
425                self.inactive.remove(&id);
426            }
427        }
428        for name in names {
429            let id = component_id(channel, &name);
430            let loaded = (|| -> Result<_> {
431                let (account, settings) = channel.snapshot(&name)?;
432                Ok((
433                    account.ok_or_else(|| anyhow::anyhow!("missing account"))?,
434                    settings,
435                ))
436            })();
437            let (credentials, settings) = match loaded {
438                Ok(value) => value,
439                Err(error) => {
440                    self.account_error(channel, &name, error).await;
441                    continue;
442                }
443            };
444            if self.desired.get(&id) == Some(&(credentials.clone(), settings.clone())) {
445                continue;
446            }
447            self.supervisor.stop(&id).await;
448            self.inactive.remove(&id);
449            let mut health = initial_health(channel, &name, Some(&credentials), settings.enabled);
450            let tool_owner = tool_owner(&credentials, &settings);
451            if tool_owner.is_some() {
452                health.remote_tools = RemoteTools::Owner;
453            }
454            if settings.enabled {
455                let workspace = settings
456                    .workspace
457                    .clone()
458                    .unwrap_or_else(|| self.workspace.clone());
459                if !workspace.is_absolute() || !workspace.is_dir() {
460                    health.state = ComponentState::Failed;
461                    health.error =
462                        Some("Component workspace must be an existing absolute directory".into());
463                    self.inactive.insert(id.clone(), health);
464                    self.desired.remove(&id);
465                    continue;
466                }
467                self.supervisor.start(
468                    Arc::new(ChannelAccount {
469                        channel,
470                        account: name.clone(),
471                        credentials: credentials.clone(),
472                        workspace,
473                        socket: self.socket.clone(),
474                        tool_owner,
475                    }),
476                    health,
477                );
478            } else {
479                health.state = ComponentState::Disabled;
480                self.inactive.insert(id.clone(), health);
481            }
482            self.desired.insert(id, (credentials, settings));
483        }
484    }
485
486    async fn account_error(&mut self, channel: Channel, name: &str, error: anyhow::Error) {
487        // A bridge state commit briefly holds this same lock. Retry next refresh
488        // rather than interrupting healthy work for ordinary lock contention.
489        if is_busy(&error) {
490            return;
491        }
492        let id = component_id(channel, name);
493        self.supervisor.stop(&id).await;
494        self.desired.remove(&id);
495        let mut health = initial_health(channel, name, None, true);
496        health.state = ComponentState::Failed;
497        health.error = Some(
498            "Invalid or inaccessible account or settings; `scv config show` says which".into(),
499        );
500        self.inactive.insert(id, health);
501    }
502
503    pub async fn control(&mut self, command: DaemonCommand) -> Result<DaemonStatus> {
504        match command {
505            // Delegations belong to the connection handler, which adds them.
506            DaemonCommand::Status
507            | DaemonCommand::Delegations { .. }
508            | DaemonCommand::DelegationKill { .. } => return Ok(self.status()),
509            DaemonCommand::Reload => {}
510            DaemonCommand::ChannelSet {
511                channel,
512                account,
513                enabled,
514                workspace,
515                remote_tools,
516            } => {
517                let channel = Channel::parse(&channel)?;
518                state::validate_name(&account)?;
519                let workspace = match workspace {
520                    Some(path) => {
521                        let path = PathBuf::from(path);
522                        if !path.is_absolute() || !path.is_dir() {
523                            bail!("Invalid component workspace");
524                        }
525                        Some(std::fs::canonicalize(path)?)
526                    }
527                    None => None,
528                };
529                retry_while_busy(|| {
530                    if !channel.signed_in(&account)? {
531                        bail!("Account is not logged in");
532                    }
533                    let mut settings = channel.settings(&account)?;
534                    settings.enabled = enabled;
535                    if let Some(path) = &workspace {
536                        settings.workspace = Some(path.clone());
537                    }
538                    if let Some(mode) = remote_tools {
539                        settings.remote_tools = mode;
540                    }
541                    channel.save_settings(&account, &settings)
542                })
543                .await?;
544            }
545            DaemonCommand::ChannelLogout { channel, account } => {
546                let channel = Channel::parse(&channel)?;
547                state::validate_name(&account)?;
548                // Persist disabled and tool-free first, so failed deletion can
549                // neither resurrect a live account nor hand a later login the grant.
550                retry_while_busy(|| {
551                    let mut settings = channel.settings(&account)?;
552                    settings.enabled = false;
553                    settings.remote_tools = RemoteTools::None;
554                    channel.save_settings(&account, &settings)
555                })
556                .await?;
557                let id = component_id(channel, &account);
558                self.supervisor.stop(&id).await;
559                self.desired.remove(&id);
560                self.inactive.remove(&id);
561                retry_while_busy(|| channel.remove(&account)).await?;
562            }
563        }
564        self.reconcile().await?;
565        Ok(self.status())
566    }
567
568    pub async fn shutdown(&mut self) {
569        self.supervisor.shutdown().await;
570    }
571}
572
573/// The longest tool call an owner session in `workspace` may make, from the
574/// configuration its sessions load. Read at each (re)start of the component.
575fn max_tool_timeout(workspace: &std::path::Path) -> std::time::Duration {
576    let seconds = crate::Config::load(workspace, crate::ConfigOverrides::default())
577        .map(|config| config.tools.max_timeout_seconds)
578        .unwrap_or_else(|error| {
579            tracing::warn!("Channel owner turns use the default tool timeout ceiling: {error:#}");
580            crate::config::ToolConfig::default().max_timeout_seconds
581        });
582    std::time::Duration::from_secs(seconds)
583}
584
585/// Account transactions fail fast while their lock is held, and a running
586/// bridge holds it for every state commit. Operator commands retry through
587/// that contention instead of failing whenever they coincide with a commit.
588async fn retry_while_busy<T>(mut operation: impl FnMut() -> Result<T>) -> Result<T> {
589    let deadline = tokio::time::Instant::now() + BUSY_RETRY;
590    loop {
591        match operation() {
592            Err(error) if is_busy(&error) && tokio::time::Instant::now() < deadline => {
593                tokio::time::sleep(Duration::from_millis(10)).await;
594            }
595            result => return result,
596        }
597    }
598}
599
600fn is_busy(error: &anyhow::Error) -> bool {
601    error
602        .downcast_ref::<std::io::Error>()
603        .is_some_and(|error| error.kind() == std::io::ErrorKind::WouldBlock)
604}
605
606/// Only the authenticated account owner may receive tools. Credentials without
607/// a known owner ID grant tools to nobody, even when the setting asks for it.
608fn tool_owner(credentials: &Credentials, settings: &AccountSettings) -> Option<String> {
609    (settings.remote_tools == RemoteTools::Owner)
610        .then(|| credentials.owner().map(str::to_owned))
611        .flatten()
612        .filter(|owner| !owner.is_empty())
613}
614
615fn component_id(channel: Channel, account: &str) -> String {
616    format!("{}:{account}", channel.name())
617}
618
619fn initial_health(
620    channel: Channel,
621    account: &str,
622    credentials: Option<&Credentials>,
623    enabled: bool,
624) -> ComponentHealth {
625    ComponentHealth {
626        id: component_id(channel, account),
627        channel: channel.name().into(),
628        account: account.into(),
629        bot_id: credentials.and_then(Credentials::bot_id),
630        user_id: credentials.and_then(|c| c.owner().map(str::to_owned)),
631        enabled,
632        state: ComponentState::Starting,
633        last_success_unix_seconds: None,
634        error: None,
635        restarts: 0,
636        remote_tools: RemoteTools::None,
637    }
638}
639
640#[cfg(test)]
641mod tests {
642    use super::*;
643    use std::sync::atomic::{AtomicUsize, Ordering};
644
645    struct Fake {
646        starts: Arc<AtomicUsize>,
647        stops: Arc<AtomicUsize>,
648        fail_first: bool,
649    }
650    #[async_trait]
651    impl Component for Fake {
652        async fn run(&self, cancellation: CancellationToken, health: HealthReporter) -> Result<()> {
653            let attempt = self.starts.fetch_add(1, Ordering::SeqCst);
654            if self.fail_first && attempt == 0 {
655                bail!("secret error must never enter status");
656            }
657            health.contact(true);
658            cancellation.cancelled().await;
659            self.stops.fetch_add(1, Ordering::SeqCst);
660            Ok(())
661        }
662    }
663
664    #[tokio::test]
665    async fn starts_once_recovers_reports_contact_and_joins_before_restoration() {
666        let starts = Arc::new(AtomicUsize::new(0));
667        let stops = Arc::new(AtomicUsize::new(0));
668        let fake = Arc::new(Fake {
669            starts: starts.clone(),
670            stops: stops.clone(),
671            fail_first: true,
672        });
673        let mut supervisor = Supervisor {
674            initial_backoff: Duration::from_millis(10),
675            ..Supervisor::default()
676        };
677        supervisor.start(
678            fake.clone(),
679            initial_health(Channel::Wechat, "test", None, true),
680        );
681        supervisor.start(
682            fake.clone(),
683            initial_health(Channel::Wechat, "test", None, true),
684        );
685        tokio::time::timeout(Duration::from_secs(2), async {
686            loop {
687                if supervisor.health()[0].state == ComponentState::Connected {
688                    break;
689                }
690                tokio::time::sleep(Duration::from_millis(1)).await;
691            }
692        })
693        .await
694        .unwrap();
695        let health = &supervisor.health()[0];
696        assert_eq!(starts.load(Ordering::SeqCst), 2);
697        assert_eq!(health.restarts, 1);
698        assert!(health.last_success_unix_seconds.is_some());
699        assert!(health.error.is_none());
700        supervisor.shutdown().await;
701        assert_eq!(stops.load(Ordering::SeqCst), 1);
702        supervisor.start(fake, initial_health(Channel::Wechat, "test", None, true));
703        tokio::time::sleep(Duration::from_millis(20)).await;
704        supervisor.shutdown().await;
705        assert_eq!(starts.load(Ordering::SeqCst), 3);
706        assert_eq!(stops.load(Ordering::SeqCst), 2);
707    }
708
709    #[test]
710    fn credentials_are_not_connection_evidence() {
711        let health = initial_health(Channel::Wechat, "saved", None, true);
712        assert_eq!(health.state, ComponentState::Starting);
713        assert_eq!(health.last_success_unix_seconds, None);
714    }
715
716    #[tokio::test]
717    async fn busy_account_snapshot_preserves_live_work_but_invalid_settings_stop_it() {
718        let starts = Arc::new(AtomicUsize::new(0));
719        let stops = Arc::new(AtomicUsize::new(0));
720        let mut components = Components::new(PathBuf::from("/unused.sock"), PathBuf::from("/"));
721        components.supervisor.start(
722            Arc::new(Fake {
723                starts: starts.clone(),
724                stops: stops.clone(),
725                fail_first: false,
726            }),
727            initial_health(Channel::Wechat, "test", None, true),
728        );
729        tokio::time::timeout(Duration::from_secs(1), async {
730            while starts.load(Ordering::SeqCst) == 0 {
731                tokio::task::yield_now().await;
732            }
733        })
734        .await
735        .unwrap();
736        components
737            .account_error(
738                Channel::Wechat,
739                "test",
740                std::io::Error::from(std::io::ErrorKind::WouldBlock).into(),
741            )
742            .await;
743        assert_eq!(
744            components.status().components[0].state,
745            ComponentState::Connected
746        );
747        assert_eq!(starts.load(Ordering::SeqCst), 1);
748        assert_eq!(stops.load(Ordering::SeqCst), 0);
749        components
750            .account_error(Channel::Wechat, "test", anyhow::anyhow!("invalid settings"))
751            .await;
752        assert_eq!(stops.load(Ordering::SeqCst), 1);
753        assert_eq!(
754            components.status().components[0].state,
755            ComponentState::Failed
756        );
757    }
758
759    struct Stubborn;
760    #[async_trait]
761    impl Component for Stubborn {
762        async fn run(&self, _: CancellationToken, _: HealthReporter) -> Result<()> {
763            std::future::pending().await
764        }
765    }
766
767    #[tokio::test]
768    async fn bounded_stop_aborts_uncooperative_component_and_cancels_backoff() {
769        let mut supervisor = Supervisor {
770            grace: Duration::from_millis(20),
771            ..Supervisor::default()
772        };
773        supervisor.start(
774            Arc::new(Stubborn),
775            initial_health(Channel::Wechat, "stubborn", None, true),
776        );
777        tokio::task::yield_now().await;
778        tokio::time::timeout(Duration::from_secs(1), supervisor.shutdown())
779            .await
780            .unwrap();
781        assert!(supervisor.health().is_empty());
782        let fake = Arc::new(Fake {
783            starts: Arc::new(AtomicUsize::new(0)),
784            stops: Arc::new(AtomicUsize::new(0)),
785            fail_first: true,
786        });
787        supervisor.start(fake, initial_health(Channel::Wechat, "backoff", None, true));
788        tokio::time::sleep(Duration::from_millis(10)).await;
789        assert_eq!(supervisor.health()[0].state, ComponentState::Backoff);
790        assert_eq!(
791            supervisor.health()[0].error.as_deref(),
792            Some("Component stopped unexpectedly; retrying")
793        );
794        tokio::time::timeout(Duration::from_millis(100), supervisor.shutdown())
795            .await
796            .unwrap();
797    }
798
799    #[test]
800    fn remote_tools_require_owner_mode_and_known_owner() {
801        let wechat = |user_id: Option<&str>| {
802            Credentials::Wechat(scv_clawbot::state::Account {
803                token: "token".into(),
804                base_url: "https://example.invalid".into(),
805                bot_id: Some("bot".into()),
806                user_id: user_id.map(Into::into),
807            })
808        };
809        let feishu = |owner: Option<&str>| {
810            Credentials::Feishu(scv_feishu::state::Account {
811                app_id: "cli_a1b2".into(),
812                app_secret: "secret".into(),
813                brand: scv_feishu::state::Brand::Feishu,
814                owner_open_id: owner.map(Into::into),
815            })
816        };
817        let owner = AccountSettings {
818            remote_tools: RemoteTools::Owner,
819            ..Default::default()
820        };
821        assert_eq!(
822            tool_owner(&wechat(Some("owner@im.wechat")), &owner).as_deref(),
823            Some("owner@im.wechat")
824        );
825        assert_eq!(tool_owner(&wechat(None), &owner), None);
826        assert_eq!(tool_owner(&wechat(Some("")), &owner), None);
827        assert_eq!(
828            tool_owner(
829                &wechat(Some("owner@im.wechat")),
830                &AccountSettings::default()
831            ),
832            None
833        );
834        assert_eq!(
835            tool_owner(&feishu(Some("ou_owner")), &owner).as_deref(),
836            Some("ou_owner")
837        );
838        assert_eq!(tool_owner(&feishu(None), &owner), None);
839        assert_eq!(
840            tool_owner(&feishu(Some("ou_owner")), &AccountSettings::default()),
841            None
842        );
843    }
844
845    #[test]
846    fn channels_are_named_and_health_shows_the_app_and_owner() {
847        assert_eq!(Channel::parse("wechat").unwrap(), Channel::Wechat);
848        assert_eq!(Channel::parse("feishu").unwrap(), Channel::Feishu);
849        assert!(Channel::parse("lark").is_err());
850        let credentials = Credentials::Feishu(scv_feishu::state::Account {
851            app_id: "cli_a1b2".into(),
852            app_secret: "secret".into(),
853            brand: scv_feishu::state::Brand::Feishu,
854            owner_open_id: Some("ou_owner".into()),
855        });
856        let health = initial_health(Channel::Feishu, "default", Some(&credentials), true);
857        assert_eq!(health.id, "feishu:default");
858        assert_eq!(health.channel, "feishu");
859        assert_eq!(health.bot_id.as_deref(), Some("cli_a1b2"));
860        assert_eq!(health.user_id.as_deref(), Some("ou_owner"));
861        assert!(!serde_json::to_string(&health).unwrap().contains("secret"));
862    }
863}