feat(conversation): persist plan snapshots and expose them for rehydration

crates/aionui-conversation/src/routes.rs

@@ -122,6 +122,9 @@ pub fn conversation_routes(state: ConversationRouterState) -> Router {
         .route("/api/conversations/{id}/fork", post(fork))
         .route("/api/conversations/{id}/associated", get(associated))
         .route("/api/conversations/{id}/messages", get(list_msg).post(send_msg))
+        // MUST precede the `{messageId}` wildcard below: registered after it,
+        // "latest" would be captured as a message id and 404.
+        .route("/api/conversations/{id}/messages/latest", get(latest_msg))
         .route("/api/conversations/{id}/messages/{messageId}", get(get_msg))
         .route("/api/conversations/{id}/artifacts", get(list_artifacts))
         .route("/api/conversations/{id}/artifacts/{artifactId}", patch(update_artifact))
@@ -266,6 +269,25 @@ struct MessagePathParams {
     message_id: String,
 }
 
+#[derive(serde::Deserialize)]
+struct LatestMessageQuery {
+    r#type: String,
+}
+
+async fn latest_msg(
+    State(state): State<ConversationRouterState>,
+    Extension(user): Extension<CurrentUser>,
+    Path(id): Path<String>,
+    Query(query): Query<LatestMessageQuery>,
+) -> Result<Json<ApiResponse<Option<MessageResponse>>>, ApiError> {
+    let result = state
+        .service
+        .latest_message_of_type(&user.id, &id, &query.r#type)
+        .await
+        .map_err(ApiError::from)?;
+    Ok(Json(ApiResponse::ok(result)))
+}
+
 async fn get_msg(
     State(state): State<ConversationRouterState>,
     Extension(user): Extension<CurrentUser>,
@@ -606,3 +628,24 @@ mod error_mapping_tests {
         assert_eq!(details["port"], 18789);
     }
 }
+
+#[cfg(test)]
+mod route_shape_tests {
+    /// axum builds its route trie eagerly, so an overlapping registration panics
+    /// at construction, not at request time — `cargo check` would never catch it.
+    /// `/messages/latest` deliberately sits before the `{messageId}` wildcard;
+    /// this test is the guard that the pair stays registrable together.
+    #[test]
+    fn latest_message_route_coexists_with_the_message_id_wildcard() {
+        let router: axum::Router<()> = axum::Router::new()
+            .route(
+                "/api/conversations/{id}/messages/latest",
+                axum::routing::get(|| async { "latest" }),
+            )
+            .route(
+                "/api/conversations/{id}/messages/{messageId}",
+                axum::routing::get(|| async { "by-id" }),
+            );
+        let _ = router.into_make_service();
+    }
+}

crates/aionui-conversation/src/service.rs

@@ -3234,6 +3234,32 @@ impl ConversationService {
     }
 
     /// Return one full message for a conversation after verifying ownership.
+    /// Newest message of one type, or `None`.
+    ///
+    /// Serves the plan bar's rehydration: the paginated load alone cannot find a
+    /// plan row that its own turn buried under later messages (`upsert_message`
+    /// does not refresh `created_at`).
+    pub async fn latest_message_of_type(
+        &self,
+        user_id: &str,
+        conversation_id: &str,
+        message_type: &str,
+    ) -> Result<Option<MessageResponse>, ConversationError> {
+        self.conversation_repo
+            .get(user_id, conversation_id)
+            .await?
+            .ok_or_else(|| ConversationError::NotFound {
+                id: conversation_id.to_owned(),
+            })?;
+
+        let row = self
+            .conversation_repo
+            .latest_message_of_type(user_id, conversation_id, message_type)
+            .await?;
+
+        row.map(row_to_message_response).transpose()
+    }
+
     pub async fn get_message(
         &self,
         user_id: &str,

crates/aionui-conversation/src/stream_persistence.rs

@@ -498,6 +498,58 @@ impl StreamPersistenceAdapter {
         }
     }
 
+    /// Persist a plan / to-do snapshot.
+    ///
+    /// One row per turn (`plan:{msg_id}`), upserted: a plan is a
+    /// FULL-REPLACEMENT snapshot (ACP spec — "the Agent MUST send a complete
+    /// list of all plan entries in each update"), so every frame overwrites the
+    /// previous entries instead of stacking rows.
+    ///
+    /// `msg_id` stores the BARE turn msg_id, not the `plan:` form: the live WS
+    /// frame carries the bare id, and the renderer dedupes history against live
+    /// frames on `${type}:${msg_id}`. Storing the prefixed form here would make
+    /// a reloaded conversation show one live card plus one history card.
+    ///
+    /// Gated as `ToolCallPersist` — the same "mid-turn content write" lifecycle
+    /// class; a plan needs no gating rule of its own.
+    #[tracing::instrument(skip_all)]
+    pub async fn persist_plan(
+        &self,
+        data: &aionui_ai_agent::protocol::events::session_updates::PlanEventData,
+        turn_id: &str,
+    ) {
+        if !self.allows_write(RuntimeWriteKind::ToolCallPersist) {
+            return;
+        }
+
+        // `turn_id` rides INSIDE the content JSON: the column set is fixed and
+        // this feature deliberately ships without a migration. The frontend
+        // gates the plan bar on it matching the running turn, so a finished
+        // turn's checklist cannot linger over the next one.
+        let mut value = serde_json::to_value(data).unwrap_or_default();
+        if let Some(obj) = value.as_object_mut() {
+            obj.insert("turn_id".into(), serde_json::Value::String(turn_id.to_owned()));
+        }
+        let content = value.to_string();
+
+        let row = MessageRow {
+            id: format!("plan:{}", self.msg_id),
+            conversation_id: self.conversation_id.clone(),
+            msg_id: Some(self.msg_id.clone()),
+            r#type: "plan".into(),
+            content,
+            position: Some("left".into()),
+            status: Some("finish".into()),
+            hidden: false,
+            created_at: now_ms(),
+            backend_turn_id: self.current_backend_turn_id(),
+        };
+
+        if let Err(e) = self.repo.upsert_message(&self.user_id, &row).await {
+            log_persist_error(&e, "Failed to upsert plan message");
+        }
+    }
+
     /// Persist an ACP (Claude CLI) tool call event.
     #[tracing::instrument(skip_all)]
     pub async fn persist_acp_tool_call(

crates/aionui-conversation/src/stream_relay.rs

@@ -704,6 +704,17 @@ impl StreamRelay {
                             // The raw frame still reaches the frontend via message.stream.
                             self.forward_to_websocket(&event);
                         }
+                        AgentStreamEvent::Plan(data) => {
+                            // A plan is a side-channel SNAPSHOT, not turn work. It
+                            // deliberately does NOT set `saw_tool_or_side_effect` (that
+                            // would make an otherwise-replayable turn look unsafe to
+                            // retry) and does NOT close the active text segment — a plan
+                            // refresh lands mid-reply and would otherwise shatter that
+                            // reply into a fresh bubble, the same reasoning as the
+                            // WorkflowProgress arm above.
+                            self.forward_to_websocket(&event);
+                            self.adapter.persist_plan(data, &self.turn_id).await;
+                        }
                         _ => {
                             self.forward_to_websocket(&event);
                         }
@@ -2533,6 +2544,68 @@ mod tests {
 
     // ── Tool persistence tests ────────────────────────────────────
 
+    /// A plan snapshot must reach the DB, not just the WebSocket: a turn that
+    /// keeps running in the background has to rehydrate its plan bar when the
+    /// user comes back to the conversation.
+    ///
+    /// One row per turn, upserted — a plan is a FULL-REPLACEMENT snapshot, so a
+    /// second frame overwrites the first rather than stacking a second card.
+    #[tokio::test]
+    async fn run_plan_persists_message() {
+        use aionui_ai_agent::protocol::events::session_updates::PlanEventData;
+
+        let repo = Arc::new(RecordingRepo::new());
+        let bus = Arc::new(aionui_realtime::BroadcastEventBus::new(64));
+        let (tx, _) = broadcast::channel(64);
+
+        let relay = StreamRelay::new(
+            "conv-1".into(),
+            "asst-1".into(),
+            "turn-1".into(),
+            "user-1".into(),
+            repo.clone(),
+            bus.clone(),
+        );
+
+        let rx = tx.subscribe();
+
+        tx.send(AgentStreamEvent::Plan(PlanEventData {
+            session_id: None,
+            entries: vec![json!({"content": "step one", "status": "pending"})],
+        }))
+        .unwrap();
+        tx.send(AgentStreamEvent::Plan(PlanEventData {
+            session_id: None,
+            entries: vec![json!({"content": "step one", "status": "completed"})],
+        }))
+        .unwrap();
+        tx.send(AgentStreamEvent::Finish(FinishEventData::default())).unwrap();
+
+        relay.consume(rx).await;
+
+        let inserts = repo.take_inserts();
+        let plans: Vec<_> = inserts.iter().filter(|m| m.r#type == "plan").collect();
+        assert_eq!(plans.len(), 1, "one row per turn, not one per frame: {inserts:?}");
+
+        let row = plans[0];
+        assert_eq!(row.id, "plan:asst-1");
+        // BARE msg_id: the live WS frame carries the turn msg_id, and the
+        // renderer dedupes history against live frames on `${type}:${msg_id}`.
+        assert_eq!(row.msg_id.as_deref(), Some("asst-1"));
+
+        let updates = repo.take_updates();
+        let (_, upd) = updates
+            .iter()
+            .find(|(id, _)| id == "plan:asst-1")
+            .expect("the second frame must upsert the same row");
+
+        let content: serde_json::Value = serde_json::from_str(upd.content.as_deref().unwrap()).unwrap();
+        assert_eq!(content["entries"][0]["status"], "completed");
+        // turn_id rides inside content (the column set is fixed); the plan bar
+        // gates on it matching the running turn.
+        assert_eq!(content["turn_id"], "turn-1");
+    }
+
     #[tokio::test]
     async fn run_tool_call_persists_message() {
         use aionui_ai_agent::protocol::events::tool_call::{ToolCallEventData, ToolCallStatus};

crates/aionui-db/src/repository/conversation.rs

@@ -194,6 +194,26 @@ pub trait IConversationRepository: Send + Sync {
         ))
     }
 
+    /// Newest message of one type in a conversation, or `None`.
+    ///
+    /// Exists for the plan bar: `upsert_message` does not refresh `created_at`,
+    /// so a plan row stays anchored at the start of its turn and a busy turn
+    /// buries it outside the default message page. Deliberately NOT a filter on
+    /// the shared paginator — that has four SQL variants and cursor semantics
+    /// (`has_more_before` / `has_more_after`) that a type filter would muddy.
+    ///
+    /// Default is unsupported so test doubles that never need it can skip it.
+    async fn latest_message_of_type(
+        &self,
+        _user_id: &str,
+        _conversation_id: &str,
+        _message_type: &str,
+    ) -> Result<Option<MessageRow>, DbError> {
+        Err(DbError::Init(
+            "latest_message_of_type is not supported by this repository".into(),
+        ))
+    }
+
     /// Resolves the backend turn anchor for a fork point: the `backend_turn_id`
     /// of the nearest row at or before the `(created_at, id)` cursor that has
     /// one. `Ok(None)` when no row up to the fork point carries an anchor

crates/aionui-db/src/repository/sqlite_conversation.rs

@@ -760,6 +760,31 @@ impl IConversationRepository for SqliteConversationRepository {
 
     // ── Message operations ──────────────────────────────────────────
 
+    async fn latest_message_of_type(
+        &self,
+        user_id: &str,
+        conversation_id: &str,
+        message_type: &str,
+    ) -> Result<Option<MessageRow>, DbError> {
+        self.ensure_conversation_for_user(user_id, conversation_id).await?;
+        // Hits idx_messages_type_created (type, created_at DESC).
+        let row = sqlx::query_as::<_, MessageRow>(
+            "SELECT m.* FROM messages m \
+               INNER JOIN conversations c ON c.id = m.conversation_id \
+               WHERE c.user_id = ? \
+                 AND m.conversation_id = ? \
+                 AND m.type = ? \
+               ORDER BY m.created_at DESC, m.id DESC \
+               LIMIT 1",
+        )
+        .bind(user_id)
+        .bind(conversation_id)
+        .bind(message_type)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row)
+    }
+
     async fn list_messages_page(
         &self,
         user_id: &str,
@@ -2310,6 +2335,49 @@ mod tests {
         assert!(!page1.has_more_after);
     }
 
+    /// The plan bar needs the newest plan row regardless of how many messages the
+    /// turn produced after it: `upsert_message` does not refresh `created_at`, so a
+    /// plan row stays anchored at the START of its turn and a busy turn buries it
+    /// far outside the default 50-message page.
+    #[tokio::test]
+    async fn latest_message_of_type_returns_the_newest_matching_row() {
+        let (repo, _db) = setup().await;
+        let conv = sample_conversation(SYSTEM_USER_ID);
+        repo.create(&conv).await.unwrap();
+
+        for created_at in [100, 300] {
+            let mut msg = sample_message(&conv.id);
+            msg.id = format!("plan-{created_at}");
+            msg.r#type = "plan".to_string();
+            msg.content = format!(r#"{{"entries":[],"turn_id":"turn-{created_at}"}}"#);
+            msg.created_at = created_at;
+            repo.insert_message(&conv.user_id, &msg).await.unwrap();
+        }
+        // Bury the plan rows well past any realistic page size.
+        for i in 0..60 {
+            let mut msg = sample_message(&conv.id);
+            msg.id = aionui_common::generate_prefixed_id("msg");
+            msg.created_at = 400 + i;
+            repo.insert_message(&conv.user_id, &msg).await.unwrap();
+        }
+
+        let found = repo
+            .latest_message_of_type(&conv.user_id, &conv.id, "plan")
+            .await
+            .unwrap()
+            .expect("the newest plan row must be reachable");
+        assert_eq!(found.id, "plan-300");
+        assert_eq!(found.created_at, 300);
+
+        assert!(
+            repo.latest_message_of_type(&conv.user_id, &conv.id, "skill_suggest")
+                .await
+                .unwrap()
+                .is_none(),
+            "a type with no rows must return None, not an error"
+        );
+    }
+
     #[tokio::test]
     async fn before_pages_walk_history_without_duplicates() {
         let (repo, _db) = setup().await;