feat: added _meta.conversationBefore for ACP forking session (#9821)

crates/goose/src/acp/server/fork_session.rs

@@ -8,6 +8,7 @@ impl GooseAcpAgent {
         args: ForkSessionRequest,
     ) -> Result<ForkSessionResponse, agent_client_protocol::Error> {
         validate_absolute_cwd(&args.cwd)?;
+        let conversation_before = conversation_before_from_meta(args.meta.as_ref())?;
         let source_session_id = &*args.session_id.0;
 
         let source = self
@@ -28,7 +29,14 @@ impl GooseAcpAgent {
             .internal_err()?;
         let new_session_id = new_session.id.clone();
 
-        let goose_session = self
+        if let Some(conversation_before) = conversation_before {
+            self.session_manager
+                .truncate_conversation(&new_session_id, conversation_before)
+                .await
+                .internal_err()?;
+        }
+
+        let new_session = self
             .session_manager
             .get_session(&new_session_id, false)
             .await
@@ -36,7 +44,7 @@ impl GooseAcpAgent {
 
         let goose_session = self
             .prepare_session_for_activation(
-                goose_session,
+                new_session.clone(),
                 args.cwd.clone(),
                 args.mcp_servers,
                 false,
@@ -74,3 +82,70 @@ impl GooseAcpAgent {
         Ok(response)
     }
 }
+
+fn conversation_before_from_meta(
+    meta: Option<&Meta>,
+) -> Result<Option<i64>, agent_client_protocol::Error> {
+    let Some(value) = meta.and_then(|meta| meta.get("conversationBefore")) else {
+        return Ok(None);
+    };
+    if value.is_null() {
+        return Ok(None);
+    }
+
+    value.as_i64().map(Some).ok_or_else(|| {
+        agent_client_protocol::Error::invalid_params()
+            .data("conversationBefore must be an integer timestamp")
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn meta_with_conversation_before(value: serde_json::Value) -> Meta {
+        let mut meta = Meta::new();
+        meta.insert("conversationBefore".to_string(), value);
+        meta
+    }
+
+    #[test]
+    fn conversation_before_from_meta_returns_none_when_absent() {
+        assert_eq!(conversation_before_from_meta(None).unwrap(), None);
+        assert_eq!(
+            conversation_before_from_meta(Some(&Meta::new())).unwrap(),
+            None
+        );
+    }
+
+    #[test]
+    fn conversation_before_from_meta_treats_null_as_absent() {
+        let meta = meta_with_conversation_before(serde_json::Value::Null);
+
+        assert_eq!(conversation_before_from_meta(Some(&meta)).unwrap(), None);
+    }
+
+    #[test]
+    fn conversation_before_from_meta_reads_integer_timestamp() {
+        let meta = meta_with_conversation_before(serde_json::json!(1_718_000_000));
+
+        assert_eq!(
+            conversation_before_from_meta(Some(&meta)).unwrap(),
+            Some(1_718_000_000)
+        );
+    }
+
+    #[test]
+    fn conversation_before_from_meta_rejects_non_integer_timestamp() {
+        for value in [
+            serde_json::json!("1718000000"),
+            serde_json::json!(1718000000.5),
+            serde_json::json!(true),
+            serde_json::json!({ "created": 1718000000 }),
+        ] {
+            assert!(
+                conversation_before_from_meta(Some(&meta_with_conversation_before(value))).is_err()
+            );
+        }
+    }
+}

crates/goose/tests/acp_fork_session_test.rs

@@ -0,0 +1,129 @@
+#[allow(dead_code)]
+#[path = "acp_common_tests/mod.rs"]
+mod common_tests;
+
+use agent_client_protocol::schema::{ForkSessionRequest, ForkSessionResponse, SessionId};
+use common_tests::fixtures::server::AcpServerConnection;
+use common_tests::fixtures::{run_test, Connection, OpenAiFixture, TestConnectionConfig};
+use goose::config::GooseMode;
+use goose::conversation::message::{Message, MessageContent};
+use goose::session::{SessionManager, SessionType};
+use std::path::Path;
+
+async fn new_connection(data_root: &Path) -> AcpServerConnection {
+    let openai = OpenAiFixture::new(
+        vec![],
+        <AcpServerConnection as Connection>::expected_session_id(),
+    )
+    .await;
+    <AcpServerConnection as Connection>::new(
+        TestConnectionConfig {
+            data_root: data_root.to_path_buf(),
+            ..Default::default()
+        },
+        openai,
+    )
+    .await
+}
+
+async fn fork_session_request(
+    conn: &AcpServerConnection,
+    request: ForkSessionRequest,
+) -> anyhow::Result<ForkSessionResponse> {
+    conn.cx()
+        .send_request(request)
+        .block_task()
+        .await
+        .map_err(Into::into)
+}
+
+async fn seed_session_with_messages(
+    session_manager: &SessionManager,
+    cwd: &Path,
+    messages: &[(&str, i64)],
+) -> goose::session::Session {
+    let session = session_manager
+        .create_session(
+            cwd.to_path_buf(),
+            "Fork before".to_string(),
+            SessionType::Acp,
+            GooseMode::default(),
+        )
+        .await
+        .unwrap();
+
+    for (text, created) in messages {
+        let mut message = Message::user().with_text(*text);
+        message.created = *created;
+        session_manager
+            .add_message(&session.id, &message)
+            .await
+            .unwrap();
+    }
+
+    session
+}
+
+async fn session_texts(session_manager: &SessionManager, session_id: &str) -> Vec<String> {
+    session_manager
+        .get_session(session_id, true)
+        .await
+        .unwrap()
+        .conversation
+        .unwrap()
+        .messages()
+        .iter()
+        .flat_map(|message| {
+            message.content.iter().filter_map(|content| match content {
+                MessageContent::Text(text) => Some(text.text.clone()),
+                _ => None,
+            })
+        })
+        .collect()
+}
+
+fn conversation_before_meta(timestamp: i64) -> serde_json::Map<String, serde_json::Value> {
+    let mut meta = serde_json::Map::new();
+    meta.insert(
+        "conversationBefore".to_string(),
+        serde_json::Value::Number(timestamp.into()),
+    );
+    meta
+}
+
+#[test]
+fn fork_session_conversation_before_matches_rest_cutoff() {
+    run_test(async {
+        let data_root = tempfile::tempdir().unwrap();
+        let cwd = tempfile::tempdir().unwrap();
+        let session_manager = SessionManager::new(data_root.path().to_path_buf());
+        let session = seed_session_with_messages(
+            &session_manager,
+            cwd.path(),
+            &[
+                ("first", 1_718_000_000),
+                ("second", 1_718_000_060),
+                ("third", 1_718_000_120),
+            ],
+        )
+        .await;
+        let conn = new_connection(data_root.path()).await;
+
+        let response = fork_session_request(
+            &conn,
+            ForkSessionRequest::new(SessionId::new(session.id.clone()), cwd.path())
+                .meta(conversation_before_meta(1_718_000_120)),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(
+            session_texts(&session_manager, response.session_id.0.as_ref()).await,
+            vec!["first", "second"]
+        );
+        assert_eq!(
+            session_texts(&session_manager, &session.id).await,
+            vec!["first", "second", "third"]
+        );
+    });
+}