feat(ecc2): add session heartbeat stale detection

ecc2/src/session/runtime.rs

@@ -5,6 +5,7 @@ use anyhow::{Context, Result};
 use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader};
 use tokio::process::Command;
 use tokio::sync::{mpsc, oneshot};
+use tokio::time::{self, MissedTickBehavior};
 
 use super::output::{OutputStream, SessionOutputStore};
 use super::store::StateStore;
@@ -26,6 +27,9 @@ enum DbMessage {
         line: String,
         ack: oneshot::Sender<DbAck>,
     },
+    TouchHeartbeat {
+        ack: oneshot::Sender<DbAck>,
+    },
 }
 
 #[derive(Clone)]
@@ -53,6 +57,10 @@ impl DbWriter {
             .await
     }
 
+    async fn touch_heartbeat(&self) -> Result<()> {
+        self.send(|ack| DbMessage::TouchHeartbeat { ack }).await
+    }
+
     async fn send<F>(&self, build: F) -> Result<()>
     where
         F: FnOnce(oneshot::Sender<DbAck>) -> DbMessage,
@@ -111,6 +119,17 @@ fn run_db_writer(db_path: PathBuf, session_id: String, mut rx: mpsc::UnboundedRe
                 };
                 let _ = ack.send(result);
             }
+            DbMessage::TouchHeartbeat { ack } => {
+                let result = match opened.as_ref() {
+                    Some(db) => db
+                        .touch_heartbeat(&session_id)
+                        .map_err(|error| error.to_string()),
+                    None => Err(open_error
+                        .clone()
+                        .unwrap_or_else(|| "Failed to open state store".to_string())),
+                };
+                let _ = ack.send(result);
+            }
         }
     }
 }
@@ -120,6 +139,7 @@ pub async fn capture_command_output(
     session_id: String,
     mut command: Command,
     output_store: SessionOutputStore,
+    heartbeat_interval: std::time::Duration,
 ) -> Result<ExitStatus> {
     let db_writer = DbWriter::start(db_path, session_id.clone());
 
@@ -152,6 +172,19 @@ pub async fn capture_command_output(
             .ok_or_else(|| anyhow::anyhow!("Spawned process did not expose a process id"))?;
         db_writer.update_pid(Some(pid)).await?;
         db_writer.update_state(SessionState::Running).await?;
+        db_writer.touch_heartbeat().await?;
+
+        let heartbeat_writer = db_writer.clone();
+        let heartbeat_task = tokio::spawn(async move {
+            let mut ticker = time::interval(heartbeat_interval);
+            ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
+            loop {
+                ticker.tick().await;
+                if heartbeat_writer.touch_heartbeat().await.is_err() {
+                    break;
+                }
+            }
+        });
 
         let stdout_task = tokio::spawn(capture_stream(
             session_id.clone(),
@@ -169,6 +202,8 @@ pub async fn capture_command_output(
         ));
 
         let status = child.wait().await?;
+        heartbeat_task.abort();
+        let _ = heartbeat_task.await;
         stdout_task.await??;
         stderr_task.await??;
 
@@ -244,6 +279,7 @@ mod tests {
             worktree: None,
             created_at: now,
             updated_at: now,
+            last_heartbeat_at: now,
             metrics: SessionMetrics::default(),
         })?;
 
@@ -254,9 +290,14 @@ mod tests {
             .arg("-c")
             .arg("printf 'alpha\\n'; printf 'beta\\n' >&2");
 
-        let status =
-            capture_command_output(db_path.clone(), session_id.clone(), command, output_store)
-                .await?;
+        let status = capture_command_output(
+            db_path.clone(),
+            session_id.clone(),
+            command,
+            output_store,
+            std::time::Duration::from_millis(10),
+        )
+        .await?;
 
         assert!(status.success());
 
@@ -286,4 +327,49 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn capture_command_output_updates_heartbeat_for_quiet_processes() -> Result<()> {
+        let db_path = env::temp_dir().join(format!("ecc2-runtime-heartbeat-{}.db", Uuid::new_v4()));
+        let db = StateStore::open(&db_path)?;
+        let session_id = "session-heartbeat".to_string();
+        let now = Utc::now();
+
+        db.insert_session(&Session {
+            id: session_id.clone(),
+            task: "quiet process".to_string(),
+            agent_type: "test".to_string(),
+            working_dir: env::temp_dir(),
+            state: SessionState::Pending,
+            pid: None,
+            worktree: None,
+            created_at: now,
+            updated_at: now,
+            last_heartbeat_at: now,
+            metrics: SessionMetrics::default(),
+        })?;
+
+        let mut command = Command::new("/bin/sh");
+        command.arg("-c").arg("sleep 0.05");
+
+        let _ = capture_command_output(
+            db_path.clone(),
+            session_id.clone(),
+            command,
+            SessionOutputStore::default(),
+            std::time::Duration::from_millis(10),
+        )
+        .await?;
+
+        let db = StateStore::open(&db_path)?;
+        let session = db
+            .get_session(&session_id)?
+            .expect("session should still exist");
+
+        assert!(session.last_heartbeat_at > now);
+        assert_eq!(session.state, SessionState::Completed);
+
+        let _ = std::fs::remove_file(db_path);
+        Ok(())
+    }
 }