refactor(session): connection-scoped turn-intent helpers for the lifecycle stack

src-tauri/crates/session-persistence/src/turn_intents.rs

@@ -29,6 +29,7 @@
 use chrono::Utc;
 use rusqlite::{params, Connection, OptionalExtension, Result as SqliteResult};
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 
 use super::connection::get_connection;
 
@@ -284,6 +285,28 @@ pub fn upsert_initial(
     status: TurnIntentStatus,
 ) -> Result<TurnIntentRow, IntentError> {
     let conn = get_connection()?;
+    upsert_initial_on(
+        &conn,
+        session_id,
+        turn_intent_id,
+        client_message_id,
+        org_run_id,
+        source,
+        status,
+    )
+}
+
+/// Connection-scoped variant for callers that atomically update adjacent
+/// session lifecycle state in the same SQLite transaction.
+pub fn upsert_initial_on(
+    conn: &Connection,
+    session_id: &str,
+    turn_intent_id: &str,
+    client_message_id: Option<&str>,
+    org_run_id: Option<&str>,
+    source: TurnIntentSource,
+    status: TurnIntentStatus,
+) -> Result<TurnIntentRow, IntentError> {
     let now = Utc::now().to_rfc3339();
     let inserted = conn.execute(
         "INSERT OR IGNORE INTO session_turn_intents
@@ -347,7 +370,17 @@ pub fn update_status(
     new_status: TurnIntentStatus,
 ) -> Result<TurnIntentRow, IntentError> {
     let conn = get_connection()?;
-    let existing = get_intent(&conn, session_id, turn_intent_id)?
+    update_status_on(&conn, session_id, turn_intent_id, new_status)
+}
+
+/// Connection-scoped variant for atomic session + turn-intent transitions.
+pub fn update_status_on(
+    conn: &Connection,
+    session_id: &str,
+    turn_intent_id: &str,
+    new_status: TurnIntentStatus,
+) -> Result<TurnIntentRow, IntentError> {
+    let existing = get_intent(conn, session_id, turn_intent_id)?
         .ok_or_else(|| IntentError::NotFound(turn_intent_id.to_string(), session_id.to_string()))?;
     if !transition_allowed(existing.status, new_status) {
         return Err(IntentError::IllegalTransition {
@@ -443,6 +476,28 @@ pub fn list_for_session(session_id: &str) -> SqliteResult<Vec<TurnIntentRow>> {
     Ok(rows)
 }
 
+/// Latest lifecycle row for each requested session, read through one
+/// connection so reconnect/focus reconciliation does not fan out into one
+/// SQLite task per session.
+pub fn latest_for_sessions(session_ids: &[String]) -> SqliteResult<HashMap<String, TurnIntentRow>> {
+    let conn = get_connection()?;
+    let mut stmt = conn.prepare_cached(
+        "SELECT session_id, turn_intent_id, client_message_id, org_run_id,
+                source, status, created_at, updated_at
+           FROM session_turn_intents
+          WHERE session_id = ?1
+          ORDER BY updated_at DESC, created_at DESC
+          LIMIT 1",
+    )?;
+    let mut latest = HashMap::with_capacity(session_ids.len());
+    for session_id in session_ids {
+        if let Some(row) = stmt.query_row([session_id], row_from_sql).optional()? {
+            latest.insert(session_id.clone(), row);
+        }
+    }
+    Ok(latest)
+}
+
 // ============================================
 // Tests
 // ============================================
@@ -730,6 +785,36 @@ mod tests {
         });
     }
 
+    #[test]
+    fn latest_for_sessions_returns_one_current_intent_per_requested_session() {
+        with_temp_orgii_home(|| {
+            let first_session = "test-session-batch-a";
+            let second_session = "test-session-batch-b";
+            let _ = fresh_intent(first_session, "intent-a-old");
+            update_status(first_session, "intent-a-old", TurnIntentStatus::Running).unwrap();
+            update_status(first_session, "intent-a-old", TurnIntentStatus::Completed).unwrap();
+            let _ = fresh_intent(first_session, "intent-a-current");
+            update_status(first_session, "intent-a-current", TurnIntentStatus::Running).unwrap();
+            let _ = fresh_intent(second_session, "intent-b-current");
+
+            let rows = latest_for_sessions(&[
+                first_session.to_string(),
+                second_session.to_string(),
+                "missing-session".to_string(),
+            ])
+            .expect("batch lifecycle lookup succeeds");
+
+            assert_eq!(rows.len(), 2);
+            assert_eq!(rows[first_session].turn_intent_id, "intent-a-current");
+            assert_eq!(rows[first_session].status, TurnIntentStatus::Running);
+            assert_eq!(rows[second_session].turn_intent_id, "intent-b-current");
+            assert_eq!(
+                rows[second_session].org_run_id, None,
+                "batch projection must preserve the complete row shape"
+            );
+        });
+    }
+
     #[test]
     fn restart_reconciliation_closes_every_in_flight_intent() {
         with_temp_orgii_home(|| {