feat(calendar): relay Google Calendar push to local stacks via dev

.github/services-config.json

@@ -35,6 +35,16 @@
         "upload_extractor_lambda_trigger"
       ]
     },
+    "calendar-event-local-tunnel": {
+      "source_paths": [
+        "services/calendar_event_local_tunnel/**",
+        "crates/calendar_watch_relay/**"
+      ],
+      "stack_path": "infra/stacks/calendar-event-local-tunnel/**",
+      "deploy_binaries": [
+        "calendar_event_local_tunnel"
+      ]
+    },
     "call-recording": {
       "source_paths": [
         "services/call_recording_preview_handler/**"

.github/workflows/deploy-service-generic.yml

@@ -10,6 +10,7 @@ on:
           - agent-schedule-service
           - authentication-service
           - bulk-upload
+          - calendar-event-local-tunnel
           - connection-gateway
           - contacts-service
           - convert-service

.github/workspace-dep-closures.json

@@ -606,6 +606,17 @@
       "crates/model-error-response",
       "crates/workspace-hack"
     ],
+    "calendar_event_local_tunnel": [
+      "crates/calendar_watch_relay",
+      "crates/macro_config",
+      "crates/macro_config_derive",
+      "crates/macro_entrypoint",
+      "crates/macro_env",
+      "crates/macro_env_var",
+      "crates/remote_env_var",
+      "crates/workspace-hack",
+      "services/calendar_event_local_tunnel"
+    ],
     "calendar_events": [
       "crates/ai_toolset",
       "crates/bot_id",
@@ -635,6 +646,11 @@
       "crates/remote_env_var",
       "crates/workspace-hack"
     ],
+    "calendar_watch_relay": [
+      "crates/calendar_watch_relay",
+      "crates/macro_env_var",
+      "crates/workspace-hack"
+    ],
     "call": [
       "crates/activity",
       "crates/agent",
@@ -2294,6 +2310,7 @@
       "crates/authentication_service_client",
       "crates/bot_id",
       "crates/calendar_events",
+      "crates/calendar_watch_relay",
       "crates/channel_sender",
       "crates/connection_gateway_client",
       "crates/connection_gateway_models",

Cargo.toml

@@ -11,6 +11,7 @@ members = [
   "crates/bots",
   "crates/broadcast",
   "crates/calendar_events",
+  "crates/calendar_watch_relay",
   "crates/channel_bots",
   "crates/client/cache-core",
   "crates/client/cache-idb",
@@ -60,6 +61,7 @@ members = [
   "crates/workspace-hack",
   "services/ai_projections_refresh_handler",
   "services/authentication_service",
+  "services/calendar_event_local_tunnel",
   "services/call_recording_preview_handler",
   "services/connection_gateway",
   "services/contacts_service",

crates/calendar_events/src/domain/models.rs

@@ -866,6 +866,32 @@ pub struct GoogleWatchChannel {
     pub expires_at: DateTime<Utc>,
 }
 
+/// Outcome of a best-effort pass stopping every open push channel.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct WatchChannelStopSummary {
+    /// Channels confirmed stopped (or already gone) and cleared.
+    pub stopped: usize,
+    /// Channels not fully torn down this pass — the stop call or its
+    /// bookkeeping cleanup failed. Their bookkeeping is kept so a later pass
+    /// or natural expiry finishes the job.
+    pub failed: usize,
+}
+
+/// An open push notification channel joined to the identity able to stop it.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ActiveWatchChannel {
+    /// Calendar row holding the channel bookkeeping.
+    pub calendar_id: Uuid,
+    /// Client-minted channel identifier.
+    pub channel_id: String,
+    /// Provider-assigned resource identifier.
+    pub resource_id: String,
+    /// Link whose grant opened the channel (also the request-gate key).
+    pub email_link_id: Uuid,
+    /// Refresh-token identity used to mint an access token for the stop call.
+    pub token_identity: CalendarLinkTokenIdentity,
+}
+
 /// How the provider adapter must reconcile one calendar this run.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum GoogleSyncPlan {

crates/calendar_events/src/domain/ports.rs

@@ -7,7 +7,7 @@ use rootcause::Report;
 use uuid::Uuid;
 
 use super::models::{
-    AppliedGoogleGrant, AttendeeResponseStatus, CalendarBackfillClaim,
+    ActiveWatchChannel, AppliedGoogleGrant, AttendeeResponseStatus, CalendarBackfillClaim,
     CalendarBackfillFailureDisposition, CalendarBackfillFailureOutcome, CalendarBackfillJobKey,
     CalendarCreationTarget, CalendarEvent, CalendarEventDraft, CalendarEventMutationTarget,
     CalendarEventPatch, CalendarEventUpsert, CalendarLinkTokenIdentity, CalendarOccurrence,
@@ -483,6 +483,35 @@ pub trait GoogleCalendarProvider: Send + Sync + 'static {
     ) -> impl Future<Output = Result<GoogleWatchChannel, GoogleProviderError>> + Send;
 }
 
+/// Provider-side teardown of push notification channels, kept separate from
+/// [`GoogleCalendarProvider`] because only explicit channel teardown needs it.
+pub trait GoogleWatchChannelStopper: Send + Sync + 'static {
+    /// Stop one push channel. An already-gone channel counts as success.
+    fn stop_watch_channel(
+        &self,
+        access_token: &str,
+        email_link_id: Uuid,
+        channel_id: &str,
+        resource_id: &str,
+    ) -> impl Future<Output = Result<(), GoogleProviderError>> + Send;
+}
+
+/// Bookkeeping needed to stop every open push channel at teardown.
+pub trait WatchChannelTeardownRepository: Send + Sync + 'static {
+    /// List every unexpired push channel with the identity that can stop it.
+    fn list_active_watch_channels(
+        &self,
+    ) -> impl Future<Output = Result<Vec<ActiveWatchChannel>, Report>> + Send;
+
+    /// Clear one channel's bookkeeping, guarded by channel id so a
+    /// concurrently reopened channel is never clobbered.
+    fn clear_watch_channel(
+        &self,
+        calendar_id: Uuid,
+        channel_id: &str,
+    ) -> impl Future<Output = Result<(), Report>> + Send;
+}
+
 /// Durable scheduling operations for periodic provider maintenance.
 pub trait GoogleCalendarSyncRepository: Send + Sync + 'static {
     /// Reset completed current-grant jobs that are due for another incremental poll.