feat(analytics): record every routine run and time in call

apps/web/src/features/channel/Call/CallContext.tsx

@@ -1637,6 +1637,8 @@ function createCallState() {
     }
   }
 
+  // When this client joined its current call, for time-in-call analytics.
+  let joinedAt: number | undefined;
   const lifecycle = createCallLifecycle({
     shouldRequestToken: callSession.shouldRequestToken,
     prepareToken: prepareNativeCallJoin,
@@ -1679,17 +1681,26 @@ function createCallState() {
         });
       }
       fireAndForget(invalidateActiveCallQueries(), 'call-membership');
+      joinedAt = Date.now();
       analytics.track('call_action', {
         action: 'joined',
         channelId: call.channelId,
+        callId: call.callId ?? undefined,
       });
     },
-    onLeft: (channelId) =>
+    onLeft: (channelId) => {
+      const secondsInCall =
+        joinedAt === undefined
+          ? undefined
+          : Math.round((Date.now() - joinedAt) / 1000);
+      joinedAt = undefined;
       analytics.track('call_action', {
         action: 'left',
         channelId,
         leaveReason: 'user_initiated',
-      }),
+        secondsInCall,
+      });
+    },
     reportError: (error) => console.error('call lifecycle failed', error),
   });
   const [lifecycleSnapshot, setLifecycleSnapshot] = createSignal(

apps/web/src/lib/analytics/app-events.ts

@@ -259,6 +259,8 @@ export type AppEvents = {
       | 'screen_share_toggled';
     channelId: string;
     callId?: string;
+    /** On `left`: seconds since this client joined the call. */
+    secondsInCall?: number;
   } & Record<string, unknown>;
   // Mic audio-processing state transitions (noise suppression attach /
   // detach / fallback). `constraintMismatch` flags engines that report

crates/scheduled_action/src/domain/execution.rs

@@ -18,8 +18,8 @@ use crate::domain::models::{
     MAX_ACTION_TIME, ResolvedTaskTarget, ScheduledAction, ScheduledActionUpdate,
 };
 use crate::domain::ports::{
-    RoutineRun, ScheduledActionExecutor, ScheduledActionLiveUpdate, ScheduledActionRepo,
-    ScheduledAgentRunner,
+    FinishedRoutineRun, RoutineRun, RoutineRunObserver, RunEnd, RunTarget, RunTrigger,
+    ScheduledActionExecutor, ScheduledActionLiveUpdate, ScheduledActionRepo, ScheduledAgentRunner,
 };
 
 /// Run identity allocated before any awaited preparation. Runners can cancel
@@ -54,6 +54,7 @@ pub struct InProcessExecutor<Rpo, Live, Runner> {
     live_updates: Arc<Live>,
     runner: Arc<Runner>,
     admission: Arc<dyn AiAdmissionService>,
+    observer: Arc<dyn RoutineRunObserver>,
     tracker: TaskTracker,
     cancellation: CancellationToken,
 }
@@ -65,6 +66,7 @@ impl<Rpo, Live, Runner> Clone for InProcessExecutor<Rpo, Live, Runner> {
             live_updates: self.live_updates.clone(),
             runner: self.runner.clone(),
             admission: self.admission.clone(),
+            observer: self.observer.clone(),
             tracker: self.tracker.clone(),
             cancellation: self.cancellation.clone(),
         }
@@ -91,6 +93,7 @@ where
             runner,
             live_updates,
             admission: Arc::new(DisabledAiAdmissionService),
+            observer: Arc::new(()),
             tracker,
             cancellation,
         }
@@ -103,6 +106,12 @@ where
         self
     }
 
+    /// Tell `observer` about every run that starts and then ends.
+    pub fn with_observer(mut self, observer: Arc<dyn RoutineRunObserver>) -> Self {
+        self.observer = observer;
+        self
+    }
+
     async fn prepare(
         &self,
         action: &ScheduledAction,
@@ -200,6 +209,39 @@ async fn bounded<T>(
     }
 }
 
+fn finished_run(
+    action: &ScheduledAction,
+    trigger: RunTrigger,
+    started_at: DateTime<Utc>,
+    result: &Result<()>,
+) -> Option<FinishedRoutineRun> {
+    let target = serde_json::from_value::<AgentTask>(action.task.clone())
+        .ok()
+        .and_then(|task| {
+            task.resolve_target().ok().map(|target| match target {
+                ResolvedTaskTarget::Model { .. } => RunTarget::Model,
+                ResolvedTaskTarget::Agent { .. } => RunTarget::Agent,
+            })
+        })
+        .unwrap_or(RunTarget::Unknown);
+    let end = match result {
+        Ok(()) => RunEnd::Succeeded,
+        Err(error) if error.is::<ExecutionCancelled>() => RunEnd::Interrupted,
+        Err(error) => match error.downcast_ref::<AiAdmissionError>() {
+            Some(AiAdmissionError::Denied(_)) => RunEnd::AiLimitReached,
+            _ => RunEnd::Failed,
+        },
+    };
+    Some(FinishedRoutineRun {
+        owner: action.owner_user().ok()?.clone(),
+        routine_id: action.id?,
+        trigger,
+        target,
+        end,
+        duration: Utc::now() - started_at,
+    })
+}
+
 fn execution_record(
     action: &ScheduledAction,
     resource: Option<ExecutionResource>,
@@ -283,6 +325,13 @@ where
             if let Err(error) = &result {
                 executor.cancel(&action, &handle, error).await;
             }
+            let trigger = match firing {
+                RoutineRun::Scheduled { .. } => RunTrigger::Scheduled,
+                _ => RunTrigger::Manual,
+            };
+            if let Some(run) = finished_run(&action, trigger, start_time, &result) {
+                executor.observer.run_finished(run);
+            }
             let record = execution_record(&action, handle.resource.clone(), start_time, &result);
             let end_time = record.end_time;
             let bookkeeping = async {
@@ -356,6 +405,10 @@ where
         if let Err(error) = &result {
             self.cancel(&run.action, &handle, error).await;
         }
+        let trigger = RunTrigger::Event(run.run.pending.event.event_name());
+        if let Some(finished) = finished_run(&run.action, trigger, run.started_at, &result) {
+            self.observer.run_finished(finished);
+        }
         let record = execution_record(
             &run.action,
             handle.resource.clone(),

crates/scheduled_action/src/domain/execution/test.rs

@@ -2,7 +2,7 @@ use super::*;
 use crate::domain::event_runs::{
     AuthorizedEventRun, ClaimToken, ConfigurationRevision, PendingEventRun,
 };
-use crate::domain::event_trigger::EventReference;
+use crate::domain::event_trigger::{EventName, EventReference};
 use crate::domain::models::{ActionKind, ExecutionResourceType, RunVisibility};
 use entity_access::domain::models::{EntityAccessReceipt, ViewAccessLevel};
 use macro_user_id::user_id::MacroUserIdStr;
@@ -834,3 +834,113 @@ async fn event_cancellation_is_terminal_interruption() {
     assert_eq!(result.outcome, EventRunOutcome::Interrupted);
     assert_eq!(executor.runner.calls.load(Ordering::SeqCst), 0);
 }
+
+#[derive(Default)]
+struct Observer(Mutex<Vec<FinishedRoutineRun>>);
+impl RoutineRunObserver for Observer {
+    fn run_finished(&self, run: FinishedRoutineRun) {
+        self.0.lock().unwrap().push(run);
+    }
+}
+
+fn observed(
+    executor: InProcessExecutor<Repo, Live, Runner>,
+) -> (InProcessExecutor<Repo, Live, Runner>, Arc<Observer>) {
+    let observer = Arc::new(Observer::default());
+    (executor.with_observer(observer.clone()), observer)
+}
+
+fn ends(observer: &Observer) -> Vec<(RunTrigger, RunTarget, RunEnd)> {
+    observer
+        .0
+        .lock()
+        .unwrap()
+        .iter()
+        .map(|run| (run.trigger, run.target, run.end))
+        .collect()
+}
+
+#[tokio::test]
+async fn due_and_manual_runs_are_reported_once_with_their_trigger() {
+    let mut manual = action();
+    manual.next_run_at = Some(Utc::now() + chrono::Duration::hours(1));
+    for (action, trigger) in [
+        (action(), RunTrigger::Scheduled),
+        (manual, RunTrigger::Manual),
+    ] {
+        let (executor, observer) = observed(executor(Repo::default(), Runner::default()));
+        executor.runner.finish.cancel();
+        executor.execute_action(action.clone()).await.unwrap();
+        drain(&executor).await;
+        assert_eq!(
+            ends(&observer),
+            [(trigger, RunTarget::Model, RunEnd::Succeeded)]
+        );
+        let runs = observer.0.lock().unwrap();
+        assert_eq!(runs[0].owner.as_ref(), USER);
+        assert_eq!(Some(runs[0].routine_id), action.id);
+    }
+}
+
+#[tokio::test]
+async fn failed_and_refused_runs_report_how_they_ended() {
+    let (failing, observer) = observed(executor(
+        Repo::default(),
+        Runner {
+            fail_run: true,
+            ..Runner::default()
+        },
+    ));
+    failing.runner.finish.cancel();
+    failing.execute_action(action()).await.unwrap();
+    drain(&failing).await;
+    assert_eq!(ends(&observer)[0].2, RunEnd::Failed);
+
+    for (error, end) in admission_failures()
+        .into_iter()
+        .zip([RunEnd::AiLimitReached, RunEnd::Failed])
+    {
+        let (executor, observer) = observed(
+            executor(Repo::default(), Runner::default()).with_admission(admission(Err(error))),
+        );
+        let _ = executor.execute_action(action()).await;
+        drain(&executor).await;
+        assert_eq!(
+            ends(&observer),
+            [(RunTrigger::Scheduled, RunTarget::Model, end)]
+        );
+    }
+}
+
+#[tokio::test]
+async fn interrupted_runs_are_reported_as_interrupted() {
+    let (executor, observer) = observed(executor(Repo::default(), Runner::default()));
+    executor.execute_action(action()).await.unwrap();
+    executor.cancellation.cancel();
+    drain(&executor).await;
+    assert_eq!(ends(&observer)[0].2, RunEnd::Interrupted);
+}
+
+#[tokio::test]
+async fn event_runs_report_the_event_that_fired_them() {
+    let (executor, observer) = observed(executor(Repo::default(), Runner::default()));
+    executor.runner.finish.cancel();
+    executor.execute(&event_run(), std::future::pending()).await;
+    assert_eq!(
+        ends(&observer),
+        [(
+            RunTrigger::Event(EventName::DocumentUpdated),
+            RunTarget::Model,
+            RunEnd::Succeeded
+        )]
+    );
+}
+
+#[tokio::test]
+async fn runs_that_never_claim_are_not_reported() {
+    let (executor, observer) = observed(executor(Repo::default(), Runner::default()));
+    *executor.repo.claim.lock().unwrap() = Some(ClaimToken::generate());
+    assert!(executor.execute_action(action()).await.is_err());
+    drain(&executor).await;
+    assert!(observer.0.lock().unwrap().is_empty());
+}

crates/scheduled_action/src/domain/ports.rs

@@ -1,5 +1,5 @@
 use super::event_runs::{AuthorizedEventRun, ClaimToken, ConfigurationRevision};
-use super::event_trigger::EventReference;
+use super::event_trigger::{EventName, EventReference};
 use super::execution::ExecutionHandle;
 use super::models::{
     ActionExecutionRecord, CreateScheduledAction, DispatchEvent, InProgressExecution,
@@ -261,3 +261,57 @@ pub trait ScheduledAgentRunner: Send + Sync + 'static {
 pub trait ScheduledActionLiveUpdate: Send + Sync + 'static {
     fn publish_update(&self, update: ScheduledActionUpdate) -> impl Future<Output = ()> + Send;
 }
+
+/// What started a finished routine run.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RunTrigger {
+    /// One of its schedules came due.
+    Scheduled,
+    /// Its owner started it by hand.
+    Manual,
+    /// An event it watches happened.
+    Event(EventName),
+}
+
+/// Who did a routine run's work.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RunTarget {
+    /// A model, prompted directly.
+    Model,
+    /// One of the owner's agents.
+    Agent,
+    /// The task could not be read.
+    Unknown,
+}
+
+/// How a routine run ended.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RunEnd {
+    Succeeded,
+    Failed,
+    /// Shutdown or the event worker stopped it.
+    Interrupted,
+    /// The owner's AI allowance refused the run before it started.
+    AiLimitReached,
+}
+
+/// A routine run that started and then ended, whatever the outcome.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct FinishedRoutineRun {
+    pub owner: MacroUserIdStr<'static>,
+    pub routine_id: Uuid,
+    pub trigger: RunTrigger,
+    pub target: RunTarget,
+    pub end: RunEnd,
+    pub duration: chrono::Duration,
+}
+
+/// Told about every finished routine run, e.g. for product analytics.
+/// Called on the run's task, so implementations must be quick or spawn.
+pub trait RoutineRunObserver: Send + Sync + 'static {
+    fn run_finished(&self, run: FinishedRoutineRun);
+}
+
+impl RoutineRunObserver for () {
+    fn run_finished(&self, _run: FinishedRoutineRun) {}
+}