1use 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);
17const BUSY_RETRY: Duration = Duration::from_secs(5);
19
20#[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 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 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 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#[derive(Clone, Copy, Debug, PartialEq, Eq)]
181enum Channel {
182 Wechat,
183 Feishu,
184}
185
186const CHANNELS: [Channel; 2] = [Channel::Wechat, Channel::Feishu];
187
188#[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 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 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 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 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 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 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 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 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 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
576fn 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
588async 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
609fn 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}