fix(telemetry): close recovered task lifecycles

app/src-tauri/src/lib.rs

@@ -152,9 +152,34 @@ pub fn run() {
             let handle = app.handle().clone();
             let media_handle = handle.clone();
             tauri::async_runtime::spawn(async move {
-                // A graceful app shutdown marks active tasks interrupted in
-                // tasks.json before this process starts. If its trace drop
-                // guard finished during shutdown, recover and upload it now.
+                // Startup recovery marks tasks left running by the previous
+                // process as interrupted. Their start telemetry already
+                // exists, so close each lifecycle once before uploading any
+                // trace that became ready during shutdown.
+                for snapshot in tasks.take_recovered_interrupted_tasks().await {
+                    let interruption_reason = snapshot
+                        .error
+                        .clone()
+                        .unwrap_or_else(|| "app was closed before this task finished".into());
+                    telemetry.capture(
+                        "socai_agent_task_end",
+                        json!({
+                            "task_id": snapshot.task_id,
+                            "provider": snapshot.provider,
+                            "run_id": snapshot.run_id,
+                            "model": snapshot.model,
+                            "outcome": "interrupted",
+                            "steps": snapshot.steps,
+                            "input_tokens": snapshot.input_tokens,
+                            "output_tokens": snapshot.output_tokens,
+                            "points_used": snapshot.points_used,
+                            "duration_ms": duration_ms(snapshot.started_at, snapshot.finished_at),
+                            "error": crate::telemetry::short_error(&interruption_reason),
+                            "recovered_on_startup": true,
+                        }),
+                    );
+                }
+
                 for snapshot in tasks.list().await {
                     if snapshot.status == "interrupted" {
                         if let Some(run_dir) = snapshot.run_dir.as_deref() {

app/src-tauri/src/tasks.rs

@@ -26,6 +26,7 @@ struct AgentTaskRegistryInner {
     abort_handles: HashMap<String, AbortHandle>,
     timeline_next_seq: HashMap<String, u64>,
     timeline_locks: HashMap<String, Arc<Mutex<()>>>,
+    recovered_interrupted_tasks: Vec<AgentTaskSnapshot>,
 }
 
 #[derive(serde::Serialize, serde::Deserialize, Clone)]
@@ -66,8 +67,10 @@ impl Default for AgentTaskRegistry {
     fn default() -> Self {
         let mut tasks = load_task_index();
         let interrupted_at = now_ms();
+        let mut recovered_interrupted_tasks = Vec::new();
         for task in &mut tasks {
             if matches!(task.status.as_str(), "queued" | "running") {
+                let started = task.started_at.is_some();
                 task.status = "interrupted".into();
                 task.finished_at = Some(interrupted_at);
                 task.error = Some("app was closed before this task finished".into());
@@ -96,6 +99,9 @@ impl Default for AgentTaskRegistry {
                         );
                     }
                 }
+                if started {
+                    recovered_interrupted_tasks.push(task.clone());
+                }
             }
         }
         let next_seq = tasks.len() as u64;
@@ -109,6 +115,7 @@ impl Default for AgentTaskRegistry {
                 abort_handles: HashMap::new(),
                 timeline_next_seq: HashMap::new(),
                 timeline_locks: HashMap::new(),
+                recovered_interrupted_tasks,
             })),
             runner_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_AGENT_TASKS)),
         }
@@ -127,6 +134,17 @@ fn persisted_run_task(run_dir: &str) -> Option<String> {
 }
 
 impl AgentTaskRegistry {
+    /// Return running tasks recovered as interrupted during startup exactly
+    /// once. Their start event was already captured by the previous process;
+    /// the new process uses this queue to close that telemetry lifecycle.
+    pub(crate) async fn take_recovered_interrupted_tasks(&self) -> Vec<AgentTaskSnapshot> {
+        let recovered = {
+            let mut guard = self.inner.lock().await;
+            std::mem::take(&mut guard.recovered_interrupted_tasks)
+        };
+        recovered.into_iter().map(hydrate_task_snapshot).collect()
+    }
+
     pub(crate) async fn create(
         &self,
         task: String,

docs/telemetry-schema.md

@@ -189,7 +189,7 @@ and model in use are captured on `socai_agent_task_start`.
 | `socai_subscription_checkout` | A WeChat Pay or Alipay order is created or fails | `provider`, `plan_id`, `outcome`, `order_id`, `amount_fen`, `points`, `duration_days`, `error` |
 | `socai_subscription_paid` | Polling first observes a paid subscription order | `order_id`, `amount_fen`, `added_points`, `duration_days`, `pro_active_until` |
 | `socai_agent_task_start` | A task begins running | `task_id`, `provider`, `model`, `task_len`, `task_text` |
-| `socai_agent_task_end` | A task reaches a terminal state | `task_id`, `run_id`, `provider`, `model`, `outcome`, `steps`, token/cache usage, estimated cost breakdown, authoritative `points_used` when settlement completes, `duration_ms`, `error` |
+| `socai_agent_task_end` | A task reaches a terminal state, including a previously running task recovered as interrupted on the next startup | `task_id`, `run_id`, `provider`, `model`, `outcome`, `steps`, token/cache usage, estimated cost breakdown, authoritative `points_used` when settlement completes, `duration_ms`, `error`, and `recovered_on_startup` for a recovered lifecycle |
 | `socai_tool_call` | Each tool call completes | `task_id`, `run_id`, `tool_name`, `turn`, `sequence`, `duration_ms`, `ok`, `error`, query/result summaries, and bounded unexpected-page diagnostics when present |
 | `socai_feishu_export` | A Feishu export completes/fails, including user-visible setup failures before the native export command starts | `task_id`, `run_id`, `destination`, optional `stage`, `outcome`, `duration_ms`, `error`; chat sends also include privacy-safe CLI failure metadata (`cli_exit_code`, `cli_error_type`, `cli_error_subtype`, `cli_error_code`, `cli_log_id`, `cli_update_available`) and `message_id_present` |
 | `socai_server_payment_callback` | The backend accepts, rejects, or fails a merchant callback | `provider`, `stage`, `outcome`, `order_id`, `amount_fen`, `added_points`, `duration_days`, `error` |
@@ -226,6 +226,7 @@ profile, document/chat ID, URL, or credential is reported.
 | `cost_pricing_source` | string | Model catalog pricing provenance. |
 | `task_len` | number | Agent prompt length in Unicode scalar values. |
 | `task_text` | string | Full agent prompt. Always sent on desktop; see privacy boundaries. |
+| `recovered_on_startup` | boolean | Present and true when startup recovery emits the terminal event for a task left running by the previous process. |
 | `turn` / `sequence` | number | Position of a tool call within the run. |
 | `destination` | string | Feishu export target: `document`, `chat`, or `setup` for failures before a destination can be used. |
 | `stage` | string | Feishu operation stage, such as `load_accounts`, `connect_account`, `prepare_document`, `export_document`, `load_chats`, or `send_chat`. |