diff --git a/.repository-projection.json b/.repository-projection.json index 0c6731d07..a4d276714 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "083eacdbac0fef20630986791529393aa4e2683c", + "sourceSha": "be83027d2b5902447ca1ba418d2b4f85df867bbc", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "402b1d7e517279b33d3f6d6aef9c77a0633a8266", + "priorProjectedBase": "d5e2b3dd85095b0190afc25bb7492414ad27a29b", "definitionDigest": "cb9d429542ebb0a2de9b42a7aad60d9d8696a648ceba47c30f05c0b285ca0db7", "toolDigest": "f8cb071b0f27267120ccf45a00d0982f45113bd23535bef6a1555b4933f99f13", - "contentDigest": "b19cd9d7b8b22856b6565af0c70d61f3b0125ade7f5dd8e4494610dce7cd538d", + "contentDigest": "4676d676f45ea745c21a6ebb2d20aedcf3d83bcf978ca61d2f303c5ca4a09c82", "publicationEligible": true } diff --git a/packages/runtime-gateway-rs/src/chat.rs b/packages/runtime-gateway-rs/src/chat.rs index 2c12104a5..f1421e7a6 100644 --- a/packages/runtime-gateway-rs/src/chat.rs +++ b/packages/runtime-gateway-rs/src/chat.rs @@ -1,4 +1,5 @@ use super::*; +use crate::turn_diffs::{begin_chat_snapshot, finish_chat_snapshot}; use maestro_local_host::embedding::{EmbeddedAgent, EmbeddedAgentBuilder}; pub(crate) fn is_chat_endpoint(head: &RequestHead) -> bool { @@ -179,19 +180,21 @@ async fn handle_codex_app_server_chat( model: &str, prompt: &str, attachment_paths: &[String], + scope: Option<&str>, ) -> Result { - handle_codex_app_server_chat_transport( + handle_codex_app_server_chat_transport_scoped( stream, state, session_id, model, prompt, attachment_paths, - CodexBridgeTransport::Sse, + (CodexBridgeTransport::Sse, scope), ) .await } +#[cfg(test)] pub(crate) async fn handle_codex_app_server_chat_transport( stream: &mut TcpStream, state: &AppState, @@ -201,6 +204,28 @@ pub(crate) async fn handle_codex_app_server_chat_transport( attachment_paths: &[String], transport: CodexBridgeTransport, ) -> Result { + handle_codex_app_server_chat_transport_scoped( + stream, + state, + session_id, + model, + prompt, + attachment_paths, + (transport, None), + ) + .await +} + +async fn handle_codex_app_server_chat_transport_scoped( + stream: &mut TcpStream, + state: &AppState, + session_id: Option<&str>, + model: &str, + prompt: &str, + attachment_paths: &[String], + (transport, scope): (CodexBridgeTransport, Option<&str>), +) -> Result { + let mut turn_snapshot = begin_chat_snapshot(state, session_id, prompt, scope).await; let session_approval_mode = approval_mode_for_session(state, session_id).await; let approval_mode = codex_app_server_approval_mode(&session_approval_mode); send_codex_bridge_event( @@ -263,7 +288,7 @@ pub(crate) async fn handle_codex_app_server_chat_transport( send_codex_bridge_tool_event(stream, transport, tool_event).await?; } - let message = + let mut message = composer_assistant_message(&assistant_output.text, "", assistant_output.usage.clone()); if !assistant_output.text.is_empty() { send_codex_bridge_event( @@ -281,6 +306,7 @@ pub(crate) async fn handle_codex_app_server_chat_transport( ) .await?; } + finish_chat_snapshot(&mut turn_snapshot, &mut message).await; record_chat_assistant_message(state, session_id, message.clone()).await; record_usage_entry( state, @@ -327,15 +353,16 @@ async fn handle_codex_app_server_chat_ws( model: &str, prompt: &str, attachment_paths: &[String], + scope: Option<&str>, ) -> Result { - handle_codex_app_server_chat_transport( + handle_codex_app_server_chat_transport_scoped( stream, state, session_id, model, prompt, attachment_paths, - CodexBridgeTransport::WebSocket, + (CodexBridgeTransport::WebSocket, scope), ) .await } @@ -620,6 +647,7 @@ pub(crate) async fn handle_chat_endpoint( &codex_model, &prompt, &prepared_attachments.paths, + turn_scope.as_deref(), ) .await?; if prompt_succeeded { @@ -635,6 +663,13 @@ pub(crate) async fn handle_chat_endpoint( cleanup_prepared_attachments(prepared_attachments).await; return Ok(()); } + let mut turn_snapshot = begin_chat_snapshot( + &state, + session_id.as_deref(), + &prompt, + turn_scope.as_deref(), + ) + .await; let (usage_provider, usage_model) = usage_provider_model(&chat, &state, &model).await; let (mut client_tools, client_tool_names) = client_tool_definitions(&chat); // Gateway-handled session-messaging tools give an agent turn the same reach @@ -1304,12 +1339,13 @@ pub(crate) async fn handle_chat_endpoint( take_client_tool_results(&state, &client_tool_call_ids).await; finish_client_tool_metadata(&mut assistant_tools, &client_tool_results); let usage = last_usage.take(); - let message = composer_assistant_message_with_tools( + let mut message = composer_assistant_message_with_tools( &assistant_text, &thinking_text, usage, &assistant_tools, ); + finish_chat_snapshot(&mut turn_snapshot, &mut message).await; record_chat_assistant_message(&state, session_id.as_deref(), message.clone()).await; send_sse( &mut stream, @@ -1588,6 +1624,7 @@ pub(crate) async fn handle_chat_websocket_endpoint( &codex_model, &prompt, &prepared_attachments.paths, + turn_scope.as_deref(), ) .await?; if prompt_succeeded { @@ -1604,6 +1641,13 @@ pub(crate) async fn handle_chat_websocket_endpoint( cleanup_prepared_attachments(prepared_attachments).await; return Ok(()); } + let mut turn_snapshot = begin_chat_snapshot( + &state, + session_id.as_deref(), + &prompt, + turn_scope.as_deref(), + ) + .await; let (usage_provider, usage_model) = usage_provider_model(&chat, &state, &model).await; let (mut client_tools, client_tool_names) = client_tool_definitions(&chat); // Gateway-handled session-messaging tools give an agent turn the same reach @@ -2251,12 +2295,13 @@ pub(crate) async fn handle_chat_websocket_endpoint( take_client_tool_results(&state, &client_tool_call_ids).await; finish_client_tool_metadata(&mut assistant_tools, &client_tool_results); let usage = last_usage.take(); - let message = composer_assistant_message_with_tools( + let mut message = composer_assistant_message_with_tools( &assistant_text, &thinking_text, usage, &assistant_tools, ); + finish_chat_snapshot(&mut turn_snapshot, &mut message).await; record_chat_assistant_message(&state, session_id.as_deref(), message.clone()).await; send_ws_json( &mut stream, diff --git a/packages/runtime-gateway-rs/src/lib.rs b/packages/runtime-gateway-rs/src/lib.rs index 5d1d46efc..edde766ad 100644 --- a/packages/runtime-gateway-rs/src/lib.rs +++ b/packages/runtime-gateway-rs/src/lib.rs @@ -55,6 +55,7 @@ mod migrations; mod model_catalog; mod session_messaging; mod sessions; +mod turn_diffs; #[allow(unused_imports)] pub(crate) use a2a::{ @@ -89,16 +90,18 @@ use a2a_platform_registration::{ #[cfg(test)] use a2a_skill_catalog::A2A_SUBAGENT_REQUEST_METADATA_PATH; use auth::*; +#[cfg(test)] +use chat::handle_codex_app_server_chat_transport; #[allow(unused_imports)] pub(crate) use chat::{ ChatAttachment, ChatMessage, ChatRequest, ExtractAttachmentRequest, ExtractDocumentOutput, PreparedAttachments, approval_blocked_tool_event, build_prompt_from_chat, composer_assistant_message, composer_assistant_message_with_tools, composer_text_content, finish_client_tool_metadata, finish_tool_metadata, handle_chat_endpoint, - handle_chat_websocket_endpoint, handle_codex_app_server_chat_transport, is_chat_endpoint, - is_chat_websocket_endpoint, prepare_chat_attachments, record_chat_user_message, - record_tool_call_metadata, send_sse, send_ws_json, sse_headers, strip_data_url_prefix, - try_parse_websocket_text_message, update_tool_metadata_status, websocket_accept_key, + handle_chat_websocket_endpoint, is_chat_endpoint, is_chat_websocket_endpoint, + prepare_chat_attachments, record_chat_user_message, record_tool_call_metadata, send_sse, + send_ws_json, sse_headers, strip_data_url_prefix, try_parse_websocket_text_message, + update_tool_metadata_status, websocket_accept_key, }; pub(crate) use codex_bridge::*; use extended::{ExtendedApiState, handle_extended_endpoint, is_extended_endpoint}; @@ -461,6 +464,7 @@ struct AppState { shared_sessions: Arc>>, approval_modes: Arc>>, pending_tool_responses: Arc>>, + native_snapshot_registry: Arc, // Maps a pending-request id to the session or authenticated principal that // owns the blocked agent turn. Sessionless client-tool turns retain a // resumable owner without weakening the cross-tenant resume check. @@ -796,6 +800,7 @@ pub async fn serve_listener( shared_sessions: Arc::new(Mutex::new(shared_sessions)), approval_modes: Arc::new(Mutex::new(HashMap::new())), pending_tool_responses: Arc::new(Mutex::new(HashMap::new())), + native_snapshot_registry: Arc::new(turn_diffs::NativeSnapshotRegistry::default()), pending_tool_response_sessions: Arc::new(Mutex::new(HashMap::new())), completed_client_tool_results: Arc::new(Mutex::new(HashMap::new())), a2a_tasks: Arc::new(Mutex::new(a2a_tasks)), diff --git a/packages/runtime-gateway-rs/src/sessions.rs b/packages/runtime-gateway-rs/src/sessions.rs index 86f2645a3..8dfe7edc2 100644 --- a/packages/runtime-gateway-rs/src/sessions.rs +++ b/packages/runtime-gateway-rs/src/sessions.rs @@ -692,6 +692,9 @@ pub(super) async fn handle_session_get( match session_path.tail { None => json_response(200, &session_full_value(&session)), + Some("turn-diff") => { + crate::turn_diffs::session_turn_diff_response(head, state, &session).await + } Some("timeline") => json_response(200, &session_timeline_value(&session)), Some("share") => json_response( 200, diff --git a/packages/runtime-gateway-rs/src/tests.rs b/packages/runtime-gateway-rs/src/tests.rs index 978b566dd..1f9c2ede5 100644 --- a/packages/runtime-gateway-rs/src/tests.rs +++ b/packages/runtime-gateway-rs/src/tests.rs @@ -2739,6 +2739,7 @@ fn test_app_state_with_sessions(sessions: HashMap) -> App shared_sessions: Arc::new(Mutex::new(HashMap::new())), approval_modes: Arc::new(Mutex::new(HashMap::new())), pending_tool_responses: Arc::new(Mutex::new(HashMap::new())), + native_snapshot_registry: Arc::new(turn_diffs::NativeSnapshotRegistry::default()), pending_tool_response_sessions: Arc::new(Mutex::new(HashMap::new())), completed_client_tool_results: Arc::new(Mutex::new(HashMap::new())), extended_api: Arc::new(Mutex::new(ExtendedApiState::default())), @@ -13272,6 +13273,7 @@ async fn delete_session_subpath_returns_404_without_removing_session() { shared_sessions: Arc::new(Mutex::new(HashMap::new())), approval_modes: Arc::new(Mutex::new(HashMap::new())), pending_tool_responses: Arc::new(Mutex::new(HashMap::new())), + native_snapshot_registry: Arc::new(turn_diffs::NativeSnapshotRegistry::default()), pending_tool_response_sessions: Arc::new(Mutex::new(HashMap::new())), completed_client_tool_results: Arc::new(Mutex::new(HashMap::new())), extended_api: Arc::new(Mutex::new(ExtendedApiState::default())), @@ -13363,6 +13365,7 @@ async fn invalid_session_store_is_left_untouched_and_future_writes_are_blocked() shared_sessions: Arc::new(Mutex::new(HashMap::new())), approval_modes: Arc::new(Mutex::new(HashMap::new())), pending_tool_responses: Arc::new(Mutex::new(HashMap::new())), + native_snapshot_registry: Arc::new(turn_diffs::NativeSnapshotRegistry::default()), pending_tool_response_sessions: Arc::new(Mutex::new(HashMap::new())), completed_client_tool_results: Arc::new(Mutex::new(HashMap::new())), extended_api: Arc::new(Mutex::new(ExtendedApiState::default())), @@ -13502,81 +13505,6 @@ fn enterprise_policy_admin_routes_are_implemented() { assert!(is_extended_endpoint(&head)); } } -#[test] -fn undo_endpoint_reads_and_consumes_tui_checkpoint_store() { - use maestro_local_host::checkpoints::{Checkpoint, CheckpointStore, EntryKind, FileEntry}; - use sha2::{Digest, Sha256}; - - let temp = unique_test_dir("maestro-undo-checkpoint"); - std::fs::create_dir_all(&temp).unwrap(); - // Match checkpoint capture: persist the resolved root, not a platform alias. - let temp = dunce::canonicalize(temp).unwrap(); - let file = temp.join("src.txt"); - let before = b"before"; - let after = b"after"; - std::fs::write(&file, after).unwrap(); - let before_hash = Sha256::digest(before) - .iter() - .map(|byte| format!("{byte:02x}")) - .collect::(); - let after_hash = Sha256::digest(after) - .iter() - .map(|byte| format!("{byte:02x}")) - .collect::(); - - let store = CheckpointStore::new(&temp.join("sessions"), "session-1"); - let checkpoint_dir = store.root().join("checkpoint-1"); - std::fs::create_dir_all(checkpoint_dir.join("blobs")).unwrap(); - std::fs::write(checkpoint_dir.join("blobs").join(&before_hash), before).unwrap(); - let checkpoint = Checkpoint { - id: "checkpoint-1".to_string(), - created_at: "2026-08-05T00:00:00Z".to_string(), - prompt: "edit src.txt".to_string(), - repo_root: temp.clone(), - head: None, - user_turn_index: None, - entries: vec![FileEntry { - path: "src.txt".to_string(), - kind: EntryKind::Modified, - pre_blob: Some(before_hash), - post_hash: Some(after_hash), - }], - }; - std::fs::write( - checkpoint_dir.join("checkpoint.json"), - serde_json::to_vec(&checkpoint).unwrap(), - ) - .unwrap(); - - let head = RequestHead { - method: "GET".to_string(), - path: "/api/undo".to_string(), - query: HashMap::from([(String::from("sessionId"), String::from("session-1"))]), - headers: HashMap::new(), - }; - let summary = undo_response_for_store(&head, &store); - assert_eq!(summary["totalChanges"], 1); - assert_eq!(summary["canUndo"], true); - - let restored = restore_undo_response_for_store(&store); - assert_eq!(restored["success"], true, "{restored}"); - assert_eq!(std::fs::read(&file).unwrap(), before); - assert_eq!(store.list().len(), 0); - // A missing restore blob must not become success after other file work. - std::fs::create_dir_all(&checkpoint_dir).unwrap(); - std::fs::write( - checkpoint_dir.join("checkpoint.json"), - serde_json::to_vec(&checkpoint).unwrap(), - ) - .unwrap(); - std::fs::write(&file, after).unwrap(); - let failed = restore_undo_response_for_store(&store); - assert_eq!(failed["success"], false); - assert_eq!(failed["failedFiles"].as_array().unwrap().len(), 1); - assert_eq!(std::fs::read(&file).unwrap(), after); - assert_eq!(store.list().len(), 1); - let _ = std::fs::remove_dir_all(temp); -} // --------------------------------------------------------------------------- // Session-to-session messaging @@ -14599,3 +14527,6 @@ async fn platform_a2a_push_evicts_terminal_payloads_and_replay_history() { assert!(!tasks.contains_key("callback-0000")); assert!(!histories.contains_key("callback-0000")); } + +#[path = "tests/turn_diffs.rs"] +mod turn_diff_tests; diff --git a/packages/runtime-gateway-rs/src/tests/turn_diffs.rs b/packages/runtime-gateway-rs/src/tests/turn_diffs.rs new file mode 100644 index 000000000..527b780bf --- /dev/null +++ b/packages/runtime-gateway-rs/src/tests/turn_diffs.rs @@ -0,0 +1,381 @@ +use super::*; + +#[test] +fn undo_endpoint_reads_and_consumes_tui_checkpoint_store() { + use maestro_local_host::checkpoints::{Checkpoint, CheckpointStore, EntryKind, FileEntry}; + use sha2::{Digest, Sha256}; + + let temp = unique_test_dir("maestro-undo-checkpoint"); + std::fs::create_dir_all(&temp).unwrap(); + // Match checkpoint capture: persist the resolved root, not a platform alias. + let temp = dunce::canonicalize(temp).unwrap(); + let file = temp.join("src.txt"); + let before = b"before"; + let after = b"after"; + std::fs::write(&file, after).unwrap(); + let before_hash = Sha256::digest(before) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + let after_hash = Sha256::digest(after) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + + let store = CheckpointStore::new(&temp.join("sessions"), "session-1"); + let checkpoint_dir = store.root().join("checkpoint-1"); + std::fs::create_dir_all(checkpoint_dir.join("blobs")).unwrap(); + std::fs::write(checkpoint_dir.join("blobs").join(&before_hash), before).unwrap(); + let checkpoint = Checkpoint { + id: "checkpoint-1".to_string(), + created_at: "2026-08-05T00:00:00Z".to_string(), + prompt: "edit src.txt".to_string(), + repo_root: temp.clone(), + head: None, + user_turn_index: None, + entries: vec![FileEntry { + path: "src.txt".to_string(), + kind: EntryKind::Modified, + pre_blob: Some(before_hash), + post_hash: Some(after_hash), + post_snapshot_oversized: false, + }], + }; + std::fs::write( + checkpoint_dir.join("checkpoint.json"), + serde_json::to_vec(&checkpoint).unwrap(), + ) + .unwrap(); + + let head = RequestHead { + method: "GET".to_string(), + path: "/api/undo".to_string(), + query: HashMap::from([(String::from("sessionId"), String::from("session-1"))]), + headers: HashMap::new(), + }; + let summary = undo_response_for_store(&head, &store); + assert_eq!(summary["totalChanges"], 1); + assert_eq!(summary["canUndo"], true); + + let restored = restore_undo_response_for_store(&store); + assert_eq!(restored["success"], true, "{restored}"); + assert_eq!(std::fs::read(&file).unwrap(), before); + assert_eq!(store.list().len(), 0); + // A missing restore blob must not become success after other file work. + std::fs::create_dir_all(&checkpoint_dir).unwrap(); + std::fs::write( + checkpoint_dir.join("checkpoint.json"), + serde_json::to_vec(&checkpoint).unwrap(), + ) + .unwrap(); + std::fs::write(&file, after).unwrap(); + let failed = restore_undo_response_for_store(&store); + assert_eq!(failed["success"], false); + assert_eq!(failed["failedFiles"].as_array().unwrap().len(), 1); + assert_eq!(std::fs::read(&file).unwrap(), after); + assert_eq!(store.list().len(), 1); + let _ = std::fs::remove_dir_all(temp); +} + +#[tokio::test] +async fn native_turn_diff_query_preserves_completed_identity_and_session_authority() { + let root = TestDir::new("native-turn-diff"); + for args in [ + vec!["init", "--quiet"], + vec!["config", "user.email", "test@example.test"], + vec!["config", "user.name", "Test"], + ] { + assert!( + std::process::Command::new("git") + .args(args) + .current_dir(root.path()) + .status() + .unwrap() + .success() + ); + } + let session: SessionRecord = serde_json::from_value(serde_json::json!({ + "id": "native-session", "owner": "owner", "organizationId": "org", "workspaceId": "ws", + "title": "Turn", "createdAt": "2026-10-03T00:00:00Z", "updatedAt": "2026-10-03T00:00:00Z", "messageCount": 2, + "messages": [{"role":"user", "content":"first"}, {"role":"user", "content":"concurrent later turn"}] + })).unwrap(); + let mut state = + test_app_state_with_sessions(HashMap::from([(session.id.clone(), session.clone())])); + let mut config = auth_test_config(); + config.cwd = dunce::canonicalize(root.path()).unwrap(); + config.session_store_path = root.path().join("gateway-state").join("sessions.json"); + state.config = Arc::new(config); + let mut snapshot = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "first", + Some("native-session:created:1"), + ) + .await; + fs::write(root.path().join("result.txt"), "saved result").unwrap(); + let mut message = serde_json::json!({"role":"assistant", "content":"done"}); + crate::turn_diffs::finish_chat_snapshot(&mut snapshot, &mut message).await; + assert_eq!(message["fileSnapshotTurnIndex"], 0); + let mut completed = session.clone(); + completed.messages.push(message); + state + .sessions + .lock() + .await + .sessions + .insert(completed.id.clone(), completed.clone()); + fs::write(root.path().join("result.txt"), "later mutation").unwrap(); + let auth = AuthContext { + subject: Some("owner".to_string()), + organization_id: Some("org".to_string()), + workspace_id: Some("ws".to_string()), + ..AuthContext::default() + }; + let mut head = csrf_head_for_path("GET", "/api/sessions/native-session/turn-diff", None); + let response = crate::sessions::handle_session_get( + &head, + &state, + session_path_from_path(&head.path).unwrap(), + &auth, + ) + .await; + let text = String::from_utf8(response).unwrap(); + assert!(text.starts_with("HTTP/1.1 200")); + assert!(text.contains("saved result")); + assert!(!text.contains("later mutation")); + for forbidden in [ + AuthContext { + subject: Some("other".to_string()), + ..auth.clone() + }, + AuthContext { + workspace_id: Some("other".to_string()), + ..auth.clone() + }, + ] { + let response = crate::sessions::handle_session_get( + &head, + &state, + session_path_from_path(&head.path).unwrap(), + &forbidden, + ) + .await; + assert!( + String::from_utf8(response) + .unwrap() + .starts_with("HTTP/1.1 404") + ); + } + head.query.insert("turnIndex".to_string(), "1".to_string()); + let response = crate::turn_diffs::session_turn_diff_response(&head, &state, &completed).await; + assert!( + String::from_utf8(response) + .unwrap() + .contains("\"availability\":\"unavailable\"") + ); + head.query.insert("turnIndex".to_string(), "2".to_string()); + let response = crate::turn_diffs::session_turn_diff_response(&head, &state, &completed).await; + assert!( + String::from_utf8(response) + .unwrap() + .starts_with("HTTP/1.1 400") + ); + // A public ID can be reused, but old generation and ownership cannot. + let mut new_generation = completed.clone(); + new_generation.created_at = "2026-10-03T00:00:01Z".to_string(); + let mut new_owner = completed.clone(); + new_owner.owner = Some("new-owner".to_string()); + for recreated in [new_generation, new_owner] { + head.query.insert("turnIndex".to_string(), "0".to_string()); + let response = + crate::turn_diffs::session_turn_diff_response(&head, &state, &recreated).await; + assert!( + String::from_utf8(response) + .unwrap() + .contains("\"availability\":\"unavailable\"") + ); + } + head.query.clear(); + completed + .messages + .push(serde_json::json!({"role":"assistant", "fileSnapshotTurnIndex":1})); + let response = crate::turn_diffs::session_turn_diff_response(&head, &state, &completed).await; + assert!( + String::from_utf8(response) + .unwrap() + .contains("\"availability\":\"unavailable\"") + ); +} + +#[tokio::test] +async fn native_overlapping_turn_snapshots_are_unavailable_without_serializing_turns() { + let root = TestDir::new("native-overlap-snapshots"); + assert!( + std::process::Command::new("git") + .args(["init", "--quiet"]) + .current_dir(root.path()) + .status() + .unwrap() + .success() + ); + let mut session = test_session_record("overlap-session"); + session.messages = vec![ + serde_json::json!({"role":"user", "content":"a"}), + serde_json::json!({"role":"user", "content":"b"}), + ]; + let mut state = + test_app_state_with_sessions(HashMap::from([(session.id.clone(), session.clone())])); + let mut config = auth_test_config(); + config.cwd = dunce::canonicalize(root.path()).unwrap(); + config.session_store_path = root.path().join("gateway-state").join("sessions.json"); + state.config = Arc::new(config); + let mut first = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "a", + Some("session:created:1"), + ) + .await; + let mut second = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "b", + Some("session:created:2"), + ) + .await; + assert!( + first.is_some() && second.is_some(), + "both turns progress independently" + ); + fs::write(root.path().join("shared.txt"), "overlapping edits").unwrap(); + let mut a = serde_json::json!({"role":"assistant", "content":"a done"}); + let mut b = serde_json::json!({"role":"assistant", "content":"b done"}); + crate::turn_diffs::finish_chat_snapshot(&mut first, &mut a).await; + crate::turn_diffs::finish_chat_snapshot(&mut second, &mut b).await; + assert_eq!(a["fileSnapshotAvailable"], false); + assert_eq!(b["fileSnapshotAvailable"], false); + assert_eq!(a["fileSnapshotTurnIndex"], 0); + assert_eq!(b["fileSnapshotTurnIndex"], 1); + session.messages.extend([a, b]); + let mut head = csrf_head_for_path("GET", "/api/sessions/overlap-session/turn-diff", None); + for index in [0, 1] { + head.query + .insert("turnIndex".to_string(), index.to_string()); + let response = crate::turn_diffs::session_turn_diff_response(&head, &state, &session).await; + assert!( + String::from_utf8(response) + .unwrap() + .contains("\"availability\":\"unavailable\"") + ); + } + // Dropping a canceled capture releases detection state and pending files. + let canceled = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "cancel", + Some("session:created:2"), + ) + .await; + drop(canceled); + let mut final_turn = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "alone", + Some("session:created:2"), + ) + .await; + let mut message = serde_json::json!({"role":"assistant"}); + crate::turn_diffs::finish_chat_snapshot(&mut final_turn, &mut message).await; + assert_eq!(message["fileSnapshotAvailable"], true); +} + +#[tokio::test] +async fn native_sessionless_turns_invalidate_overlapping_session_snapshots() { + let root = TestDir::new("native-sessionless-overlap-snapshots"); + assert!( + std::process::Command::new("git") + .args(["init", "--quiet"]) + .current_dir(root.path()) + .status() + .unwrap() + .success() + ); + let mut session = test_session_record("sessionless-overlap-session"); + session.messages = vec![serde_json::json!({"role":"user", "content":"scoped"})]; + let mut state = + test_app_state_with_sessions(HashMap::from([(session.id.clone(), session.clone())])); + let mut config = auth_test_config(); + config.cwd = dunce::canonicalize(root.path()).unwrap(); + config.session_store_path = root.path().join("gateway-state").join("sessions.json"); + state.config = Arc::new(config); + for sessionless_first in [false, true] { + let (mut scoped, mut sessionless) = if sessionless_first { + let sessionless = + crate::turn_diffs::begin_chat_snapshot(&state, None, "sessionless", None).await; + let scoped = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "scoped", + Some("session:created:1"), + ) + .await; + (scoped, sessionless) + } else { + let scoped = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "scoped", + Some("session:created:1"), + ) + .await; + let sessionless = + crate::turn_diffs::begin_chat_snapshot(&state, None, "sessionless", None).await; + (scoped, sessionless) + }; + assert!(scoped.is_some() && sessionless.is_some()); + fs::write(root.path().join("shared.txt"), "sessionless edit").unwrap(); + let mut scoped_message = serde_json::json!({"role":"assistant", "content":"scoped done"}); + let mut sessionless_message = + serde_json::json!({"role":"assistant", "content":"sessionless done"}); + if sessionless_first { + crate::turn_diffs::finish_chat_snapshot(&mut sessionless, &mut sessionless_message) + .await; + } + crate::turn_diffs::finish_chat_snapshot(&mut scoped, &mut scoped_message).await; + if !sessionless_first { + crate::turn_diffs::finish_chat_snapshot(&mut sessionless, &mut sessionless_message) + .await; + } + assert_eq!(scoped_message["fileSnapshotAvailable"], false); + assert_eq!(scoped_message["fileSnapshotTurnIndex"], 0); + assert!(sessionless_message.get("fileSnapshotAvailable").is_none()); + assert!(sessionless_message.get("fileSnapshotTurnIndex").is_none()); + let mut completed = session.clone(); + completed.messages.push(scoped_message); + let head = csrf_head_for_path( + "GET", + "/api/sessions/sessionless-overlap-session/turn-diff", + None, + ); + let response = + crate::turn_diffs::session_turn_diff_response(&head, &state, &completed).await; + assert!( + String::from_utf8(response) + .unwrap() + .contains("\"availability\":\"unavailable\"") + ); + } + // A rejected or canceled sessionless turn releases its lease on drop. + let canceled = crate::turn_diffs::begin_chat_snapshot(&state, None, "cancel", None).await; + drop(canceled); + let mut alone = crate::turn_diffs::begin_chat_snapshot( + &state, + Some(&session.id), + "alone", + Some("session:created:1"), + ) + .await; + let mut message = serde_json::json!({"role":"assistant"}); + crate::turn_diffs::finish_chat_snapshot(&mut alone, &mut message).await; + assert_eq!(message["fileSnapshotAvailable"], true); +} diff --git a/packages/runtime-gateway-rs/src/turn_diffs.rs b/packages/runtime-gateway-rs/src/turn_diffs.rs new file mode 100644 index 000000000..97a76881a --- /dev/null +++ b/packages/runtime-gateway-rs/src/turn_diffs.rs @@ -0,0 +1,253 @@ +//! Native completed-turn review reads the session checkpoint owner, never Git HEAD +//! or the current worktree. Gateway state roots and authorized session IDs scope it. +use super::*; +use maestro_local_host::checkpoints::{ + CheckpointStore, PendingTurn, begin_turn_snapshot, finalize_turn_snapshot, +}; + +fn snapshot_root(state: &AppState) -> &Path { + state + .config + .session_store_path + .parent() + .unwrap_or(&state.config.cwd) +} + +fn native_snapshot_key(session: &SessionRecord, workspace: &Path) -> String { + use base64::Engine; + use sha2::{Digest, Sha256}; + // Session IDs may be reused after deletion. Creation and ownership are + // immutable within a generation; include the canonical workspace as well. + let workspace = dunce::canonicalize(workspace).unwrap_or_else(|_| workspace.to_path_buf()); + let identity = serde_json::json!([ + session.id, + session.created_at, + session.owner, + session.organization_id, + session.workspace_id, + workspace + ]); + format!( + "native-{}", + base64::engine::general_purpose::URL_SAFE_NO_PAD + .encode(Sha256::digest(identity.to_string().as_bytes())) + ) +} + +/// Ephemeral ambiguity detection only. Turns never wait for each other. +#[derive(Default)] +pub(super) struct NativeSnapshotRegistry { + active: std::sync::Mutex>>, +} + +struct CaptureLease { + registry: Arc, + ambiguous: Arc, +} + +impl NativeSnapshotRegistry { + fn acquire(self: &Arc) -> CaptureLease { + use std::sync::atomic::{AtomicBool, Ordering}; + let mut active = self.active.lock().expect("snapshot registry lock"); + active.retain(|entry| entry.strong_count() > 0); + let ambiguous = Arc::new(AtomicBool::new(!active.is_empty())); + for other in active.iter().filter_map(std::sync::Weak::upgrade) { + other.store(true, Ordering::SeqCst); + } + active.push(Arc::downgrade(&ambiguous)); + CaptureLease { + registry: self.clone(), + ambiguous, + } + } +} + +impl CaptureLease { + fn complete(&self) -> bool { + let mut active = self.registry.active.lock().expect("snapshot registry lock"); + let this = Arc::downgrade(&self.ambiguous); + active.retain(|entry| !entry.ptr_eq(&this) && entry.strong_count() > 0); + self.ambiguous.load(std::sync::atomic::Ordering::SeqCst) + } +} + +impl Drop for CaptureLease { + fn drop(&mut self) { + self.complete(); + } +} + +pub(super) struct ChatSnapshot { + capture: Option, + lease: CaptureLease, +} + +struct SessionSnapshot { + turn_index: usize, + pending: Option, + store: CheckpointStore, +} + +pub(super) async fn begin_chat_snapshot( + state: &AppState, + session_id: Option<&str>, + prompt: &str, + scope: Option<&str>, +) -> Option { + // Every accepted native turn can edit this workspace, including turns with + // no durable session. Hold its lease until completion or error/drop. + let mut snapshot = ChatSnapshot { + capture: None, + lease: state.native_snapshot_registry.acquire(), + }; + let Some(session_id) = session_id else { + return Some(snapshot); + }; + let sessions = state.sessions.lock().await; + let Some(session) = sessions.sessions.get(session_id) else { + return Some(snapshot); + }; + // The accepted message count identifies this submission even when another + // actor appends a message before capture. Do not reassign it to the tail. + let message_count = scope + .and_then(|scope| scope.rsplit(':').next()?.parse::().ok()) + .unwrap_or(session.messages.len()); + let Some(turn_index) = session + .messages + .iter() + .take(message_count) + .filter(|message| message.get("role").and_then(Value::as_str) == Some("user")) + .count() + .checked_sub(1) + else { + return Some(snapshot); + }; + let checkpoint_key = native_snapshot_key(session, &state.config.cwd); + drop(sessions); + let cwd = state.config.cwd.clone(); + let root = snapshot_root(state).to_path_buf(); + let prompt = prompt.to_string(); + let store = CheckpointStore::new(&root, &checkpoint_key); + let pending = tokio::task::spawn_blocking(move || { + let mut pending = begin_turn_snapshot(&cwd, &root, &checkpoint_key, &prompt)?; + pending.user_turn_index = Some(turn_index); + Some(pending) + }) + .await + .ok() + .flatten(); + snapshot.capture = Some(SessionSnapshot { + turn_index, + pending, + store, + }); + Some(snapshot) +} + +pub(super) async fn finish_chat_snapshot(snapshot: &mut Option, message: &mut Value) { + let Some(snapshot) = snapshot.take() else { + return; + }; + let Some(capture) = snapshot.capture else { + return; + }; + let mut saved = None; + if let Some(pending) = capture.pending { + if !snapshot + .lease + .ambiguous + .load(std::sync::atomic::Ordering::SeqCst) + { + match tokio::task::spawn_blocking(move || finalize_turn_snapshot(pending)).await { + Ok(Ok(checkpoint)) => saved = checkpoint, + outcome => tracing::warn!(?outcome, "completed turn checkpoint could not be saved"), + } + } + } + let ambiguous = snapshot.lease.complete(); + if ambiguous { + if let Some(checkpoint) = &saved { + if let Err(error) = capture.store.remove(&checkpoint.id) { + tracing::warn!(%error, "ambiguous snapshot cleanup failed"); + } + } + } + message["fileSnapshotAvailable"] = serde_json::json!(saved.is_some() && !ambiguous); + // Completion identity is durable even when file capture failed. Latest-turn + // reads therefore report unavailable instead of showing an older turn. + message["fileSnapshotTurnIndex"] = serde_json::json!(capture.turn_index); +} + +pub(super) async fn session_turn_diff_response( + head: &RequestHead, + state: &AppState, + session: &SessionRecord, +) -> Vec { + let turn_index = match head.query.get("turnIndex") { + Some(value) => match value.parse::() { + Ok(index) => Some(index), + Err(_) => { + return json_response(400, &serde_json::json!({"error": "Invalid turn index"})); + } + }, + None => session + .messages + .iter() + .rev() + .find(|message| message.get("role").and_then(Value::as_str) == Some("assistant")) + .and_then(|message| message.get("fileSnapshotTurnIndex").and_then(Value::as_u64)) + .and_then(|index| usize::try_from(index).ok()), + }; + let user_turns = session + .messages + .iter() + .filter(|message| message.get("role").and_then(Value::as_str) == Some("user")) + .count(); + if turn_index.is_some_and(|index| index >= user_turns) { + return json_response( + 400, + &serde_json::json!({"error": "Turn range is unavailable"}), + ); + } + let completed_snapshot = turn_index.is_some_and(|index| { + session.messages.iter().any(|message| { + message.get("role").and_then(Value::as_str) == Some("assistant") + && message.get("fileSnapshotTurnIndex").and_then(Value::as_u64) + == Some(index as u64) + && message + .get("fileSnapshotAvailable") + .and_then(Value::as_bool) + == Some(true) + }) + }); + let store = CheckpointStore::new( + snapshot_root(state), + &native_snapshot_key(session, &state.config.cwd), + ); + let cwd = state.config.cwd.clone(); + let snapshot = if let Some(turn_index) = turn_index.filter(|_| completed_snapshot) { + tokio::task::spawn_blocking(move || store.turn_snapshot(Some(turn_index), &cwd)).await + } else { + Ok(Ok(None)) + }; + match snapshot { + Ok(Ok(Some((index, files)))) if index < user_turns => json_response( + 200, + &serde_json::json!({ + "sessionId": session.id, "workspacePath": state.config.cwd, + "turnIndex": index, "availability": "ready", "files": files, + }), + ), + Ok(Ok(_)) => json_response( + 200, + &serde_json::json!({ + "sessionId": session.id, "workspacePath": state.config.cwd, + "turnIndex": turn_index, "availability": "unavailable", "files": [], + }), + ), + error => json_response( + 500, + &serde_json::json!({"error": format!("Cannot read turn checkpoint: {error:?}")}), + ), + } +} diff --git a/packages/runtime-rs/src/agent/codex_app_server_turns.rs b/packages/runtime-rs/src/agent/codex_app_server_turns.rs index deef51165..cf3f54b22 100644 --- a/packages/runtime-rs/src/agent/codex_app_server_turns.rs +++ b/packages/runtime-rs/src/agent/codex_app_server_turns.rs @@ -2713,3 +2713,7 @@ mod tests { } } } + +#[cfg(test)] +#[path = "codex_app_server_turns_provider_replay.rs"] +mod provider_replay; diff --git a/packages/runtime-rs/src/agent/codex_app_server_turns_provider_replay.rs b/packages/runtime-rs/src/agent/codex_app_server_turns_provider_replay.rs new file mode 100644 index 000000000..488738a34 --- /dev/null +++ b/packages/runtime-rs/src/agent/codex_app_server_turns_provider_replay.rs @@ -0,0 +1,529 @@ +//! Test-only provider-shaped transcripts through the native stdio parser. +//! +//! The fixture subprocess is this Rust test binary, never a model SDK or a +//! second runtime. See fixtures/provider_replay/README.md for provenance. + +use super::*; +use std::io::{BufRead, Write}; +use std::path::PathBuf; + +const TRANSCRIPT: &str = include_str!("fixtures/provider_replay/restart.ndjson"); +const SUBPROCESS_TEST: &str = + "agent::codex_app_server_turns::provider_replay::provider_replay_subprocess"; + +fn materialize(frame: &mut Value, workspace: &str, request_id: &Value) { + match frame { + Value::String(value) if value == "$workspace" => *frame = json!(workspace), + Value::String(value) if value == "$request_id" => *frame = request_id.clone(), + Value::Object(fields) => { + for value in fields.values_mut() { + materialize(value, workspace, request_id); + } + } + Value::Array(values) => { + for value in values { + materialize(value, workspace, request_id); + } + } + _ => {} + } +} + +// Match all declared contract fields while permitting provider/client metadata +// outside the fixture. Arrays are exact, so an extra prompt cannot be hidden. +fn assert_frame(expected: &Value, actual: &Value) { + match expected { + Value::Object(fields) => { + let actual = actual.as_object().expect("expected JSON object"); + for (key, value) in fields { + assert_frame( + value, + actual.get(key).unwrap_or_else(|| panic!("missing {key}")), + ); + } + } + Value::Array(values) => { + let actual = actual.as_array().expect("expected JSON array"); + assert_eq!(values.len(), actual.len(), "unexpected additional input"); + for (expected, actual) in values.iter().zip(actual) { + assert_frame(expected, actual); + } + } + _ => assert_eq!(expected, actual, "outbound frame differs from transcript"), + } +} + +/// Launched by the parent tests with child-only environment, over real stdio. +#[test] +#[ignore = "provider fixture subprocess; invoked by provider_replay tests"] +fn provider_replay_subprocess() { + let Ok(phase) = env::var("MAESTRO_REPLAY_PHASE") else { + return; + }; + let result = std::panic::catch_unwind(|| { + let workspace = env::var("MAESTRO_REPLAY_WORKSPACE").expect("workspace"); + let audit = env::var("MAESTRO_REPLAY_AUDIT").expect("audit path"); + let mut audit = std::fs::File::create(audit).expect("create audit"); + // A hung parent cannot leave a fixture child blocked indefinitely on stdin. + let (line_tx, line_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + for line in std::io::stdin().lock().lines() { + if line_tx.send(line).is_err() { + break; + } + } + }); + let stdout = std::io::stdout(); + let mut stdout = stdout.lock(); + let mut request_id = Value::Null; + let mut steps = 0; + for line in TRANSCRIPT.lines() { + let step: Value = serde_json::from_str(line).expect("fixture JSON"); + if step["phase"] != phase { + continue; + } + steps += 1; + let mut frame = step["frame"].clone(); + materialize(&mut frame, &workspace, &request_id); + match step["direction"].as_str().expect("direction") { + "outbound" => { + let line = line_rx + .recv_timeout(Duration::from_secs(5)) + .expect("expected bounded client frame") + .expect("read stdin"); + let actual: Value = serde_json::from_str(&line).expect("client JSON"); + assert_frame(&frame, &actual); + if actual.get("method").is_some() && actual["id"].is_number() { + request_id = actual["id"].clone(); + } + writeln!(audit, "{actual}").expect("audit frame"); + audit.flush().expect("flush audit"); + } + "inbound" => { + writeln!(stdout, "{frame}").expect("write provider frame"); + stdout.flush().expect("flush provider frame"); + } + direction => panic!("unknown fixture direction {direction}"), + } + } + assert!(steps > 0, "unknown transcript phase {phase}"); + // Completion is authoritative only after the parent has dropped its + // client and the child has exhausted stdin, including any extra frames. + assert!( + matches!( + line_rx.recv_timeout(Duration::from_secs(5)), + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) + ), + "unexpected client frame or no EOF after transcript" + ); + writeln!(audit, "{{\"complete\":true}}").expect("audit completion"); + audit.flush().expect("flush completion"); + }); + let address = env::var("MAESTRO_REPLAY_COMPLETION") + .expect("completion address") + .parse() + .expect("socket address"); + let mut completion = std::net::TcpStream::connect_timeout(&address, Duration::from_secs(5)) + .expect("completion connection"); + completion + .set_write_timeout(Some(Duration::from_secs(5))) + .expect("write timeout"); + completion + .write_all(if result.is_ok() { + b"complete\n" + } else { + b"failed\n" + }) + .expect("completion signal"); + assert!(result.is_ok(), "transcript subprocess failed"); +} + +struct ReplayProcess { + client: CodexAppServerClient, + audit: PathBuf, + completion: tokio::net::TcpListener, +} + +async fn replay_process(phase: &str, workspace: &Path, audit_root: &Path) -> ReplayProcess { + let audit = audit_root.join(format!("{phase}.ndjson")); + let completion = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("completion listener"); + let child_env = HashMap::from([ + ( + "MAESTRO_REPLAY_COMPLETION".to_owned(), + completion.local_addr().expect("address").to_string(), + ), + ("MAESTRO_REPLAY_PHASE".to_owned(), phase.to_owned()), + ( + "MAESTRO_REPLAY_WORKSPACE".to_owned(), + workspace.to_string_lossy().into_owned(), + ), + ( + "MAESTRO_REPLAY_AUDIT".to_owned(), + audit.to_string_lossy().into_owned(), + ), + ]); + let client = CodexAppServerClient::spawn_with_env( + Some( + env::current_exe() + .expect("test binary") + .to_string_lossy() + .into_owned(), + ), + Some(vec![ + "--exact".to_owned(), + SUBPROCESS_TEST.to_owned(), + "--ignored".to_owned(), + "--nocapture".to_owned(), + "--format=terse".to_owned(), + ]), + Some(5_000), + &child_env, + ) + .await + .expect("spawn Rust provider fixture"); + ReplayProcess { + client, + audit, + completion, + } +} + +fn manifest(workspace: &Path, session_id: &str) -> CodexSessionManifest { + CodexSessionManifest { + key: CodexSessionKey::new("fixture-profile", workspace, "gpt-5.5") + .expect("session key") + .with_session_id(Some(session_id)), + approval_policy: "on-request".to_owned(), + sandbox: "read-only".to_owned(), + capabilities: CodexCapabilities::default(), + } +} + +async fn completion_signal(listener: tokio::net::TcpListener) -> Vec { + use tokio::io::AsyncReadExt; + tokio::time::timeout(Duration::from_secs(5), async { + let (mut socket, _) = listener.accept().await.expect("child completion"); + let mut signal = Vec::new(); + socket + .read_to_end(&mut signal) + .await + .expect("completion signal"); + signal + }) + .await + .expect("child must exhaust stdin and acknowledge completion") +} + +async fn completed_audit(listener: tokio::net::TcpListener, path: &Path) -> Vec { + assert_eq!(completion_signal(listener).await, b"complete\n"); + let frames = audit_frames(path); + assert_eq!(frames.last(), Some(&json!({"complete": true}))); + frames +} + +fn audit_frames(path: &Path) -> Vec { + std::fs::read_to_string(path) + .expect("read audit") + .lines() + .map(|line| serde_json::from_str(line).expect("audit JSON")) + .collect() +} + +async fn replay_turn( + session: &CodexAppServerTurnSession, + prompt: &str, + expected_text: &str, + calls: &mut Vec<(String, String, Value)>, + approvals: &mut Vec, +) { + let turn_id = session + .start_text_turn(prompt, Some(5_000)) + .await + .expect("start turn"); + loop { + match session + .wait_server_request_or_turn_complete(&turn_id, Some(5_000)) + .await + .expect("provider event") + { + TurnWaitEvent::ServerRequest(request) => { + assert_eq!( + request.params.as_ref().unwrap()["threadId"], + session.thread_id() + ); + assert_eq!(request.params.as_ref().unwrap()["turnId"], turn_id); + match request.method.as_str() { + "item/tool/call" => { + let call = parse_tool_call_params(request.params.as_ref().unwrap()) + .expect("production tool parser"); + let result = match call.0.as_str() { + "spawn_subagent" | "resume_subagent" => "child-session", + "read" => "fixture contents", + name => panic!("unexpected tool {name}"), + }; + calls.push(call); + request.respond(tool_call_success_result(result)); + } + "item/commandExecution/requestApproval" => { + approvals.push(request.id.clone()); + request.respond(approval_decision(false)); + } + method => panic!("unexpected server request {method}"), + } + } + TurnWaitEvent::Completed(result) => { + assert_eq!(result.thread_id, session.thread_id()); + assert_eq!(result.turn_id, turn_id); + assert_eq!( + result.assistant_text, expected_text, + "delta and completion must reconcile once" + ); + assert!(result.assistant_text_is_full); + assert!(result.provider_failure().is_none()); + break; + } + TurnWaitEvent::Pending => panic!("transcript did not complete"), + } + } + // A real provider method fences the transcript: unexpected injection or + // duplicated starts/replies fail the subprocess before this response. + session + .client() + .request( + "thread/read", + Some(json!({ + "threadId": session.thread_id(), "includeTurns": false + })), + Some(5_000), + ) + .await + .expect("transcript fence"); +} + +#[tokio::test] +async fn provider_replay_parent_and_child_resume_after_runtime_recreation() { + tokio::time::timeout(Duration::from_secs(30), async { + let state = tempfile::tempdir().expect("state"); + let workspace = tempfile::tempdir().expect("workspace"); + let parent = manifest(workspace.path(), "parent-session"); + let child = manifest(workspace.path(), "child-session"); + let mut calls = Vec::new(); + let mut approvals = Vec::new(); + let cases = [ + ( + "parent_created", + &parent, + "thread-parent", + "Inspect the fixture with a child.", + "Child queued.", + CodexSessionOpen::Created, + ), + ( + "child_created", + &child, + "thread-child", + "Inspect fixture.txt.", + "Fixture inspected.", + CodexSessionOpen::Created, + ), + ( + "parent_resumed", + &parent, + "thread-parent", + "Ask the same child to report.", + "Existing child resumed.", + CodexSessionOpen::Resumed, + ), + ( + "child_resumed", + &child, + "thread-child", + "Report the inspection.", + "Fixture report complete.", + CodexSessionOpen::Resumed, + ), + ]; + let mut wire_frames = Vec::new(); + for (phase, manifest, thread_id, prompt, text, open_kind) in cases { + let process = replay_process(phase, workspace.path(), state.path()).await; + let history = if open_kind == CodexSessionOpen::Resumed { + vec![Message { + role: Role::User, + content: MessageContent::text("Previously accepted prompt must not replay"), + }] + } else { + Vec::new() + }; + let names = if manifest.key.session_id.as_deref() == Some("parent-session") { + vec!["spawn_subagent", "resume_subagent"] + } else { + vec!["read"] + }; + let tools = names + .into_iter() + .map(|name| DynamicToolSpec { + name: name.to_owned(), + description: format!("Fixture {name}"), + input_schema: json!({"type": "object"}), + }) + .collect::>(); + let session = CodexAppServerTurnSession::connect_with_client_and_manifest( + process.client, + manifest.clone(), + state.path(), + &tools, + Some("Fixture replay instructions.".to_owned()), + &history, + ) + .await + .expect("open persistent session"); + assert_eq!(session.thread_id(), thread_id); + assert_eq!(session.open_kind(), open_kind); + replay_turn(&session, prompt, text, &mut calls, &mut approvals).await; + let binding = CodexThreadBinding::load_at(state.path(), &manifest.key) + .expect("read durable binding") + .expect("binding exists"); + assert_eq!(binding.thread_id, thread_id); + // Read the audit only after a wire fence; no sleep/poll determines + // correctness. The next case gets a fresh process and runtime. + drop(session); + let frames = completed_audit(process.completion, &process.audit).await; + assert_eq!( + frames + .iter() + .filter(|frame| frame["method"] == "turn/start") + .count(), + 1 + ); + assert!( + !frames + .iter() + .any(|frame| frame["method"] == "thread/inject_items") + ); + wire_frames.extend(frames); + } + assert_eq!( + calls.iter().map(|call| call.1.as_str()).collect::>(), + ["call-spawn", "call-read", "call-resume"] + ); + assert_eq!(calls[2].2["id"], child.key.session_id.as_deref().unwrap()); + assert_eq!(approvals, [json!("approval-command-1")]); + assert_eq!( + wire_frames + .iter() + .filter(|frame| frame["method"] == "thread/start") + .count(), + 2 + ); + assert_eq!( + wire_frames + .iter() + .filter(|frame| frame["method"] == "thread/resume") + .count(), + 2 + ); + }) + .await + .expect("bounded provider replay"); +} + +#[tokio::test] +async fn provider_replay_resume_failure_preserves_binding_without_replacement() { + tokio::time::timeout(Duration::from_secs(10), async { + let state = tempfile::tempdir().expect("state"); + let workspace = tempfile::tempdir().expect("workspace"); + let manifest = manifest(workspace.path(), "parent-session"); + let binding = CodexThreadBinding::new( + manifest.key.clone(), + "thread-parent", + Some("2025-01-01".to_owned()), + 1, + ); + binding.store_at(state.path()).expect("store binding"); + let process = replay_process("resume_unavailable", workspace.path(), state.path()).await; + let error = CodexAppServerTurnSession::connect_with_client_and_manifest( + process.client, + manifest.clone(), + state.path(), + &[], + None, + &[], + ) + .await + .err() + .expect("resume must fail"); + assert!( + error + .to_string() + .contains("Provider temporarily unavailable") + ); + assert_eq!( + CodexThreadBinding::load_at(state.path(), &manifest.key).expect("load binding"), + Some(binding) + ); + let frames = completed_audit(process.completion, &process.audit).await; + assert_eq!( + frames + .iter() + .filter(|frame| frame["method"] == "thread/resume") + .count(), + 1 + ); + assert!(!frames.iter().any(|frame| matches!( + frame["method"].as_str(), + Some("thread/start" | "thread/inject_items" | "turn/start") + ))); + }) + .await + .expect("bounded provider failure replay"); +} + +#[tokio::test] +async fn provider_replay_rejects_extra_action_after_final_fence() { + tokio::time::timeout(Duration::from_secs(15), async { + let state = tempfile::tempdir().expect("state"); + let workspace = tempfile::tempdir().expect("workspace"); + let process = replay_process("child_created", workspace.path(), state.path()).await; + let tools = vec![DynamicToolSpec { + name: "read".to_owned(), + description: "Fixture read".to_owned(), + input_schema: json!({"type": "object"}), + }]; + let session = CodexAppServerTurnSession::connect_with_client_and_manifest( + process.client, + manifest(workspace.path(), "child-session"), + state.path(), + &tools, + Some("Fixture replay instructions.".to_owned()), + &[], + ) + .await + .expect("open child"); + replay_turn( + &session, + "Inspect fixture.txt.", + "Fixture inspected.", + &mut Vec::new(), + &mut Vec::new(), + ) + .await; + let duplicate = session.client().request("turn/start", Some(json!({ + "threadId": "thread-child", "input": [{"type": "text", "text": "Inspect fixture.txt."}] + })), Some(5_000)).await; + assert!( + duplicate.is_err(), + "unsolicited action must terminate transcript" + ); + drop(session); + assert_eq!(completion_signal(process.completion).await, b"failed\n"); + assert!( + !audit_frames(&process.audit) + .iter() + .any(|frame| frame["complete"] == true) + ); + }) + .await + .expect("bounded negative replay"); +} diff --git a/packages/runtime-rs/src/agent/fixtures/provider_replay/README.md b/packages/runtime-rs/src/agent/fixtures/provider_replay/README.md new file mode 100644 index 000000000..881a1cb91 --- /dev/null +++ b/packages/runtime-rs/src/agent/fixtures/provider_replay/README.md @@ -0,0 +1,46 @@ +# Native Codex restart replay fixtures + +`restart.ndjson` is a deterministic, synthesized and scrubbed transcript. It +adapts the restart/resume scenario from T3 Code commit +`a5b34b25378fbbf90ce36ee101a11528ccf11c7e`, especially +`provider_thread_resume/codex_transcript.ndjson` and +`claude_subagent_resume_after_restart/claude_transcript.ndjson`. No customer +transcript, credentials, model reasoning, signatures, filesystem roots, or +provider installation identifiers were copied. Fixed session, call, item, and +thread identifiers and benign prompts replace original data. + +The Codex paginated thread result, turn lifecycle, agent-message delta, and +completed item shapes follow the provider thread-resume recording. The +capability declaration, dynamic subagent/read calls, approval denial, and +transport-unavailable error are synthesized against Maestro's current native +wire contract. Claude's restart scenario contributes the child-identity +invariant only; this is not a Claude SDK replay or a recorded Codex subagent +transcript. There is no verified recorded missing-thread error in this corpus. + +Tests use the existing Codex app-server client and turn-session adapter, +spawning the Rust test binary as a strict stdio transcript peer. Fresh parent +and keyed child sessions store real Codex thread bindings. Both processes and +turn-session adapters are recreated against those bindings; restored messages +must not be injected, accepted prompts must not be resubmitted, and the parent +addresses the same child identity. Ordered outbound checks and audit assertions +reject repeated tool replies or starts. A bounded child completion signal is +emitted only after stdin is exhausted; an explicit negative test sends an +extra action after the final wire fence and requires failure, preventing a +premature completion marker from masking replay. Streamed and completed assistant text +must reconcile to a single message. Approval requests pass through the native +server-request queue and receive one denial; no command executes in the fixture. + +This covers transport parsing and persistent provider binding recreation. It +does not claim live provider operation, actual subagent scheduler recovery, +production command execution, or whole-actor session recovery. + +Run from the repository root on a host that passes the build capacity guard: + +```sh +cargo test --manifest-path products/maestro/Cargo.toml -p maestro-runtime provider_replay --locked +``` + +The ignored `provider_replay_subprocess` test is the child entry point and is +invoked by the replay tests, not manually in CI. Each parent test has a bounded +outer timeout; no sleeps, model calls, or external network services are needed. +A test-only loopback socket acknowledges bounded child completion. diff --git a/packages/runtime-rs/src/agent/fixtures/provider_replay/restart.ndjson b/packages/runtime-rs/src/agent/fixtures/provider_replay/restart.ndjson new file mode 100644 index 000000000..eaa0474cc --- /dev/null +++ b/packages/runtime-rs/src/agent/fixtures/provider_replay/restart.ndjson @@ -0,0 +1,65 @@ +{"phase":"parent_created","direction":"outbound","frame":{"method":"initialize","params":{"capabilities":{"experimentalApi":true}}}} +{"phase":"parent_created","direction":"inbound","frame":{"id":"$request_id","result":{"protocolVersion":"2025-01-01","capabilities":{"methods":["thread/start","thread/resume","turn/start","turn/interrupt"],"notifications":["item/tool/call","item/agentMessage/delta","turn/completed"]}}}} +{"phase":"parent_created","direction":"outbound","frame":{"method":"initialized"}} +{"phase":"parent_created","direction":"outbound","frame":{"method":"thread/start","params":{"model":"gpt-5.5","cwd":"$workspace","approvalPolicy":"on-request","sandbox":"read-only","dynamicTools":[{"name":"spawn_subagent","description":"Fixture spawn_subagent","inputSchema":{"type":"object"}},{"name":"resume_subagent","description":"Fixture resume_subagent","inputSchema":{"type":"object"}}],"developerInstructions":"Fixture replay instructions."}}} +{"phase":"parent_created","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-parent","sessionId":"thread-parent","parentThreadId":null,"historyMode":"paginated","cwd":"$workspace","status":{"type":"idle"},"turns":[]}}}} +{"phase":"parent_created","direction":"outbound","frame":{"method":"turn/start","params":{"threadId":"thread-parent","input":[{"type":"text","text":"Inspect the fixture with a child."}]}}} +{"phase":"parent_created","direction":"inbound","frame":{"id":"$request_id","result":{"turn":{"id":"turn-parent-1","items":[],"itemsView":"notLoaded","status":"inProgress","error":null}}}} +{"phase":"parent_created","direction":"inbound","frame":{"method":"turn/started","params":{"threadId":"thread-parent","turn":{"id":"turn-parent-1","status":"inProgress","items":[]}}}} +{"phase":"parent_created","direction":"inbound","frame":{"id":"approval-command-1","method":"item/commandExecution/requestApproval","params":{"threadId":"thread-parent","turnId":"turn-parent-1","itemId":"command-1","command":"touch fixture.txt","cwd":"$workspace"}}} +{"phase":"parent_created","direction":"outbound","frame":{"id":"approval-command-1","result":{"decision":"decline"}}} +{"phase":"parent_created","direction":"inbound","frame":{"id":"rpc-call-spawn","method":"item/tool/call","params":{"threadId":"thread-parent","turnId":"turn-parent-1","callId":"call-spawn","tool":"spawn_subagent","arguments":{"task":"Inspect fixture.txt"}}}} +{"phase":"parent_created","direction":"outbound","frame":{"id":"rpc-call-spawn","result":{"success":true,"contentItems":[{"type":"inputText","text":"child-session"}]}}} +{"phase":"parent_created","direction":"inbound","frame":{"method":"item/agentMessage/delta","params":{"threadId":"thread-parent","turnId":"turn-parent-1","itemId":"message-turn-parent-1","delta":"Child queued."}}} +{"phase":"parent_created","direction":"inbound","frame":{"method":"item/completed","params":{"threadId":"thread-parent","turnId":"turn-parent-1","item":{"id":"message-turn-parent-1","type":"agentMessage","text":"Child queued.","phase":"final_answer"}}}} +{"phase":"parent_created","direction":"inbound","frame":{"method":"turn/completed","params":{"threadId":"thread-parent","turn":{"id":"turn-parent-1","items":[],"status":"completed","error":null}}}} +{"phase":"parent_created","direction":"outbound","frame":{"method":"thread/read","params":{"threadId":"thread-parent","includeTurns":false}}} +{"phase":"parent_created","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-parent"}}}} +{"phase":"child_created","direction":"outbound","frame":{"method":"initialize","params":{"capabilities":{"experimentalApi":true}}}} +{"phase":"child_created","direction":"inbound","frame":{"id":"$request_id","result":{"protocolVersion":"2025-01-01","capabilities":{"methods":["thread/start","thread/resume","turn/start","turn/interrupt"],"notifications":["item/tool/call","item/agentMessage/delta","turn/completed"]}}}} +{"phase":"child_created","direction":"outbound","frame":{"method":"initialized"}} +{"phase":"child_created","direction":"outbound","frame":{"method":"thread/start","params":{"model":"gpt-5.5","cwd":"$workspace","approvalPolicy":"on-request","sandbox":"read-only","dynamicTools":[{"name":"read","description":"Fixture read","inputSchema":{"type":"object"}}],"developerInstructions":"Fixture replay instructions."}}} +{"phase":"child_created","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-child","sessionId":"thread-child","parentThreadId":"thread-parent","historyMode":"paginated","cwd":"$workspace","status":{"type":"idle"},"turns":[]}}}} +{"phase":"child_created","direction":"outbound","frame":{"method":"turn/start","params":{"threadId":"thread-child","input":[{"type":"text","text":"Inspect fixture.txt."}]}}} +{"phase":"child_created","direction":"inbound","frame":{"id":"$request_id","result":{"turn":{"id":"turn-child-1","items":[],"itemsView":"notLoaded","status":"inProgress","error":null}}}} +{"phase":"child_created","direction":"inbound","frame":{"method":"turn/started","params":{"threadId":"thread-child","turn":{"id":"turn-child-1","status":"inProgress","items":[]}}}} +{"phase":"child_created","direction":"inbound","frame":{"id":"rpc-call-read","method":"item/tool/call","params":{"threadId":"thread-child","turnId":"turn-child-1","callId":"call-read","tool":"read","arguments":{"path":"fixture.txt"}}}} +{"phase":"child_created","direction":"outbound","frame":{"id":"rpc-call-read","result":{"success":true,"contentItems":[{"type":"inputText","text":"fixture contents"}]}}} +{"phase":"child_created","direction":"inbound","frame":{"method":"item/agentMessage/delta","params":{"threadId":"thread-child","turnId":"turn-child-1","itemId":"message-turn-child-1","delta":"Fixture inspected."}}} +{"phase":"child_created","direction":"inbound","frame":{"method":"item/completed","params":{"threadId":"thread-child","turnId":"turn-child-1","item":{"id":"message-turn-child-1","type":"agentMessage","text":"Fixture inspected.","phase":"final_answer"}}}} +{"phase":"child_created","direction":"inbound","frame":{"method":"turn/completed","params":{"threadId":"thread-child","turn":{"id":"turn-child-1","items":[],"status":"completed","error":null}}}} +{"phase":"child_created","direction":"outbound","frame":{"method":"thread/read","params":{"threadId":"thread-child","includeTurns":false}}} +{"phase":"child_created","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-child"}}}} +{"phase":"parent_resumed","direction":"outbound","frame":{"method":"initialize","params":{"capabilities":{"experimentalApi":true}}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"protocolVersion":"2025-01-01","capabilities":{"methods":["thread/start","thread/resume","turn/start","turn/interrupt"],"notifications":["item/tool/call","item/agentMessage/delta","turn/completed"]}}}} +{"phase":"parent_resumed","direction":"outbound","frame":{"method":"initialized"}} +{"phase":"parent_resumed","direction":"outbound","frame":{"method":"thread/resume","params":{"threadId":"thread-parent","model":"gpt-5.5","cwd":"$workspace"}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-parent","sessionId":"thread-parent","parentThreadId":null,"historyMode":"paginated","cwd":"$workspace","status":{"type":"idle"},"turns":[]}}}} +{"phase":"parent_resumed","direction":"outbound","frame":{"method":"turn/start","params":{"threadId":"thread-parent","input":[{"type":"text","text":"Ask the same child to report."}]}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"turn":{"id":"turn-parent-2","items":[],"itemsView":"notLoaded","status":"inProgress","error":null}}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"method":"turn/started","params":{"threadId":"thread-parent","turn":{"id":"turn-parent-2","status":"inProgress","items":[]}}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"id":"rpc-call-resume","method":"item/tool/call","params":{"threadId":"thread-parent","turnId":"turn-parent-2","callId":"call-resume","tool":"resume_subagent","arguments":{"id":"child-session","prompt":"Report the inspection."}}}} +{"phase":"parent_resumed","direction":"outbound","frame":{"id":"rpc-call-resume","result":{"success":true,"contentItems":[{"type":"inputText","text":"child-session"}]}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"method":"item/agentMessage/delta","params":{"threadId":"thread-parent","turnId":"turn-parent-2","itemId":"message-turn-parent-2","delta":"Existing child resumed."}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"method":"item/completed","params":{"threadId":"thread-parent","turnId":"turn-parent-2","item":{"id":"message-turn-parent-2","type":"agentMessage","text":"Existing child resumed.","phase":"final_answer"}}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"method":"turn/completed","params":{"threadId":"thread-parent","turn":{"id":"turn-parent-2","items":[],"status":"completed","error":null}}}} +{"phase":"parent_resumed","direction":"outbound","frame":{"method":"thread/read","params":{"threadId":"thread-parent","includeTurns":false}}} +{"phase":"parent_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-parent"}}}} +{"phase":"child_resumed","direction":"outbound","frame":{"method":"initialize","params":{"capabilities":{"experimentalApi":true}}}} +{"phase":"child_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"protocolVersion":"2025-01-01","capabilities":{"methods":["thread/start","thread/resume","turn/start","turn/interrupt"],"notifications":["item/tool/call","item/agentMessage/delta","turn/completed"]}}}} +{"phase":"child_resumed","direction":"outbound","frame":{"method":"initialized"}} +{"phase":"child_resumed","direction":"outbound","frame":{"method":"thread/resume","params":{"threadId":"thread-child","model":"gpt-5.5","cwd":"$workspace"}}} +{"phase":"child_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-child","sessionId":"thread-child","parentThreadId":"thread-parent","historyMode":"paginated","cwd":"$workspace","status":{"type":"idle"},"turns":[]}}}} +{"phase":"child_resumed","direction":"outbound","frame":{"method":"turn/start","params":{"threadId":"thread-child","input":[{"type":"text","text":"Report the inspection."}]}}} +{"phase":"child_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"turn":{"id":"turn-child-2","items":[],"itemsView":"notLoaded","status":"inProgress","error":null}}}} +{"phase":"child_resumed","direction":"inbound","frame":{"method":"turn/started","params":{"threadId":"thread-child","turn":{"id":"turn-child-2","status":"inProgress","items":[]}}}} +{"phase":"child_resumed","direction":"inbound","frame":{"method":"item/agentMessage/delta","params":{"threadId":"thread-child","turnId":"turn-child-2","itemId":"message-turn-child-2","delta":"Fixture report complete."}}} +{"phase":"child_resumed","direction":"inbound","frame":{"method":"item/completed","params":{"threadId":"thread-child","turnId":"turn-child-2","item":{"id":"message-turn-child-2","type":"agentMessage","text":"Fixture report complete.","phase":"final_answer"}}}} +{"phase":"child_resumed","direction":"inbound","frame":{"method":"turn/completed","params":{"threadId":"thread-child","turn":{"id":"turn-child-2","items":[],"status":"completed","error":null}}}} +{"phase":"child_resumed","direction":"outbound","frame":{"method":"thread/read","params":{"threadId":"thread-child","includeTurns":false}}} +{"phase":"child_resumed","direction":"inbound","frame":{"id":"$request_id","result":{"thread":{"id":"thread-child"}}}} +{"phase":"resume_unavailable","direction":"outbound","frame":{"method":"initialize","params":{"capabilities":{"experimentalApi":true}}}} +{"phase":"resume_unavailable","direction":"inbound","frame":{"id":"$request_id","result":{"protocolVersion":"2025-01-01","capabilities":{"methods":["thread/start","thread/resume","turn/start","turn/interrupt"],"notifications":["item/tool/call","item/agentMessage/delta","turn/completed"]}}}} +{"phase":"resume_unavailable","direction":"outbound","frame":{"method":"initialized"}} +{"phase":"resume_unavailable","direction":"outbound","frame":{"method":"thread/resume","params":{"threadId":"thread-parent","model":"gpt-5.5","cwd":"$workspace"}}} +{"phase":"resume_unavailable","direction":"inbound","frame":{"id":"$request_id","error":{"code":-32000,"message":"Provider temporarily unavailable","data":{"kind":"transportUnavailable"}}}} diff --git a/packages/session-rs/src/checkpoints.rs b/packages/session-rs/src/checkpoints.rs index c84546e4c..e9f1ff9ee 100644 --- a/packages/session-rs/src/checkpoints.rs +++ b/packages/session-rs/src/checkpoints.rs @@ -230,6 +230,10 @@ pub struct FileEntry { /// SHA-256 of the post-turn content. `None` means the file was absent /// after the turn. pub post_hash: Option, + /// Post content exceeded the bounded review snapshot budget. The streamed + /// hash still protects rewind and deletion of created files. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub post_snapshot_oversized: bool, } /// A finalized checkpoint manifest. @@ -294,10 +298,36 @@ pub fn begin_turn( sessions_dir: &Path, session_id: &str, prompt: &str, +) -> Option { + begin_turn_inner(cwd, sessions_dir, session_id, prompt, false) +} + +/// Bounded native review capture. Unreadable/oversized preimages invalidate the +/// review rather than silently publishing an empty or partial turn. +#[must_use] +pub fn begin_turn_snapshot( + cwd: &Path, + sessions_dir: &Path, + session_id: &str, + prompt: &str, +) -> Option { + begin_turn_inner(cwd, sessions_dir, session_id, prompt, true) +} + +fn begin_turn_inner( + cwd: &Path, + sessions_dir: &Path, + session_id: &str, + prompt: &str, + bounded: bool, ) -> Option { let repo_root = dunce::canonicalize(git::repo_root(cwd)?).ok()?; let head = git_text(&repo_root, &["rev-parse", "--verify", "HEAD"]); - let status = status_snapshot(&repo_root)?; + let mut status = status_snapshot(&repo_root)?; + if bounded { + status.untracked = expand_native_untracked(&repo_root, &status.untracked)?; + status.dirty.extend(status.untracked.iter().cloned()); + } let store = CheckpointStore::new(sessions_dir, session_id); let id = format!( @@ -318,8 +348,25 @@ pub fn begin_turn( let mut pre_dirty = HashMap::new(); let mut unreadable = HashSet::new(); + let mut budget = 512 * 1024; for path in status.dirty { - match checked_worktree_path(&repo_root, &path).and_then(fs::read) { + match checked_worktree_path(&repo_root, &path).and_then(|path| { + if !bounded { + return fs::read(path); + } + let mut bytes = Vec::new(); + fs::File::open(path)? + .take(budget + 1) + .read_to_end(&mut bytes)?; + if bytes.len() as u64 > budget { + return Err(io::Error::new( + io::ErrorKind::FileTooLarge, + "Preimage exceeds review limit", + )); + } + budget = budget.saturating_sub(bytes.len() as u64); + Ok(bytes) + }) { Ok(bytes) => match store_blob(&cp_dir, &bytes) { Ok(hash) => { pre_dirty.insert(path, Some(hash)); @@ -356,23 +403,81 @@ pub fn begin_turn( /// Diff the worktree against the pre-turn snapshot and persist a checkpoint /// for everything the turn changed. Returns `None` when nothing changed (the /// pending directory is removed) or the repository disappeared. -pub fn finalize_turn(mut pending: PendingTurn) -> io::Result> { +pub fn finalize_turn(pending: PendingTurn) -> io::Result> { + finalize_turn_inner(pending, false) +} + +/// Retain a completed native turn even when it made no changes. Both sides are +/// saved in the existing checkpoint store; later reads never consult the worktree. +pub fn finalize_turn_snapshot(pending: PendingTurn) -> io::Result> { + finalize_turn_inner(pending, true) +} + +fn finalize_turn_inner( + mut pending: PendingTurn, + retain_empty: bool, +) -> io::Result> { let cp_dir = pending.store.new_checkpoint_dir(&pending.id); - let Some(post) = status_snapshot(&pending.repo_root) else { + let Some(mut post) = status_snapshot(&pending.repo_root) else { let _ = fs::remove_dir_all(&cp_dir); return Ok(None); }; + if retain_empty { + post.untracked = expand_native_untracked(&pending.repo_root, &post.untracked) + .ok_or_else(|| io::Error::other("Untracked file list exceeds review limit"))?; + } + if retain_empty && !pending.unreadable.is_empty() { + return Err(io::Error::other("Pre-turn files could not be captured")); + } + let mut post_budget = 512 * 1024; + let dirty_pre_size = if retain_empty { + pending + .pre_dirty + .values() + .flatten() + .filter_map(|hash| fs::metadata(cp_dir.join("blobs").join(hash)).ok()) + .map(|metadata| metadata.len()) + .sum::() + } else { + 0 + }; + let mut pre_budget = (512_u64 * 1024).saturating_sub(dirty_pre_size); let mut entries: Vec = Vec::new(); // Tracked candidates: anything dirty before or after the turn. let mut candidates: HashSet<&str> = pending.pre_dirty.keys().map(String::as_str).collect(); candidates.extend(post.dirty.iter().map(String::as_str)); + // A turn may commit its edits before ending. Compare the saved beginning + // HEAD to the final worktree; current HEAD alone would report pristine. + let committed_paths = if retain_empty { + let mut command = Command::new("git"); + if let Some(head) = pending.head.as_deref() { + command.args(["diff", "--name-only", "-z", "--no-renames", head, "--"]); + } else { + command.args(["ls-files", "-z"]); + } + let output = command.current_dir(&pending.repo_root).output()?; + if !output.status.success() { + return Err(io::Error::other( + "Cannot enumerate turn changes from saved HEAD", + )); + } + String::from_utf8(output.stdout) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))? + .split('\0') + .filter(|path| !path.is_empty()) + .map(str::to_string) + .collect::>() + } else { + Vec::new() + }; + candidates.extend(committed_paths.iter().map(String::as_str)); for path in candidates { if pending.unreadable.contains(path) { continue; } - if pending.pre_untracked.contains(path) { + if pending.pre_untracked.contains(path) && !retain_empty { // Pre-existing untracked file: content was never snapshotted. continue; } @@ -382,19 +487,51 @@ pub fn finalize_turn(mut pending: PendingTurn) -> io::Result> // Clean at turn start: pre-turn content is the HEAD blob. If // the file is not in HEAD it was created (and staged) during // the turn. + if retain_empty + && pending + .head + .as_deref() + .and_then(|head| { + git_text( + &pending.repo_root, + &["cat-file", "-s", &format!("{head}:{path}")], + ) + }) + .and_then(|size| size.trim().parse::().ok()) + .is_some_and(|size| size > pre_budget) + { + return Err(io::Error::new( + io::ErrorKind::FileTooLarge, + "Preimage exceeds review limit", + )); + } pending .head .as_deref() .and_then(|head| head_blob(&pending.repo_root, head, path)) - .map(|bytes| store_blob(&cp_dir, &bytes)) + .map(|bytes| { + if retain_empty { + if bytes.len() as u64 > pre_budget { + return Err(io::Error::new( + io::ErrorKind::FileTooLarge, + "Preimage exceeds review limit", + )); + } + pre_budget -= bytes.len() as u64; + } + store_blob(&cp_dir, &bytes) + }) .transpose()? } }; - let Ok(post_hash) = - checked_worktree_path(&pending.repo_root, path).and_then(|path| file_hash(&path)) - else { - continue; - }; + let (post_hash, post_snapshot_oversized) = + match checked_worktree_path(&pending.repo_root, path) + .and_then(|path| snapshot_file(&cp_dir, &path, retain_empty, &mut post_budget)) + { + Ok(snapshot) => snapshot, + Err(error) if retain_empty => return Err(error), + Err(_) => continue, + }; if pre_blob == post_hash { continue; } @@ -408,6 +545,7 @@ pub fn finalize_turn(mut pending: PendingTurn) -> io::Result> kind, pre_blob, post_hash, + post_snapshot_oversized, }); } @@ -423,31 +561,45 @@ pub fn finalize_turn(mut pending: PendingTurn) -> io::Result> &mut files, ); if files.len() > MAX_EXPANDED_UNTRACKED_FILES { + if retain_empty { + return Err(io::Error::other("Created file list exceeds review limit")); + } continue; } for (rel, abs) in files { - if let Ok(Some(hash)) = file_hash(&abs) { + let snapshot = snapshot_file(&cp_dir, &abs, retain_empty, &mut post_budget); + if retain_empty && snapshot.is_err() { + return Err(snapshot.expect_err("capture failed")); + } + if let Ok((Some(hash), post_snapshot_oversized)) = snapshot { entries.push(FileEntry { path: rel, kind: EntryKind::Created, pre_blob: None, post_hash: Some(hash), + post_snapshot_oversized, }); } } - } else if let Ok(Some(hash)) = - checked_worktree_path(&pending.repo_root, path).and_then(|path| file_hash(&path)) - { - entries.push(FileEntry { - path: path.clone(), - kind: EntryKind::Created, - pre_blob: None, - post_hash: Some(hash), - }); + } else { + let snapshot = checked_worktree_path(&pending.repo_root, path) + .and_then(|path| snapshot_file(&cp_dir, &path, retain_empty, &mut post_budget)); + if retain_empty && snapshot.is_err() { + return Err(snapshot.expect_err("capture failed")); + } + if let Ok((Some(hash), post_snapshot_oversized)) = snapshot { + entries.push(FileEntry { + path: path.clone(), + kind: EntryKind::Created, + pre_blob: None, + post_hash: Some(hash), + post_snapshot_oversized, + }); + } } } - if entries.is_empty() { + if entries.is_empty() && !retain_empty { let _ = fs::remove_dir_all(&cp_dir); return Ok(None); } @@ -467,7 +619,8 @@ pub fn finalize_turn(mut pending: PendingTurn) -> io::Result> let referenced: HashSet<&str> = checkpoint .entries .iter() - .filter_map(|entry| entry.pre_blob.as_deref()) + .flat_map(|entry| [entry.pre_blob.as_deref(), entry.post_hash.as_deref()]) + .flatten() .collect(); for blob in fs::read_dir(cp_dir.join("blobs"))? { let blob = blob?; @@ -481,6 +634,180 @@ pub fn finalize_turn(mut pending: PendingTurn) -> io::Result> Ok(Some(checkpoint)) } +/// Bounded read of durable before/after content for one finalized turn. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TurnSnapshotFile { + pub path: String, + pub kind: EntryKind, + pub availability: &'static str, + pub before_content: Option, + pub after_content: Option, +} + +impl CheckpointStore { + /// Returns None for missing or evicted turns. Missing old post blobs remain + /// unavailable, rather than being reconstructed from today's worktree. + pub fn turn_snapshot( + &self, + turn_index: Option, + workspace: &Path, + ) -> io::Result)>> { + let checkpoint = self.list().into_iter().rev().find(|checkpoint| { + checkpoint + .user_turn_index + .is_some_and(|index| turn_index.is_none_or(|requested| index == requested)) + }); + let Some(checkpoint) = checkpoint else { + return Ok(None); + }; + let Some(repo_root) = + git::repo_root(workspace).and_then(|path| dunce::canonicalize(path).ok()) + else { + return Ok(None); + }; + if checkpoint.repo_root != repo_root { + return Ok(None); + } + if checkpoint.entries.len() > 1000 + || checkpoint + .entries + .iter() + .map(|entry| entry.path.len()) + .sum::() + > 128 * 1024 + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Checkpoint file list exceeds review limit", + )); + } + let dir = self.checkpoint_dir_for_restore(&checkpoint.id); + let mut budget = 512 * 1024; + let mut files = Vec::new(); + for entry in checkpoint.entries { + let mut read = |hash: Option<&str>| -> io::Result> { + let Some(hash) = hash else { + return Ok(Vec::new()); + }; + if hash.len() != 64 || !hash.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Invalid checkpoint blob hash", + )); + } + let path = dir.join("blobs").join(hash); + let size = fs::metadata(&path)?.len(); + if size > budget { + return Err(io::Error::new( + io::ErrorKind::FileTooLarge, + "Snapshot exceeds review limit", + )); + } + let mut bytes = Vec::new(); + fs::File::open(path)? + .take(budget + 1) + .read_to_end(&mut bytes)?; + if bytes.len() as u64 > budget { + return Err(io::Error::new( + io::ErrorKind::FileTooLarge, + "Snapshot exceeds review limit", + )); + } + if sha256(&bytes) != hash { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Checkpoint content is corrupt", + )); + } + budget = budget.saturating_sub(bytes.len() as u64); + Ok(bytes) + }; + let content = if entry.post_snapshot_oversized { + Err(io::Error::new( + io::ErrorKind::FileTooLarge, + "Snapshot exceeds review limit", + )) + } else { + read(entry.pre_blob.as_deref()).and_then(|before| { + read(entry.post_hash.as_deref()).map(|after| (before, after)) + }) + }; + let (availability, before_content, after_content) = match content { + Ok((before, after)) if before.contains(&0) || after.contains(&0) => { + ("binary", None, None) + } + Ok((before, after)) => { + match (String::from_utf8(before), String::from_utf8(after)) { + (Ok(before), Ok(after)) => ("patch", Some(before), Some(after)), + _ => ("binary", None, None), + } + } + Err(error) if error.kind() == io::ErrorKind::FileTooLarge => { + ("truncated", None, None) + } + Err(_) => ("unavailable", None, None), + }; + files.push(TurnSnapshotFile { + path: entry.path, + kind: entry.kind, + availability, + before_content, + after_content, + }); + } + Ok(Some(( + checkpoint.user_turn_index.expect("selected linked turn"), + files, + ))) + } +} + +fn expand_native_untracked(root: &Path, paths: &HashSet) -> Option> { + let mut expanded = HashSet::new(); + for path in paths { + if path.ends_with('/') { + let mut files = Vec::new(); + collect_files(&root.join(path), root, &mut files); + expanded.extend(files.into_iter().map(|(relative, _)| relative)); + } else { + expanded.insert(path.clone()); + } + if expanded.len() > MAX_EXPANDED_UNTRACKED_FILES { + return None; + } + } + Some(expanded) +} + +fn snapshot_file( + cp_dir: &Path, + path: &Path, + retain: bool, + budget: &mut u64, +) -> io::Result<(Option, bool)> { + if !retain { + return file_hash(path).map(|hash| (hash, false)); + } + let Some(hash) = file_hash(path)? else { + return Ok((None, false)); + }; + // Hashing remains streaming. Retention never allocates an unbounded file. + let mut bytes = Vec::new(); + fs::File::open(path)? + .take(*budget + 1) + .read_to_end(&mut bytes)?; + if bytes.len() as u64 > *budget { + return Ok((Some(hash), true)); + } + if sha256(&bytes) != hash { + return Err(io::Error::other("File changed during checkpoint capture")); + } + store_blob(cp_dir, &bytes)?; + *budget = budget.saturating_sub(bytes.len() as u64); + Ok((Some(hash), false)) +} + /// Outcome of restoring one checkpoint. #[derive(Debug, Default)] pub struct RestoreReport { @@ -631,6 +958,20 @@ pub fn fork_before_turn( crate::fs_atomic::create_dir_all_synced(target_dir.join("blobs"))?; store_blob(&target_dir, &bytes)?; } + for hash in checkpoint + .entries + .iter() + .filter_map(|entry| entry.post_hash.as_ref()) + { + if source_dir.join("blobs").join(hash).is_file() { + let bytes = read_blob(&source_dir, hash)?; + if sha256(&bytes) != *hash { + return Err(io::Error::other("Retained checkpoint content is corrupt.")); + } + crate::fs_atomic::create_dir_all_synced(target_dir.join("blobs"))?; + store_blob(&target_dir, &bytes)?; + } + } target.save(&checkpoint)?; } Ok(()) @@ -1249,6 +1590,231 @@ mod tests { } } + #[test] + fn native_turn_snapshot_uses_retained_content_and_isolates_turn_session_workspace() { + let fx = git_fixture(); + let mut pending = begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "edit").unwrap(); + pending.user_turn_index = Some(0); + fs::write(fx.repo.join("a.rs"), "first saved change\n").unwrap(); + finalize_turn_snapshot(pending).unwrap().unwrap(); + fs::write(fx.repo.join("a.rs"), "later unrelated content\n").unwrap(); + let (index, files) = store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap(); + assert_eq!(index, 0); + assert_eq!( + files[0].after_content.as_deref(), + Some("first saved change\n") + ); + assert_eq!(files[0].availability, "patch"); + assert!( + store(&fx) + .turn_snapshot(Some(1), &fx.repo) + .unwrap() + .is_none() + ); + assert!( + CheckpointStore::new(&fx.sessions, "another-session") + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .is_none() + ); + let other = git_fixture(); + assert!( + store(&fx) + .turn_snapshot(Some(0), &other.repo) + .unwrap() + .is_none() + ); + } + + #[test] + fn native_empty_turn_is_saved_and_abandoned_capture_cleans_up() { + let fx = git_fixture(); + let mut pending = + begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "empty").unwrap(); + pending.user_turn_index = Some(0); + finalize_turn_snapshot(pending).unwrap().unwrap(); + assert!( + store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap() + .1 + .is_empty() + ); + let pending = begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "abandon").unwrap(); + let directory = pending.store.new_checkpoint_dir(&pending.id); + drop(pending); + assert!(!directory.exists()); + } + + #[test] + fn native_snapshot_missing_binary_corrupt_and_oversized_stay_distinct() { + let fx = git_fixture(); + let mut pending = + begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "binary").unwrap(); + pending.user_turn_index = Some(0); + fs::write(fx.repo.join("new.bin"), [0, 1, 2]).unwrap(); + let checkpoint = finalize_turn_snapshot(pending).unwrap().unwrap(); + let files = store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap() + .1; + assert_eq!(files[0].availability, "binary"); + let hash = checkpoint.entries[0].post_hash.as_ref().unwrap(); + let blob = store(&fx) + .new_checkpoint_dir(&checkpoint.id) + .join("blobs") + .join(hash); + fs::write(&blob, "corrupt").unwrap(); + assert_eq!( + store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap() + .1[0] + .availability, + "unavailable" + ); + fs::remove_file(blob).unwrap(); + assert_eq!( + store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap() + .1[0] + .availability, + "unavailable" + ); + let mut pending = + begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "large").unwrap(); + pending.user_turn_index = Some(1); + fs::write(fx.repo.join("large.bin"), vec![b'x'; 512 * 1024 + 1]).unwrap(); + let checkpoint = finalize_turn_snapshot(pending).unwrap().unwrap(); + assert!(checkpoint.entries[0].post_hash.is_some()); + assert!(checkpoint.entries[0].post_snapshot_oversized); + assert_eq!( + store(&fx) + .turn_snapshot(Some(1), &fx.repo) + .unwrap() + .unwrap() + .1[0] + .availability, + "truncated" + ); + assert!( + !store(&fx) + .new_checkpoint_dir(&checkpoint.id) + .join("blobs") + .join(checkpoint.entries[0].post_hash.as_ref().unwrap()) + .exists() + ); + } + + #[test] + fn native_snapshot_captures_existing_and_new_untracked_files() { + let fx = git_fixture(); + fs::create_dir(fx.repo.join("drafts")).unwrap(); + fs::write(fx.repo.join("drafts/existing.txt"), "before draft").unwrap(); + fs::write(fx.repo.join("stable.txt"), "unchanged").unwrap(); + let mut pending = + begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "draft").unwrap(); + pending.user_turn_index = Some(0); + fs::write(fx.repo.join("drafts/existing.txt"), "after draft").unwrap(); + fs::write(fx.repo.join("drafts/new.txt"), "new draft").unwrap(); + finalize_turn_snapshot(pending).unwrap().unwrap(); + let files = store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap() + .1; + assert_eq!( + files + .iter() + .map(|file| file.path.as_str()) + .collect::>(), + ["drafts/existing.txt", "drafts/new.txt"] + ); + assert_eq!(files[0].before_content.as_deref(), Some("before draft")); + assert_eq!(files[0].after_content.as_deref(), Some("after draft")); + assert_eq!(files[1].before_content.as_deref(), Some("")); + } + + #[test] + fn native_snapshot_reads_edit_and_commit_from_saved_start_head() { + let fx = git_fixture(); + let mut pending = + begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "commit").unwrap(); + pending.user_turn_index = Some(0); + let original = fs::read_to_string(fx.repo.join("a.rs")).unwrap(); + fs::write(fx.repo.join("a.rs"), "committed new content\n").unwrap(); + run_git(&fx.repo, &["add", "a.rs"]); + run_git(&fx.repo, &["commit", "--quiet", "-m", "turn change"]); + finalize_turn_snapshot(pending).unwrap().unwrap(); + let files = store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap() + .1; + assert_eq!(files[0].before_content.as_deref(), Some(original.as_str())); + assert_eq!( + files[0].after_content.as_deref(), + Some("committed new content\n") + ); + } + + #[test] + fn native_post_content_budget_is_shared_across_files() { + let fx = git_fixture(); + let mut pending = + begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "several files").unwrap(); + pending.user_turn_index = Some(0); + for name in ["large-a.txt", "large-b.txt", "large-c.txt"] { + fs::write(fx.repo.join(name), vec![b'x'; 256 * 1024]).unwrap(); + } + let checkpoint = finalize_turn_snapshot(pending).unwrap().unwrap(); + // Identical hashes can share disk content, but each retained pre/post + // pair still consumes the capture budget and omitted entries are typed. + assert!( + checkpoint + .entries + .iter() + .any(|entry| entry.post_snapshot_oversized) + ); + let total = fs::read_dir(store(&fx).new_checkpoint_dir(&checkpoint.id).join("blobs")) + .unwrap() + .map(|entry| entry.unwrap().metadata().unwrap().len()) + .sum::(); + assert!(total <= 512 * 1024); + assert!( + store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .unwrap() + .1 + .iter() + .any(|file| file.availability == "truncated") + ); + } + + #[test] + fn native_large_dirty_preimage_does_not_publish_pristine_turn() { + let fx = git_fixture(); + fs::write(fx.repo.join("a.rs"), vec![b'x'; 512 * 1024 + 1]).unwrap(); + let pending = + begin_turn_snapshot(&fx.repo, &fx.sessions, "session-1", "large dirty").unwrap(); + assert!(finalize_turn_snapshot(pending).is_err()); + assert!( + store(&fx) + .turn_snapshot(Some(0), &fx.repo) + .unwrap() + .is_none() + ); + } + #[test] fn finalized_checkpoint_keeps_only_referenced_blobs() { let fx = git_fixture(); diff --git a/packages/tui-rs/src/app/tests.rs b/packages/tui-rs/src/app/tests.rs index 69426a7cc..28d795fe3 100644 --- a/packages/tui-rs/src/app/tests.rs +++ b/packages/tui-rs/src/app/tests.rs @@ -3596,337 +3596,6 @@ async fn loop_command_start_and_stop_via_handler() { assert!(app.loop_schedule.is_none()); } -// ───────────────────────────────────────────────────────────────────────── -// Rewind Picker Tests (double-Esc on empty input) -// ───────────────────────────────────────────────────────────────────────── - -/// Point the app's session manager at a temp sessions dir and give it a -/// session id so file checkpoints resolve to the fixture, not `$HOME`. -fn setup_rewind_session(app: &mut App) -> (tempfile::TempDir, std::path::PathBuf) { - let tmp = tempdir().unwrap(); - let repo = tmp.path().join("repo"); - let sessions = tmp.path().join("sessions"); - std::fs::create_dir_all(&repo).unwrap(); - std::fs::create_dir_all(&sessions).unwrap(); - app.session_manager = SessionManager::with_sessions_dir( - repo.to_string_lossy().to_string(), - sessions.to_string_lossy().to_string(), - ); - app.state.session_id = Some("rewind-test".to_string()); - (tmp, repo) -} - -/// Write a one-file checkpoint manifest (plus its pre-turn blob) directly -/// into the session's checkpoint store and set the file's current content. -fn write_rewind_checkpoint( - app: &App, - repo: &std::path::Path, - id: &str, - created_at: &str, - pre: &str, - post: &str, -) { - use sha2::{Digest, Sha256}; - let hash = |content: &str| format!("{:x}", Sha256::digest(content.as_bytes())); - - let store = - crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); - let dir = store.root().join(id); - std::fs::create_dir_all(dir.join("blobs")).unwrap(); - std::fs::write(dir.join("blobs").join(hash(pre)), pre).unwrap(); - std::fs::write(repo.join("a.rs"), post).unwrap(); - - let checkpoint = crate::checkpoints::Checkpoint { - id: id.to_string(), - created_at: created_at.to_string(), - prompt: format!("prompt for {id}"), - repo_root: dunce::canonicalize(repo).unwrap(), - head: None, - user_turn_index: None, - entries: vec![crate::checkpoints::FileEntry { - path: "a.rs".to_string(), - kind: crate::checkpoints::EntryKind::Modified, - pre_blob: Some(hash(pre)), - post_hash: Some(hash(post)), - }], - }; - let bytes = serde_json::to_vec_pretty(&checkpoint).unwrap(); - std::fs::write(dir.join("checkpoint.json"), bytes).unwrap(); -} - -async fn press_esc(app: &mut App) { - app.handle_key(KeyCode::Esc, CrosstermModifiers::NONE) - .await - .unwrap(); -} - -#[test] -fn ephemeral_sessions_keep_file_checkpoints_without_turn_coordinates() { - let mut app = new_test_app(); - let (_temp, repo) = setup_rewind_session(&mut app); - assert!( - std::process::Command::new("git") - .args(["init", "--quiet"]) - .current_dir(&repo) - .status() - .unwrap() - .success() - ); - assert!(app.session_manager.writer().is_none()); - app.capture_file_checkpoint(&repo, "ephemeral edit"); - assert!( - app.pending_checkpoint - .as_ref() - .is_some_and(|pending| pending.user_turn_index.is_none()) - ); - std::fs::write(repo.join("new.txt"), "created in ephemeral turn").unwrap(); - app.finalize_file_checkpoint(); - let store = - crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); - assert_eq!(store.list().len(), 1); - let restored = crate::checkpoints::restore_latest(&store).unwrap().unwrap(); - assert!(restored.failed.is_empty()); - assert!(!repo.join("new.txt").exists()); -} - -#[test] -fn rewind_both_reopens_before_two_edits_and_keeps_manual_changes() { - let mut app = new_test_app(); - let (_tmp, repo) = setup_rewind_session(&mut app); - app.state.session_id = None; - app.record_user_message("first edit"); - app.record_user_message("second edit"); - app.session_manager.flush().unwrap(); - let source_id = app.state.session_id.clone().unwrap(); - // The fixture helper writes to rewind-test; move its completed store to - // this real saved session after assigning explicit turn coordinates. - write_rewind_checkpoint(&app, &repo, "cp-0", "2026-07-24T00:00:00Z", "v0", "v1"); - write_rewind_checkpoint(&app, &repo, "cp-1", "2026-07-24T01:00:00Z", "v1", "v2"); - let fixture_store = - crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); - for index in 0..2 { - let manifest = fixture_store - .root() - .join(format!("cp-{index}/checkpoint.json")); - let mut value: serde_json::Value = - serde_json::from_slice(&std::fs::read(&manifest).unwrap()).unwrap(); - value["user_turn_index"] = index.into(); - let mut second_file = value["entries"][0].clone(); - second_file["path"] = "b.rs".into(); - value["entries"].as_array_mut().unwrap().push(second_file); - std::fs::write(manifest, serde_json::to_vec(&value).unwrap()).unwrap(); - } - let source_store = - crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), &source_id); - std::fs::rename(fixture_store.root(), source_store.root()).unwrap(); - std::fs::write(repo.join("a.rs"), "later manual edit").unwrap(); - std::fs::write(repo.join("b.rs"), "v2").unwrap(); - app.rewind_saved_turns(2, false, true); - assert!(app.state.error.is_none(), "{:?}", app.state.error); - assert_ne!(app.state.session_id.as_deref(), Some(source_id.as_str())); - let branch = app.session_manager.current_session_path().unwrap(); - app.session_manager.flush().unwrap(); - let reopened = crate::session::SessionReader::read_file(branch).unwrap(); - assert_eq!(reopened.stats.user_messages, 0); - assert_eq!( - std::fs::read_to_string(repo.join("a.rs")).unwrap(), - "later manual edit" - ); - assert_eq!(std::fs::read_to_string(repo.join("b.rs")).unwrap(), "v0"); -} - -#[test] -fn rewind_saved_conversation_reopens_at_selected_turn_and_keeps_source() { - let mut app = new_test_app(); - let (_tmp, _) = setup_rewind_session(&mut app); - app.state.session_id = None; - app.record_user_message("first request"); - app.record_user_message("abandoned request"); - app.session_manager.flush().unwrap(); - let source_path = app.session_manager.current_session_path().unwrap(); - let original = std::fs::read(&source_path).unwrap(); - let source_id = app.state.session_id.clone(); - app.rewind_turns(1, true); - assert_eq!(app.state.session_id, source_id); - assert_eq!(std::fs::read(&source_path).unwrap(), original); - app.rewind_turns(1, false); - assert!(app.state.error.is_none(), "{:?}", app.state.error); - assert_ne!(app.state.session_id, source_id); - app.record_user_message("replacement request"); - app.session_manager.flush().unwrap(); - let branch = app.session_manager.current_session_path().unwrap(); - let reopened = crate::session::SessionReader::read_file(&branch).unwrap(); - assert_eq!( - reopened - .messages - .iter() - .map(AppMessage::text_content) - .collect::>(), - ["first request", "replacement request"] - ); - assert_eq!(std::fs::read(&source_path).unwrap(), original); - let (history, _, _) = app.agent_context_for_spawn().unwrap(); - assert_eq!(history.len(), 2); - assert_eq!(history[1].content.as_text(), Some("replacement request")); -} - -#[tokio::test] -async fn double_esc_on_empty_input_opens_rewind_picker() { - let mut app = new_test_app(); - let (_tmp, repo) = setup_rewind_session(&mut app); - write_rewind_checkpoint( - &app, - &repo, - "cp-1", - "2026-07-24T00:00:00Z", - "original\n", - "edit\n", - ); - - press_esc(&mut app).await; - assert_eq!(app.active_modal, ActiveModal::None); - assert_eq!( - app.state.status.as_deref(), - Some("Press Esc again to rewind files") - ); - - press_esc(&mut app).await; - assert_eq!(app.active_modal, ActiveModal::RewindPicker); - assert!(app.rewind_picker.is_visible()); -} - -#[test] -fn rewind_picker_conversation_uses_the_saved_turn_coordinate() { - let mut app = new_test_app(); - let (_temp, repo) = setup_rewind_session(&mut app); - app.state.session_id = None; - for prompt in ["kept first request", "second request", "third request"] { - app.record_user_message(prompt); - } - app.session_manager.flush().unwrap(); - let source = app.session_manager.current_session_path().unwrap(); - let original = std::fs::read(&source).unwrap(); - app.rewind_picker.show(vec![crate::checkpoints::Checkpoint { - id: "selected".into(), - created_at: "2026-09-10T00:00:00Z".into(), - prompt: "second request".into(), - repo_root: repo, - head: None, - user_turn_index: Some(1), - entries: vec![], - }]); - app.handle_rewind_picker_key(KeyCode::Char('c')).unwrap(); - assert!(app.state.error.is_none(), "{:?}", app.state.error); - app.session_manager.flush().unwrap(); - let child = crate::session::SessionReader::read_file( - app.session_manager.current_session_path().unwrap(), - ) - .unwrap(); - assert_eq!(child.stats.user_messages, 1); - assert_eq!(std::fs::read(source).unwrap(), original); -} - -#[tokio::test] -async fn rewind_picker_enter_restores_selected_checkpoint() { - let mut app = new_test_app(); - let (_tmp, repo) = setup_rewind_session(&mut app); - write_rewind_checkpoint(&app, &repo, "cp-1", "2026-07-24T00:00:00Z", "v0\n", "v1\n"); - write_rewind_checkpoint(&app, &repo, "cp-2", "2026-07-24T01:00:00Z", "v1\n", "v2\n"); - - press_esc(&mut app).await; - press_esc(&mut app).await; - assert_eq!(app.active_modal, ActiveModal::RewindPicker); - - // Newest checkpoint is selected by default; Enter restores it. - app.handle_key(KeyCode::Enter, CrosstermModifiers::NONE) - .await - .unwrap(); - - assert_eq!(app.active_modal, ActiveModal::None); - assert!(!app.rewind_picker.is_visible()); - assert_eq!(std::fs::read_to_string(repo.join("a.rs")).unwrap(), "v1\n"); - assert_eq!( - app.state.status.as_deref(), - Some("Files restored from checkpoint.") - ); - - // Only the applied checkpoint was consumed; the older one remains. - let store = - crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); - let remaining = store.list(); - assert_eq!(remaining.len(), 1); - assert_eq!(remaining[0].id, "cp-1"); -} - -#[tokio::test] -async fn rewind_picker_esc_dismisses_without_restoring() { - let mut app = new_test_app(); - let (_tmp, repo) = setup_rewind_session(&mut app); - write_rewind_checkpoint( - &app, - &repo, - "cp-1", - "2026-07-24T00:00:00Z", - "original\n", - "edit\n", - ); - - press_esc(&mut app).await; - press_esc(&mut app).await; - assert_eq!(app.active_modal, ActiveModal::RewindPicker); - - press_esc(&mut app).await; - assert_eq!(app.active_modal, ActiveModal::None); - assert!(!app.rewind_picker.is_visible()); - // Nothing was restored and the checkpoint was not consumed. - assert_eq!( - std::fs::read_to_string(repo.join("a.rs")).unwrap(), - "edit\n" - ); - let store = - crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); - assert_eq!(store.list().len(), 1); -} - -#[tokio::test] -async fn double_esc_without_checkpoints_shows_status() { - let mut app = new_test_app(); - let (_tmp, _repo) = setup_rewind_session(&mut app); - - press_esc(&mut app).await; - press_esc(&mut app).await; - - assert_eq!(app.active_modal, ActiveModal::None); - assert_eq!( - app.state.status.as_deref(), - Some("No file checkpoints recorded for this session.") - ); -} - -#[tokio::test] -async fn double_esc_rewind_picker_blocked_while_busy() { - let mut app = new_test_app(); - let (_tmp, repo) = setup_rewind_session(&mut app); - write_rewind_checkpoint( - &app, - &repo, - "cp-1", - "2026-07-24T00:00:00Z", - "original\n", - "edit\n", - ); - app.state.busy = true; - - press_esc(&mut app).await; - press_esc(&mut app).await; - - assert_eq!(app.active_modal, ActiveModal::None); - assert_eq!( - app.state.status.as_deref(), - Some("Wait for the active response to finish before rewinding.") - ); -} - // ───────────────────────────────────────────────────────────────────────── // Detail View Tests (Ctrl+E) // ───────────────────────────────────────────────────────────────────────── @@ -7025,3 +6694,5 @@ fn catalog_typing_does_not_trigger_manager_actions() { mod settings_experiments; mod session_restore; + +mod checkpoints; diff --git a/packages/tui-rs/src/app/tests/checkpoints.rs b/packages/tui-rs/src/app/tests/checkpoints.rs new file mode 100644 index 000000000..f79418f6f --- /dev/null +++ b/packages/tui-rs/src/app/tests/checkpoints.rs @@ -0,0 +1,333 @@ +use super::*; + +// ───────────────────────────────────────────────────────────────────────── +// Rewind Picker Tests (double-Esc on empty input) +// ───────────────────────────────────────────────────────────────────────── + +/// Point the app's session manager at a temp sessions dir and give it a +/// session id so file checkpoints resolve to the fixture, not `$HOME`. +fn setup_rewind_session(app: &mut App) -> (tempfile::TempDir, std::path::PathBuf) { + let tmp = tempdir().unwrap(); + let repo = tmp.path().join("repo"); + let sessions = tmp.path().join("sessions"); + std::fs::create_dir_all(&repo).unwrap(); + std::fs::create_dir_all(&sessions).unwrap(); + app.session_manager = SessionManager::with_sessions_dir( + repo.to_string_lossy().to_string(), + sessions.to_string_lossy().to_string(), + ); + app.state.session_id = Some("rewind-test".to_string()); + (tmp, repo) +} + +/// Write a one-file checkpoint manifest (plus its pre-turn blob) directly +/// into the session's checkpoint store and set the file's current content. +fn write_rewind_checkpoint( + app: &App, + repo: &std::path::Path, + id: &str, + created_at: &str, + pre: &str, + post: &str, +) { + use sha2::{Digest, Sha256}; + let hash = |content: &str| format!("{:x}", Sha256::digest(content.as_bytes())); + + let store = + crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); + let dir = store.root().join(id); + std::fs::create_dir_all(dir.join("blobs")).unwrap(); + std::fs::write(dir.join("blobs").join(hash(pre)), pre).unwrap(); + std::fs::write(repo.join("a.rs"), post).unwrap(); + + let checkpoint = crate::checkpoints::Checkpoint { + id: id.to_string(), + created_at: created_at.to_string(), + prompt: format!("prompt for {id}"), + repo_root: dunce::canonicalize(repo).unwrap(), + head: None, + user_turn_index: None, + entries: vec![crate::checkpoints::FileEntry { + path: "a.rs".to_string(), + kind: crate::checkpoints::EntryKind::Modified, + pre_blob: Some(hash(pre)), + post_hash: Some(hash(post)), + post_snapshot_oversized: false, + }], + }; + let bytes = serde_json::to_vec_pretty(&checkpoint).unwrap(); + std::fs::write(dir.join("checkpoint.json"), bytes).unwrap(); +} + +async fn press_esc(app: &mut App) { + app.handle_key(KeyCode::Esc, CrosstermModifiers::NONE) + .await + .unwrap(); +} + +#[test] +fn ephemeral_sessions_keep_file_checkpoints_without_turn_coordinates() { + let mut app = new_test_app(); + let (_temp, repo) = setup_rewind_session(&mut app); + assert!( + std::process::Command::new("git") + .args(["init", "--quiet"]) + .current_dir(&repo) + .status() + .unwrap() + .success() + ); + assert!(app.session_manager.writer().is_none()); + app.capture_file_checkpoint(&repo, "ephemeral edit"); + assert!( + app.pending_checkpoint + .as_ref() + .is_some_and(|pending| pending.user_turn_index.is_none()) + ); + std::fs::write(repo.join("new.txt"), "created in ephemeral turn").unwrap(); + app.finalize_file_checkpoint(); + let store = + crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); + assert_eq!(store.list().len(), 1); + let restored = crate::checkpoints::restore_latest(&store).unwrap().unwrap(); + assert!(restored.failed.is_empty()); + assert!(!repo.join("new.txt").exists()); +} + +#[test] +fn rewind_both_reopens_before_two_edits_and_keeps_manual_changes() { + let mut app = new_test_app(); + let (_tmp, repo) = setup_rewind_session(&mut app); + app.state.session_id = None; + app.record_user_message("first edit"); + app.record_user_message("second edit"); + app.session_manager.flush().unwrap(); + let source_id = app.state.session_id.clone().unwrap(); + // The fixture helper writes to rewind-test; move its completed store to + // this real saved session after assigning explicit turn coordinates. + write_rewind_checkpoint(&app, &repo, "cp-0", "2026-07-24T00:00:00Z", "v0", "v1"); + write_rewind_checkpoint(&app, &repo, "cp-1", "2026-07-24T01:00:00Z", "v1", "v2"); + let fixture_store = + crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); + for index in 0..2 { + let manifest = fixture_store + .root() + .join(format!("cp-{index}/checkpoint.json")); + let mut value: serde_json::Value = + serde_json::from_slice(&std::fs::read(&manifest).unwrap()).unwrap(); + value["user_turn_index"] = index.into(); + let mut second_file = value["entries"][0].clone(); + second_file["path"] = "b.rs".into(); + value["entries"].as_array_mut().unwrap().push(second_file); + std::fs::write(manifest, serde_json::to_vec(&value).unwrap()).unwrap(); + } + let source_store = + crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), &source_id); + std::fs::rename(fixture_store.root(), source_store.root()).unwrap(); + std::fs::write(repo.join("a.rs"), "later manual edit").unwrap(); + std::fs::write(repo.join("b.rs"), "v2").unwrap(); + app.rewind_saved_turns(2, false, true); + assert!(app.state.error.is_none(), "{:?}", app.state.error); + assert_ne!(app.state.session_id.as_deref(), Some(source_id.as_str())); + let branch = app.session_manager.current_session_path().unwrap(); + app.session_manager.flush().unwrap(); + let reopened = crate::session::SessionReader::read_file(branch).unwrap(); + assert_eq!(reopened.stats.user_messages, 0); + assert_eq!( + std::fs::read_to_string(repo.join("a.rs")).unwrap(), + "later manual edit" + ); + assert_eq!(std::fs::read_to_string(repo.join("b.rs")).unwrap(), "v0"); +} + +#[test] +fn rewind_saved_conversation_reopens_at_selected_turn_and_keeps_source() { + let mut app = new_test_app(); + let (_tmp, _) = setup_rewind_session(&mut app); + app.state.session_id = None; + app.record_user_message("first request"); + app.record_user_message("abandoned request"); + app.session_manager.flush().unwrap(); + let source_path = app.session_manager.current_session_path().unwrap(); + let original = std::fs::read(&source_path).unwrap(); + let source_id = app.state.session_id.clone(); + app.rewind_turns(1, true); + assert_eq!(app.state.session_id, source_id); + assert_eq!(std::fs::read(&source_path).unwrap(), original); + app.rewind_turns(1, false); + assert!(app.state.error.is_none(), "{:?}", app.state.error); + assert_ne!(app.state.session_id, source_id); + app.record_user_message("replacement request"); + app.session_manager.flush().unwrap(); + let branch = app.session_manager.current_session_path().unwrap(); + let reopened = crate::session::SessionReader::read_file(&branch).unwrap(); + assert_eq!( + reopened + .messages + .iter() + .map(AppMessage::text_content) + .collect::>(), + ["first request", "replacement request"] + ); + assert_eq!(std::fs::read(&source_path).unwrap(), original); + let (history, _, _) = app.agent_context_for_spawn().unwrap(); + assert_eq!(history.len(), 2); + assert_eq!(history[1].content.as_text(), Some("replacement request")); +} + +#[tokio::test] +async fn double_esc_on_empty_input_opens_rewind_picker() { + let mut app = new_test_app(); + let (_tmp, repo) = setup_rewind_session(&mut app); + write_rewind_checkpoint( + &app, + &repo, + "cp-1", + "2026-07-24T00:00:00Z", + "original\n", + "edit\n", + ); + + press_esc(&mut app).await; + assert_eq!(app.active_modal, ActiveModal::None); + assert_eq!( + app.state.status.as_deref(), + Some("Press Esc again to rewind files") + ); + + press_esc(&mut app).await; + assert_eq!(app.active_modal, ActiveModal::RewindPicker); + assert!(app.rewind_picker.is_visible()); +} + +#[test] +fn rewind_picker_conversation_uses_the_saved_turn_coordinate() { + let mut app = new_test_app(); + let (_temp, repo) = setup_rewind_session(&mut app); + app.state.session_id = None; + for prompt in ["kept first request", "second request", "third request"] { + app.record_user_message(prompt); + } + app.session_manager.flush().unwrap(); + let source = app.session_manager.current_session_path().unwrap(); + let original = std::fs::read(&source).unwrap(); + app.rewind_picker.show(vec![crate::checkpoints::Checkpoint { + id: "selected".into(), + created_at: "2026-09-10T00:00:00Z".into(), + prompt: "second request".into(), + repo_root: repo, + head: None, + user_turn_index: Some(1), + entries: vec![], + }]); + app.handle_rewind_picker_key(KeyCode::Char('c')).unwrap(); + assert!(app.state.error.is_none(), "{:?}", app.state.error); + app.session_manager.flush().unwrap(); + let child = crate::session::SessionReader::read_file( + app.session_manager.current_session_path().unwrap(), + ) + .unwrap(); + assert_eq!(child.stats.user_messages, 1); + assert_eq!(std::fs::read(source).unwrap(), original); +} + +#[tokio::test] +async fn rewind_picker_enter_restores_selected_checkpoint() { + let mut app = new_test_app(); + let (_tmp, repo) = setup_rewind_session(&mut app); + write_rewind_checkpoint(&app, &repo, "cp-1", "2026-07-24T00:00:00Z", "v0\n", "v1\n"); + write_rewind_checkpoint(&app, &repo, "cp-2", "2026-07-24T01:00:00Z", "v1\n", "v2\n"); + + press_esc(&mut app).await; + press_esc(&mut app).await; + assert_eq!(app.active_modal, ActiveModal::RewindPicker); + + // Newest checkpoint is selected by default; Enter restores it. + app.handle_key(KeyCode::Enter, CrosstermModifiers::NONE) + .await + .unwrap(); + + assert_eq!(app.active_modal, ActiveModal::None); + assert!(!app.rewind_picker.is_visible()); + assert_eq!(std::fs::read_to_string(repo.join("a.rs")).unwrap(), "v1\n"); + assert_eq!( + app.state.status.as_deref(), + Some("Files restored from checkpoint.") + ); + + // Only the applied checkpoint was consumed; the older one remains. + let store = + crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); + let remaining = store.list(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].id, "cp-1"); +} + +#[tokio::test] +async fn rewind_picker_esc_dismisses_without_restoring() { + let mut app = new_test_app(); + let (_tmp, repo) = setup_rewind_session(&mut app); + write_rewind_checkpoint( + &app, + &repo, + "cp-1", + "2026-07-24T00:00:00Z", + "original\n", + "edit\n", + ); + + press_esc(&mut app).await; + press_esc(&mut app).await; + assert_eq!(app.active_modal, ActiveModal::RewindPicker); + + press_esc(&mut app).await; + assert_eq!(app.active_modal, ActiveModal::None); + assert!(!app.rewind_picker.is_visible()); + // Nothing was restored and the checkpoint was not consumed. + assert_eq!( + std::fs::read_to_string(repo.join("a.rs")).unwrap(), + "edit\n" + ); + let store = + crate::checkpoints::CheckpointStore::new(app.session_manager.sessions_dir(), "rewind-test"); + assert_eq!(store.list().len(), 1); +} + +#[tokio::test] +async fn double_esc_without_checkpoints_shows_status() { + let mut app = new_test_app(); + let (_tmp, _repo) = setup_rewind_session(&mut app); + + press_esc(&mut app).await; + press_esc(&mut app).await; + + assert_eq!(app.active_modal, ActiveModal::None); + assert_eq!( + app.state.status.as_deref(), + Some("No file checkpoints recorded for this session.") + ); +} + +#[tokio::test] +async fn double_esc_rewind_picker_blocked_while_busy() { + let mut app = new_test_app(); + let (_tmp, repo) = setup_rewind_session(&mut app); + write_rewind_checkpoint( + &app, + &repo, + "cp-1", + "2026-07-24T00:00:00Z", + "original\n", + "edit\n", + ); + app.state.busy = true; + + press_esc(&mut app).await; + press_esc(&mut app).await; + + assert_eq!(app.active_modal, ActiveModal::None); + assert_eq!( + app.state.status.as_deref(), + Some("Wait for the active response to finish before rewinding.") + ); +} diff --git a/packages/tui-rs/src/components/rewind_picker.rs b/packages/tui-rs/src/components/rewind_picker.rs index 4c4715513..bc007fe13 100644 --- a/packages/tui-rs/src/components/rewind_picker.rs +++ b/packages/tui-rs/src/components/rewind_picker.rs @@ -176,6 +176,7 @@ mod tests { kind: EntryKind::Modified, pre_blob: Some("pre".to_string()), post_hash: Some("post".to_string()), + post_snapshot_oversized: false, }) .collect(), } diff --git a/test/fixtures/rust-cutover/web-routes.json b/test/fixtures/rust-cutover/web-routes.json index 537806878..4850dcffb 100644 --- a/test/fixtures/rust-cutover/web-routes.json +++ b/test/fixtures/rust-cutover/web-routes.json @@ -143,6 +143,7 @@ { "method": "POST", "path": "/api/chat/client-tool-result" }, { "method": "POST", "path": "/api/chat/tool-retry" }, { "method": "POST", "path": "/api/attachments/extract" }, + { "method": "GET", "path": "/api/sessions/:id/turn-diff" }, { "method": "GET", "path": "/api/sessions/:id/artifacts" }, { "method": "GET", "path": "/api/sessions/:id/artifact-access" }, { diff --git a/third-party/t3code/README.md b/third-party/t3code/README.md index 2465ac86e..1735da840 100644 --- a/third-party/t3code/README.md +++ b/third-party/t3code/README.md @@ -1,9 +1,19 @@ -# T3 Code reliability adaptations +# T3 Code adaptations Source: https://github.com/pingdotgg/t3code -Reviewed revision: `4f7760e6a0037b06917adaae1b8a220e3ee6e5cc`. +Reviewed revisions: -Maestro adapts the buffer budgeting, bounded diagnostic copies, provider capability checks, connection lifecycle coverage, lazy diff worker lifetime, and checkpoint overlap scenarios into its existing owners. The upstream application and runtime are not imported. The original MIT copyright and license are retained in `LICENSE` for these adaptations. +- `4f7760e6a0037b06917adaae1b8a220e3ee6e5cc`: bounded event buffers and diagnostics, capability enforcement, reconnect lifetime, lazy diff workers, and checkpoint overlap tests. +- `a5b34b25378fbbf90ce36ee101a11528ccf11c7e`: bounded terminal output windows, checkpoint-backed turn review, and deterministic provider restart replay scenarios. -The native session runtime remains Rust. Checkpoint restoration retains Maestro's per-file content guards; the new overlap tests protect later edits without introducing a blanket shared-workspace ban. +Maestro adapts these mechanisms into its existing desktop, session, gateway, and native provider owners. The upstream application and TypeScript server runtime are not imported. + +Second-batch references: + +- `apps/server/src/terminal/OutputProtocol.ts` +- `apps/server/src/checkpointing/CheckpointDiffQuery.ts` +- `apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts` +- `apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts` + +The original MIT copyright and license are retained in `LICENSE`. Provider fixtures authored for Maestro are synthetic, scrubbed protocol scenarios; they do not claim to be recordings of live upstream or customer sessions.