Apply live session titles as soon as the agent sends them

src-tauri/src/acp/connection.rs

@@ -11091,9 +11091,43 @@ async fn emit_conversation_update(
             // `normalize_goal_status` unchanged; its extra fields
             // (createdAt/updatedAt/iterations/lastReason/controlMethod)
             // survive inside the marker's raw goal object for the card.
-            // (`info.title` is Codex's native thread name; it is adopted via the
-            // parser auto-title path on the next conversation fetch, not here, to
-            // keep this DB-agnostic emit path unchanged — see parsers/codex.rs.)
+            // `info.title` is the agent's live session name (Codex thread name,
+            // Claude ACP 0.69+ generated titles, anyone else who publishes the
+            // field). Apply it immediately via a dedicated lifecycle event
+            // rather than waiting for the next conversation fetch. Goal-only
+            // updates leave title undefined and emit nothing. Identical repeats
+            // are skipped (CodeBuddy resends its fallback after every turn).
+            // A title that arrives before the row is bound is dropped, not
+            // remembered, so a later resend is still accepted. If it never
+            // comes back, the next detail load recovers it only for agents
+            // whose own transcript carries the name (Codex's session index,
+            // Claude's `ai-title`) — a custom ACP agent's is gone, because
+            // `parsers/acp_native.rs` records no `session_info_update` and can
+            // only ever title a session by its first prompt.
+            if let Some(title) = crate::acp::session_title::native_title_from_session_info(
+                info.title.value().map(|s| s.as_str()),
+            ) {
+                // Test and set under ONE write lock. Nothing can interleave
+                // here today — a session's notifications are handled serially,
+                // and the only other writer of `last_native_title` is the
+                // `ConversationLinked` arm, which is emitted ONLY while the row
+                // is still unbound and therefore can never race a title this
+                // admits. That safety currently rests on two guards in
+                // different files agreeing; keeping the halves in one critical
+                // section makes it hold by construction instead.
+                let admit = {
+                    let mut s = state.write().await;
+                    let admit = s.conversation_id.is_some()
+                        && s.last_native_title.as_deref() != Some(title.as_str());
+                    if admit {
+                        s.last_native_title = Some(title.clone());
+                    }
+                    admit
+                };
+                if admit {
+                    emit_with_state(state, emitter, AcpEvent::NativeSessionTitle { title }).await;
+                }
+            }
             let neutral_goal_channel = state.read().await.neutral_goal_channel;
             if let Some(goal) =
                 session_info_goal_value(neutral_goal_channel, info.meta.as_ref())
@@ -11759,6 +11793,145 @@ mod tests {
         assert!(session_info_goal_value(true, None).is_none());
     }
 
+    // --- live ACP session title (`session_info_update.title`) --------------
+
+    /// Drive one `session_info_update` carrying `title` through
+    /// `emit_conversation_update`.
+    async fn drive_session_info_title(state: &Arc<RwLock<SessionState>>, title: &str) {
+        let update: SessionUpdate = serde_json::from_value(serde_json::json!({
+            "sessionUpdate": "session_info_update",
+            "title": title,
+        }))
+        .expect("valid session_info_update wire shape");
+        let mut cache = ToolCallOutputCache::default();
+        let mut cb = CodeBuddyLiveState::default();
+        emit_conversation_update(
+            state,
+            &EventEmitter::Noop,
+            AgentType::CodeBuddy,
+            update,
+            None,
+            &mut cache,
+            &mut cb,
+        )
+        .await;
+    }
+
+    /// Titles emitted on this connection so far, oldest first.
+    async fn emitted_native_titles(state: &Arc<RwLock<SessionState>>) -> Vec<String> {
+        state
+            .read()
+            .await
+            .recent_events_after(0)
+            .unwrap_or_default()
+            .iter()
+            .filter_map(|e| match &e.payload {
+                AcpEvent::NativeSessionTitle { title } => Some(title.clone()),
+                _ => None,
+            })
+            .collect()
+    }
+
+    fn title_test_state(conversation_id: Option<i32>) -> Arc<RwLock<SessionState>> {
+        let mut st = SessionState::new(
+            "conn-title".to_string(),
+            AgentType::CodeBuddy,
+            None,
+            "win".to_string(),
+            None,
+        );
+        st.conversation_id = conversation_id;
+        Arc::new(RwLock::new(st))
+    }
+
+    /// A changed title emits; the SAME title arriving again does not. CodeBuddy
+    /// (`sendPendingTitleUpdate`) resends its 80-code-unit fallback after every
+    /// completed prompt with no last-sent guard of its own, and that string
+    /// differs from the 100-char title our own parser derives from the same
+    /// first message — so without this skip the sidebar name would flip on
+    /// every turn (live write, then session-file parse, repeat).
+    #[tokio::test]
+    async fn session_info_title_emits_once_and_skips_an_identical_repeat() {
+        let state = title_test_state(Some(7));
+
+        drive_session_info_title(&state, "Fix the login flow").await;
+        drive_session_info_title(&state, "Fix the login flow").await;
+        drive_session_info_title(&state, "  Fix the login flow  ").await; // same after trim
+        drive_session_info_title(&state, "Fix the signup flow").await;
+
+        assert_eq!(
+            emitted_native_titles(&state).await,
+            vec![
+                "Fix the login flow".to_string(),
+                "Fix the signup flow".to_string()
+            ],
+            "only a CHANGED title may reach the lifecycle worker"
+        );
+    }
+
+    /// A title published before the first prompt binds the row has nowhere to
+    /// land, so it is dropped — and deliberately NOT remembered, so the resend
+    /// that follows `ConversationLinked` is still accepted. Guards the
+    /// `conversation_id.is_some()` half of the skip: a stale cache here would
+    /// leave the row "Untitled" for the rest of the connection.
+    #[tokio::test]
+    async fn session_info_title_dropped_while_unbound_is_accepted_after_link() {
+        let state = title_test_state(None);
+
+        drive_session_info_title(&state, "Fix the login flow").await;
+        assert!(
+            emitted_native_titles(&state).await.is_empty(),
+            "no row to write to yet"
+        );
+        assert!(
+            state.read().await.last_native_title.is_none(),
+            "a dropped title must not poison the skip-cache"
+        );
+
+        state.write().await.apply_event(&AcpEvent::ConversationLinked {
+            conversation_id: 7,
+            folder_id: 1,
+            parent_conversation_id: None,
+            parent_tool_use_id: None,
+        });
+
+        drive_session_info_title(&state, "Fix the login flow").await;
+        assert_eq!(
+            emitted_native_titles(&state).await,
+            vec!["Fix the login flow".to_string()],
+            "the same title must be accepted once the row exists"
+        );
+    }
+
+    /// Goal-only / metadata-only `session_info_update`s (the common case for
+    /// codex `/goal` transitions) carry no title and must not queue the
+    /// lifecycle worker.
+    #[tokio::test]
+    async fn session_info_without_a_title_emits_no_native_title() {
+        let state = title_test_state(Some(7));
+        for wire in [
+            serde_json::json!({"sessionUpdate": "session_info_update"}),
+            serde_json::json!({"sessionUpdate": "session_info_update", "title": null}),
+            serde_json::json!({"sessionUpdate": "session_info_update", "title": "   "}),
+        ] {
+            let update: SessionUpdate =
+                serde_json::from_value(wire).expect("valid session_info_update wire shape");
+            let mut cache = ToolCallOutputCache::default();
+            let mut cb = CodeBuddyLiveState::default();
+            emit_conversation_update(
+                &state,
+                &EventEmitter::Noop,
+                AgentType::Codex,
+                update,
+                None,
+                &mut cache,
+                &mut cb,
+            )
+            .await;
+        }
+        assert!(emitted_native_titles(&state).await.is_empty());
+    }
+
     #[test]
     fn goal_advertised_control_reads_method_and_actions_or_falls_back() {
         // claude 0.66+ / codex 1.2+ advertisements.

src-tauri/src/acp/lifecycle.rs

@@ -35,7 +35,7 @@ use tokio::sync::RwLock;
 /// Per-connection worker queue depth. Sized for the **filtered** event set
 /// only (see `is_lifecycle_relevant`) — high-frequency events (ContentDelta,
 /// ToolCall*, PermissionRequest) are dropped at the dispatcher and never
-/// enter the queue. The remaining 5 event types arrive at most a handful
+/// enter the queue. The remaining 6 event types arrive at most a handful
 /// of times per turn, so 64 slots is comfortable headroom for a sustained
 /// SQLite stall without forcing the dispatcher to block on `send`.
 const WORKER_QUEUE_CAPACITY: usize = 64;
@@ -65,6 +65,7 @@ fn is_lifecycle_relevant(event: &AcpEvent) -> bool {
         AcpEvent::SessionStarted { .. }
             | AcpEvent::TurnComplete { .. }
             | AcpEvent::ConversationLinked { .. }
+            | AcpEvent::NativeSessionTitle { .. }
             | AcpEvent::StatusChanged {
                 status: ConnectionStatus::Disconnected
             }
@@ -309,6 +310,43 @@ pub(crate) async fn handle_event(
             }
             Ok(())
         }
+        AcpEvent::NativeSessionTitle { title } => {
+            // Live ACP session title: write it onto the bound row the moment
+            // the agent publishes it. `refresh_auto_title` is a no-op when the
+            // user renamed the chat, when the value is unchanged, or when the
+            // string is empty — so repeating the same title across goal
+            // snapshots does not churn the sidebar.
+            let Some((state_arc, emitter)) =
+                manager.get_state_and_emitter(&envelope.connection_id).await
+            else {
+                return Ok(());
+            };
+            let conversation_id = { state_arc.read().await.conversation_id };
+            // Title published before the first prompt binds the row: drop it.
+            // `emit_conversation_update` never caches an unbound title, so a
+            // later resend is still accepted. Failing that, the next detail
+            // load recovers it only where the agent's own transcript carries
+            // the name — see the note at that emit site.
+            let Some(cid) = conversation_id else {
+                return Ok(());
+            };
+            // Unlocked on purpose: a later ACP title or a session-file parse
+            // can still replace this. Identical repeats are filtered at emit
+            // time so CodeBuddy's per-turn fallback cannot ping-pong the
+            // sidebar. Soft-deleted rows are skipped by the UPDATE predicate.
+            if conversation_service::refresh_auto_title(db_conn, cid, title.clone()).await? {
+                crate::commands::conversations::emit_conversation_upsert(&emitter, db_conn, cid)
+                    .await;
+                if let Some(ccm) = manager.chat_channel() {
+                    crate::commands::conversations::spawn_sync_conversation_title_until_current(
+                        db_conn.clone(),
+                        ccm,
+                        cid,
+                    );
+                }
+            }
+            Ok(())
+        }
         // Other events don't need cross-connection DB persistence today; extend
         // this dispatcher with new arms as the lifecycle scope grows.
         _ => Ok(()),
@@ -1525,7 +1563,7 @@ async fn connection_worker_loop(
 /// connections, workers run independently so a slow SQLite write on one
 /// connection doesn't backpressure the others.
 ///
-/// All forwarded events (the 5 types in `is_lifecycle_relevant`) use
+/// All forwarded events (the 6 types in `is_lifecycle_relevant`) use
 /// blocking `send().await` to guarantee delivery even when the worker
 /// mailbox is full — `SessionStarted` (writes external_id) and
 /// `TurnComplete` (writes terminal status) are correctness-critical and
@@ -1790,6 +1828,100 @@ mod tests {
         assert_eq!(p["summary"]["external_id"], "ext-99");
     }
 
+    #[tokio::test]
+    async fn handle_event_native_session_title_writes_and_upserts() {
+        use crate::web::event_bridge::{WebEventBroadcaster, CONVERSATION_CHANGED_EVENT};
+        let db = test_helpers::fresh_in_memory_db().await;
+        let folder_id = test_helpers::seed_folder(&db, "/tmp/test-native-title").await;
+        let conv =
+            conversation_service::create(&db.conn, folder_id, AgentType::ClaudeCode, None, None)
+                .await
+                .unwrap();
+        let broadcaster = Arc::new(WebEventBroadcaster::new());
+        let mut rx = broadcaster.subscribe();
+        let mgr = ConnectionManager::new();
+        {
+            let mut map = mgr.connections.lock().await;
+            let mut conn = fake_connection_with_state("c1", Some(conv.id));
+            conn.emitter = EventEmitter::test_web_only(broadcaster.clone());
+            map.insert("c1".to_string(), conn);
+        }
+        let env = EventEnvelope {
+            seq: 1,
+            connection_id: "c1".to_string(),
+            payload: AcpEvent::NativeSessionTitle {
+                title: "  Fix login flow  ".into(),
+            },
+        };
+        handle_event(&db.conn, &mgr, &env, None).await.unwrap();
+
+        let reloaded = conversation_service::get_by_id(&db.conn, conv.id)
+            .await
+            .unwrap();
+        assert_eq!(reloaded.title.as_deref(), Some("Fix login flow"));
+        assert!(!reloaded.title_locked);
+
+        let evt = rx
+            .try_recv()
+            .expect("a written native title should broadcast a conversation upsert");
+        assert_eq!(evt.channel, CONVERSATION_CHANGED_EVENT);
+        let p = &*evt.payload;
+        assert_eq!(p["kind"], "upsert");
+        assert_eq!(p["summary"]["title"], "Fix login flow");
+    }
+
+    #[tokio::test]
+    async fn handle_event_native_session_title_skips_locked_and_unbound() {
+        let db = test_helpers::fresh_in_memory_db().await;
+        let folder_id = test_helpers::seed_folder(&db, "/tmp/test-native-title-skip").await;
+        let conv =
+            conversation_service::create(&db.conn, folder_id, AgentType::ClaudeCode, None, None)
+                .await
+                .unwrap();
+        conversation_service::update_title(&db.conn, conv.id, "User pick".into())
+            .await
+            .unwrap();
+
+        let mgr = ConnectionManager::new();
+        {
+            let mut map = mgr.connections.lock().await;
+            map.insert(
+                "locked".to_string(),
+                fake_connection_with_state("locked", Some(conv.id)),
+            );
+            map.insert(
+                "unbound".to_string(),
+                fake_connection_with_state("unbound", None),
+            );
+        }
+        let locked_env = EventEnvelope {
+            seq: 1,
+            connection_id: "locked".to_string(),
+            payload: AcpEvent::NativeSessionTitle {
+                title: "agent title".into(),
+            },
+        };
+        handle_event(&db.conn, &mgr, &locked_env, None)
+            .await
+            .unwrap();
+        let unbound_env = EventEnvelope {
+            seq: 2,
+            connection_id: "unbound".to_string(),
+            payload: AcpEvent::NativeSessionTitle {
+                title: "should not land anywhere".into(),
+            },
+        };
+        handle_event(&db.conn, &mgr, &unbound_env, None)
+            .await
+            .unwrap();
+
+        let reloaded = conversation_service::get_by_id(&db.conn, conv.id)
+            .await
+            .unwrap();
+        assert_eq!(reloaded.title.as_deref(), Some("User pick"));
+        assert!(reloaded.title_locked);
+    }
+
     #[tokio::test]
     async fn handle_event_session_started_skips_soft_deleted_conversation() {
         // A fork emits `SessionStarted{S2}`. If the bound conversation was
@@ -2353,6 +2485,9 @@ mod tests {
             parent_conversation_id: None,
             parent_tool_use_id: None,
         }));
+        assert!(is_lifecycle_relevant(&AcpEvent::NativeSessionTitle {
+            title: "Fix login".into(),
+        }));
         assert!(is_lifecycle_relevant(&AcpEvent::StatusChanged {
             status: ConnectionStatus::Disconnected,
         }));