feat: group access, notices

src-tauri/src/agent_schedules.rs

@@ -29,12 +29,16 @@ fn url_for(id: &str, tail: &str) -> Result<String, String> {
   Ok(format!("{}/{}{tail}", base(), urlencoding::encode(id)))
 }
 
-fn failure(action: &str, err: BackendFailure) -> String {
+fn failure(action: &str, mut err: BackendFailure) -> String {
   log::warn!(
     "Schedule request failed action={action} code={} status={}",
     err.code,
     err.status
   );
+  // The cloud names a schedule a job; the app has one word for it.
+  if err.code == "AGENT_JOB_NOT_FOUND" {
+    err.code = "AGENT_SCHEDULE_NOT_FOUND".to_string();
+  }
   err.to_error_json()
 }
 
@@ -110,6 +114,18 @@ pub async fn list_agent_schedule_runs(id: String) -> Result<Value, String> {
   call("runs", reqwest::Method::GET, url_for(&id, "/runs")?, None).await
 }
 
+/// Send a test email and webhook for the schedule `id`, as a failed run would.
+#[tauri::command]
+pub async fn test_agent_schedule_notice(id: String) -> Result<Value, String> {
+  call(
+    "test_notice",
+    reqwest::Method::POST,
+    url_for(&id, "/test-notice")?,
+    Some(serde_json::json!({})),
+  )
+  .await
+}
+
 #[cfg(test)]
 mod tests {
   use super::*;

src-tauri/src/cloud_api.rs

@@ -28,6 +28,28 @@ pub(crate) async fn call<T: DeserializeOwned>(
   body: Option<serde_json::Value>,
   codes: FailureCodes,
 ) -> Result<T, BackendFailure> {
+  let bytes = send(method, url, body, codes).await?;
+  decode_body::<T>(&bytes)
+    .map_err(|e| cloud_errors::classify_message(&format!("decode response: {e}"), codes))
+}
+
+/// `call` for an answer that is text (a CSV file), not JSON.
+pub(crate) async fn call_text(
+  method: reqwest::Method,
+  url: String,
+  codes: FailureCodes,
+) -> Result<String, BackendFailure> {
+  let bytes = send(method, url, None, codes).await?;
+  String::from_utf8(bytes)
+    .map_err(|e| cloud_errors::classify_message(&format!("decode response: {e}"), codes))
+}
+
+async fn send(
+  method: reqwest::Method,
+  url: String,
+  body: Option<serde_json::Value>,
+  codes: FailureCodes,
+) -> Result<Vec<u8>, BackendFailure> {
   crate::cloud_auth::CLOUD_AUTH
     .api_call_with_retry(|token| {
       let method = method.clone();
@@ -48,11 +70,11 @@ pub(crate) async fn call<T: DeserializeOwned>(
           // api_call_with_retry and classify_message read the status from this prefix.
           return Err(format!("({status}) {text}"));
         }
-        let bytes = response
+        response
           .bytes()
           .await
-          .map_err(|e| format!("decode response: {e}"))?;
-        decode_body::<T>(&bytes).map_err(|e| format!("decode response: {e}"))
+          .map(|bytes| bytes.to_vec())
+          .map_err(|e| format!("decode response: {e}"))
       }
     })
     .await

src-tauri/src/lib.rs

@@ -160,6 +160,7 @@ mod navigation_watch;
 mod tag_manager;
 pub mod team_activity;
 mod team_lock;
+mod team_members;
 pub mod team_roles;
 mod totp;
 mod vault;
@@ -3557,6 +3558,7 @@ pub fn run_with_builder(
       agent_schedules::delete_agent_schedule,
       agent_schedules::run_agent_schedule_now,
       agent_schedules::list_agent_schedule_runs,
+      agent_schedules::test_agent_schedule_notice,
       start_mcp_remote_bridge,
       stop_mcp_remote_bridge,
       get_mcp_remote_status,
@@ -3658,6 +3660,9 @@ pub fn run_with_builder(
       lock_profile,
       is_profile_locked,
       team_activity::list_team_activity,
+      team_activity::export_team_activity_csv,
+      team_members::list_team_members,
+      team_members::set_team_member_access,
     ]))
     .build(tauri::generate_context!())
     .expect("error while building tauri application")

src-tauri/src/sync/engine.rs

@@ -23,6 +23,23 @@ use tokio::sync::{Mutex as TokioMutex, Semaphore};
 /// entity's user-edit timestamp in unix seconds. Used to resolve sync conflicts
 /// (last-write-wins) from a HEAD request without downloading the object body.
 const UPDATED_AT_META_KEY: &str = "updated-at";
+/// The profile's group, in plain text on its metadata object, so the sync
+/// server can hold a team member to the groups the owner gave them.
+const GROUP_ID_META_KEY: &str = "group-id";
+
+/// Set when the server signed an upload without `group-id`: an older server
+/// that keeps no group, so asking it again on every sync changes nothing.
+static GROUP_META_UNSUPPORTED: std::sync::atomic::AtomicBool =
+  std::sync::atomic::AtomicBool::new(false);
+
+/// The `group-id` a profile's metadata object carries.
+fn group_meta(profile: &BrowserProfile) -> String {
+  profile
+    .group_id
+    .clone()
+    .filter(|id| !id.is_empty())
+    .unwrap_or_else(|| crate::team_members::UNGROUPED.to_string())
+}
 
 static TOMBSTONE_CHECK: KeyedStreak = KeyedStreak::new(module_path!(), "Sync tombstone check");
 
@@ -598,6 +615,15 @@ impl SyncEngine {
     String::new()
   }
 
+  /// Whether the signed-in account holds a team seat, where access by group
+  /// applies.
+  async fn is_team_seat() -> bool {
+    crate::cloud_auth::CLOUD_AUTH
+      .get_user()
+      .await
+      .is_some_and(|auth| auth.user.team_id.is_some())
+  }
+
   /// Check if this is a self-hosted sync (no cloud login).
   async fn is_self_hosted_sync() -> bool {
     !crate::cloud_auth::CLOUD_AUTH.is_logged_in().await
@@ -643,18 +669,38 @@ impl SyncEngine {
     remote_key: &str,
     json: &str,
     updated_at: u64,
+  ) -> SyncResult<()> {
+    self
+      .upload_config_json_with(remote_key, json, updated_at, HashMap::new())
+      .await
+  }
+
+  /// `upload_config_json`, with more plain-text object metadata.
+  async fn upload_config_json_with(
+    &self,
+    remote_key: &str,
+    json: &str,
+    updated_at: u64,
+    mut meta: HashMap<String, String>,
   ) -> SyncResult<()> {
     if self.skips_remote_write(&format!("key={remote_key}")) {
       return Ok(());
     }
     let (payload, content_type) = encryption::maybe_seal_for_upload(json.as_bytes())
       .map_err(|e| SyncError::InvalidData(format!("Failed to seal config: {e}")))?;
-    let mut meta = HashMap::new();
     meta.insert(UPDATED_AT_META_KEY.to_string(), updated_at.to_string());
+    let wants_group = meta.contains_key(GROUP_ID_META_KEY);
     let presign = self
       .client
       .presign_upload_with_metadata(remote_key, Some(content_type), Some(meta))
       .await?;
+    if wants_group {
+      let kept = presign
+        .metadata
+        .as_ref()
+        .is_some_and(|signed| signed.contains_key(GROUP_ID_META_KEY));
+      GROUP_META_UNSUPPORTED.store(!kept, std::sync::atomic::Ordering::Relaxed);
+    }
     self
       .client
       .upload_bytes_with_metadata(
@@ -1255,7 +1301,13 @@ impl SyncEngine {
     let local_updated = profile.updated_at.unwrap_or(0);
     let remote_updated = self.remote_updated_at(&stat, &remote_key).await;
     if !self.remote_wins(local_updated, remote_updated) {
-      if local_updated > remote_updated {
+      // An object an older app wrote has no group, and a member limited to
+      // some groups cannot reach it until it has one.
+      let group_differs = stat.metadata.as_ref().is_some_and(|meta| {
+        meta.get(GROUP_ID_META_KEY).map(String::as_str) != Some(group_meta(profile).as_str())
+      }) && !GROUP_META_UNSUPPORTED.load(std::sync::atomic::Ordering::Relaxed)
+        && Self::is_team_seat().await;
+      if local_updated > remote_updated || group_differs {
         self
           .upload_profile_metadata(&profile_id, profile, &key_prefix)
           .await?;
@@ -1322,8 +1374,9 @@ impl SyncEngine {
       .map_err(|e| SyncError::SerializationError(format!("Failed to serialize profile: {e}")))?;
 
     let remote_key = format!("{}profiles/{}/metadata.json", key_prefix, profile_id);
+    let meta = HashMap::from([(GROUP_ID_META_KEY.to_string(), group_meta(&sanitized))]);
     self
-      .upload_config_json(&remote_key, &json, sanitized.updated_at.unwrap_or(0))
+      .upload_config_json_with(&remote_key, &json, sanitized.updated_at.unwrap_or(0), meta)
       .await?;
 
     Ok(())

src-tauri/src/team_activity.rs

@@ -163,39 +163,144 @@ async fn flush() {
   }
 }
 
+/// What part of the team's log to read. Every field is optional.
+#[derive(Debug, Default, Clone, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ActivityFilter {
+  /// An action (`profile.delete`) or every action of a kind (`profile.*`).
+  pub action: Option<String>,
+  pub actor_user_id: Option<String>,
+  pub target_type: Option<String>,
+  /// The oldest time to include, ISO 8601.
+  pub since: Option<String>,
+}
+
+const TARGET_TYPES: &[&str] = &[
+  "profile",
+  "proxy",
+  "group",
+  "vpn",
+  "extension",
+  "template",
+  "status",
+  "member",
+  "invitation",
+];
+
+impl ActivityFilter {
+  /// The query pairs this filter adds, or `TEAM_ACTIVITY_INVALID`.
+  fn query(&self) -> Result<Vec<String>, String> {
+    let invalid = || crate::backend_error("TEAM_ACTIVITY_INVALID");
+    let present = |value: &Option<String>| value.clone().filter(|v| !v.is_empty());
+    let mut query = Vec::new();
+    if let Some(action) = present(&self.action) {
+      let (kind, verb) = action.split_once('.').ok_or_else(invalid)?;
+      let word = |w: &str| !w.is_empty() && w.bytes().all(|b| b.is_ascii_lowercase() || b == b'_');
+      if action.len() > 64 || !word(kind) || !(verb == "*" || word(verb)) {
+        return Err(invalid());
+      }
+      query.push(format!("action={}", urlencoding::encode(&action)));
+    }
+    if let Some(actor) = present(&self.actor_user_id) {
+      if actor.len() > 128
+        || !actor
+          .bytes()
+          .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
+      {
+        return Err(invalid());
+      }
+      query.push(format!("actorUserId={actor}"));
+    }
+    if let Some(target_type) = present(&self.target_type) {
+      if !TARGET_TYPES.contains(&target_type.as_str()) {
+        return Err(invalid());
+      }
+      query.push(format!("targetType={target_type}"));
+    }
+    if let Some(since) = present(&self.since) {
+      let at = chrono::DateTime::parse_from_rfc3339(&since).map_err(|_| invalid())?;
+      query.push(format!(
+        "since={}",
+        urlencoding::encode(
+          &at
+            .to_utc()
+            .to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
+        )
+      ));
+    }
+    Ok(query)
+  }
+}
+
+async fn signed_in_team() -> Result<String, String> {
+  crate::cloud_auth::CLOUD_AUTH
+    .get_user()
+    .await
+    .and_then(|auth| auth.user.team_id)
+    .ok_or_else(|| crate::backend_error("TEAM_ACTIVITY_FORBIDDEN"))
+}
+
+fn activity_failure(err: crate::cloud_errors::BackendFailure) -> String {
+  log::warn!(
+    "Team activity request failed code={} status={}",
+    err.code,
+    err.status
+  );
+  err.to_error_json()
+}
+
 /// A page of the team's log, newest first. Owners and admins only; the cloud
 /// checks the role and the plan.
 #[tauri::command]
 pub async fn list_team_activity(
   cursor: Option<String>,
   limit: Option<u32>,
-  action: Option<String>,
+  filter: Option<ActivityFilter>,
 ) -> Result<Value, String> {
-  let Some(team_id) = crate::cloud_auth::CLOUD_AUTH
-    .get_user()
-    .await
-    .and_then(|auth| auth.user.team_id)
-  else {
-    return Err(crate::backend_error("TEAM_ACTIVITY_FORBIDDEN"));
-  };
-  let url = activity_url(&team_id, cursor.as_deref(), limit, action.as_deref())?;
+  let team_id = signed_in_team().await?;
+  let url = activity_url(
+    &team_id,
+    "activity",
+    cursor.as_deref(),
+    limit,
+    &filter.unwrap_or_default(),
+  )?;
   crate::cloud_api::call::<Value>(reqwest::Method::GET, url, None, ACTIVITY_CODES)
     .await
-    .map_err(|err| {
-      log::warn!(
-        "Team activity request failed code={} status={}",
-        err.code,
-        err.status
-      );
-      err.to_error_json()
-    })
+    .map_err(activity_failure)
+}
+
+/// The filtered log as a CSV file at `path`, which the person chose. Returns
+/// how many rows it holds.
+#[tauri::command]
+pub async fn export_team_activity_csv(
+  path: String,
+  filter: Option<ActivityFilter>,
+) -> Result<u64, String> {
+  let team_id = signed_in_team().await?;
+  let url = activity_url(
+    &team_id,
+    "activity.csv",
+    None,
+    None,
+    &filter.unwrap_or_default(),
+  )?;
+  let csv = crate::cloud_api::call_text(reqwest::Method::GET, url, ACTIVITY_CODES)
+    .await
+    .map_err(activity_failure)?;
+  std::fs::write(&path, csv.as_bytes())
+    .map_err(|e| crate::backend_error_with_detail("TEAM_ACTIVITY_EXPORT_FAILED", e))?;
+  let rows = csv.lines().count().saturating_sub(1) as u64;
+  log::info!("Team activity exported rows={rows}");
+  Ok(rows)
 }
 
 fn activity_url(
   team_id: &str,
+  resource: &str,
   cursor: Option<&str>,
   limit: Option<u32>,
-  action: Option<&str>,
+  filter: &ActivityFilter,
 ) -> Result<String, String> {
   let mut query = Vec::new();
   if let Some(cursor) = cursor.filter(|cursor| !cursor.is_empty()) {
@@ -207,11 +312,9 @@ fn activity_url(
   if let Some(limit) = limit {
     query.push(format!("limit={}", limit.clamp(1, 200)));
   }
-  if let Some(action) = action.filter(|action| !action.is_empty()) {
-    query.push(format!("action={}", urlencoding::encode(&clip(action, 64))));
-  }
+  query.extend(filter.query()?);
   let mut url = format!(
-    "{}/api/teams/{}/activity",
+    "{}/api/teams/{}/{resource}",
     crate::cloud_auth::CLOUD_API_URL,
     urlencoding::encode(team_id)
   );
@@ -264,14 +367,52 @@ mod tests {
 
   #[test]
   fn the_activity_url_keeps_the_team_id_one_segment() {
-    let url = activity_url("a/b", Some("42"), Some(500), Some("profile.delete")).unwrap();
+    let filter = ActivityFilter {
+      action: Some("profile.*".into()),
+      actor_user_id: Some("u-1".into()),
+      target_type: Some("proxy".into()),
+      since: Some("2026-10-01T12:00:00+02:00".into()),
+    };
+    let url = activity_url("a/b", "activity", Some("42"), Some(500), &filter).unwrap();
     assert!(url.contains("/api/teams/a%2Fb/activity?"), "{url}");
     assert!(url.contains("cursor=42"));
     assert!(url.contains("limit=200"));
-    assert!(url.contains("action=profile.delete"));
-    assert!(activity_url("t1", None, None, None)
+    assert!(url.contains("action=profile.%2A"), "{url}");
+    assert!(url.contains("actorUserId=u-1"));
+    assert!(url.contains("targetType=proxy"));
+    assert!(url.contains("since=2026-10-01T10%3A00%3A00Z"), "{url}");
+    let plain = ActivityFilter::default();
+    assert!(activity_url("t1", "activity.csv", None, None, &plain)
       .unwrap()
-      .ends_with("/api/teams/t1/activity"));
-    assert!(activity_url("t1", Some("1;drop"), None, None).is_err());
+      .ends_with("/api/teams/t1/activity.csv"));
+    assert!(activity_url("t1", "activity", Some("1;drop"), None, &plain).is_err());
+  }
+
+  #[test]
+  fn a_filter_the_cloud_would_refuse_is_refused_here() {
+    for filter in [
+      ActivityFilter {
+        action: Some("Profile.delete".into()),
+        ..Default::default()
+      },
+      ActivityFilter {
+        action: Some("profile".into()),
+        ..Default::default()
+      },
+      ActivityFilter {
+        actor_user_id: Some("a&b=c".into()),
+        ..Default::default()
+      },
+      ActivityFilter {
+        target_type: Some("planet".into()),
+        ..Default::default()
+      },
+      ActivityFilter {
+        since: Some("yesterday".into()),
+        ..Default::default()
+      },
+    ] {
+      assert!(filter.query().is_err(), "{filter:?}");
+    }
   }
 }

src-tauri/src/team_members.rs

@@ -0,0 +1,135 @@
+//! The team's members and the profile groups each one may reach. Owners and
+//! admins set the access here, because group names live only in the app; the
+//! cloud stores group ids and the sync server holds each member to them.
+
+use serde_json::{json, Value};
+
+use crate::cloud_errors::FailureCodes;
+
+/// The `group-id` the sync server uses for profiles in no group.
+pub const UNGROUPED: &str = "__ungrouped__";
+const MAX_GROUPS: usize = 500;
+
+const MEMBER_CODES: FailureCodes = FailureCodes {
+  bad_request: "TEAM_ACCESS_INVALID",
+  forbidden: "TEAM_ACCESS_FORBIDDEN",
+  not_found: "TEAM_MEMBER_NOT_FOUND",
+  conflict: "TEAM_ACCESS_INVALID",
+};
+
+async fn team_id() -> Result<String, String> {
+  crate::cloud_auth::CLOUD_AUTH
+    .get_user()
+    .await
+    .and_then(|auth| auth.user.team_id)
+    .ok_or_else(|| crate::backend_error("TEAM_ACCESS_FORBIDDEN"))
+}
+
+fn failure(action: &str, err: crate::cloud_errors::BackendFailure) -> String {
+  log::warn!(
+    "Team member request failed action={action} code={} status={}",
+    err.code,
+    err.status
+  );
+  err.to_error_json()
+}
+
+fn is_group_id(id: &str) -> bool {
+  (1..=128).contains(&id.len())
+    && id
+      .bytes()
+      .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
+}
+
+/// The access to send: `None` is every group; a list is deduplicated, in order.
+fn checked_groups(groups: Option<Vec<String>>) -> Result<Option<Vec<String>>, String> {
+  let Some(groups) = groups else {
+    return Ok(None);
+  };
+  let mut seen = std::collections::HashSet::new();
+  let groups: Vec<String> = groups
+    .into_iter()
+    .map(|id| id.trim().to_string())
+    .filter(|id| seen.insert(id.clone()))
+    .collect();
+  if groups.len() > MAX_GROUPS || !groups.iter().all(|id| is_group_id(id)) {
+    return Err(crate::backend_error("TEAM_ACCESS_INVALID"));
+  }
+  Ok(Some(groups))
+}
+
+fn members_url(team_id: &str, tail: &str) -> String {
+  format!(
+    "{}/api/teams/{}/members{tail}",
+    crate::cloud_auth::CLOUD_API_URL,
+    urlencoding::encode(team_id)
+  )
+}
+
+#[tauri::command]
+pub async fn list_team_members() -> Result<Value, String> {
+  let team = team_id().await?;
+  crate::cloud_api::call::<Value>(
+    reqwest::Method::GET,
+    members_url(&team, ""),
+    None,
+    MEMBER_CODES,
+  )
+  .await
+  .map_err(|e| failure("list", e))
+}
+
+/// Limit `user_id` to `groups`, or give back every group with `None`.
+#[tauri::command]
+pub async fn set_team_member_access(
+  user_id: String,
+  groups: Option<Vec<String>>,
+) -> Result<Value, String> {
+  let team = team_id().await?;
+  if !is_group_id(user_id.trim()) {
+    return Err(crate::backend_error("TEAM_MEMBER_NOT_FOUND"));
+  }
+  let groups = checked_groups(groups)?;
+  let count = groups.as_ref().map(Vec::len);
+  let url = members_url(
+    &team,
+    &format!("/{}/access", urlencoding::encode(user_id.trim())),
+  );
+  let member = crate::cloud_api::call::<Value>(
+    reqwest::Method::PUT,
+    url,
+    Some(json!({ "groups": groups })),
+    MEMBER_CODES,
+  )
+  .await
+  .map_err(|e| failure("set_access", e))?;
+  match count {
+    Some(count) => log::info!("Team member access set groups={count}"),
+    None => log::info!("Team member access set groups=all"),
+  }
+  Ok(member)
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn access_is_every_group_or_a_clean_list_of_ids() {
+    assert_eq!(checked_groups(None).unwrap(), None);
+    assert_eq!(
+      checked_groups(Some(vec!["g1".into(), " g1 ".into(), UNGROUPED.into()])).unwrap(),
+      Some(vec!["g1".to_string(), UNGROUPED.to_string()])
+    );
+    assert_eq!(checked_groups(Some(vec![])).unwrap(), Some(vec![]));
+    assert!(checked_groups(Some(vec!["a/b".into()])).is_err());
+    assert!(checked_groups(Some(vec!["x".repeat(129)])).is_err());
+    assert!(checked_groups(Some((0..501).map(|i| format!("g{i}")).collect())).is_err());
+  }
+
+  #[test]
+  fn a_member_id_is_one_path_segment() {
+    assert!(members_url("t/1", "/u/access").contains("/api/teams/t%2F1/members/u/access"));
+    assert!(!is_group_id("../x"));
+  }
+}