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> {
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 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 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 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 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 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 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 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 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
573fn 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
585async 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
606fn 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}