diff --git a/apps/android/app/build.gradle.kts b/apps/android/app/build.gradle.kts index 78f88dce7..5e4f4b8cb 100644 --- a/apps/android/app/build.gradle.kts +++ b/apps/android/app/build.gradle.kts @@ -33,7 +33,7 @@ android { applicationId = "com.sigkitten.litter.android" minSdk = 26 targetSdk = 36 - versionCode = 200000254 + versionCode = 200000260 versionName = "2.0.0" buildConfigField("boolean", "ENABLE_ON_DEVICE_BRIDGE", "true") buildConfigField("String", "RUNTIME_STARTUP_MODE", "\"hybrid\"") diff --git a/apps/ios/Litter.xcodeproj/project.pbxproj b/apps/ios/Litter.xcodeproj/project.pbxproj index e6bcbbf2e..b16421680 100644 --- a/apps/ios/Litter.xcodeproj/project.pbxproj +++ b/apps/ios/Litter.xcodeproj/project.pbxproj @@ -2909,7 +2909,7 @@ CLANG_WARN_UNREACHABLE_CODE = YES; CLANG_WARN__DUPLICATE_METHOD_MATCH = YES; COPY_PHASE_STRIP = NO; - CURRENT_PROJECT_VERSION = 200000254; + CURRENT_PROJECT_VERSION = 200000260; DEBUG_INFORMATION_FORMAT = "dwarf-with-dsym"; DEVELOPMENT_TEAM = TZ447KHNZL; ENABLE_NS_ASSERTIONS = NO; @@ -3053,7 +3053,7 @@ CLANG_WARN_UNREACHABLE_CODE = YES; CLANG_WARN__DUPLICATE_METHOD_MATCH = YES; COPY_PHASE_STRIP = NO; - CURRENT_PROJECT_VERSION = 200000254; + CURRENT_PROJECT_VERSION = 200000260; DEBUG_INFORMATION_FORMAT = dwarf; DEVELOPMENT_TEAM = TZ447KHNZL; ENABLE_STRICT_OBJC_MSGSEND = YES; @@ -3148,7 +3148,7 @@ CLANG_WARN_UNREACHABLE_CODE = YES; CLANG_WARN__DUPLICATE_METHOD_MATCH = YES; COPY_PHASE_STRIP = NO; - CURRENT_PROJECT_VERSION = 200000254; + CURRENT_PROJECT_VERSION = 200000260; DEBUG_INFORMATION_FORMAT = "dwarf-with-dsym"; DEVELOPMENT_TEAM = TZ447KHNZL; ENABLE_NS_ASSERTIONS = NO; diff --git a/apps/ios/Sources/Litter/Views/ConversationTimelineView.swift b/apps/ios/Sources/Litter/Views/ConversationTimelineView.swift index fd51167a0..eb49274cc 100644 --- a/apps/ios/Sources/Litter/Views/ConversationTimelineView.swift +++ b/apps/ios/Sources/Litter/Views/ConversationTimelineView.swift @@ -101,11 +101,11 @@ struct ConversationTurnTimeline: View { } private var commandDisplayMode: ConversationDetailDisplayMode { - ConversationDetailDisplayMode.resolve(commandDisplayModeRaw) + ConversationDetailDisplayMode.resolveRequiredActivity(commandDisplayModeRaw) } private var toolDisplayMode: ConversationDetailDisplayMode { - ConversationDetailDisplayMode.resolve(toolDisplayModeRaw) + ConversationDetailDisplayMode.resolveRequiredActivity(toolDisplayModeRaw) } // Returns AnyView rather than `some View` with @ViewBuilder so the result diff --git a/apps/ios/Sources/Litter/Views/SettingsView.swift b/apps/ios/Sources/Litter/Views/SettingsView.swift index d0f06ddc5..e3256025e 100644 --- a/apps/ios/Sources/Litter/Views/SettingsView.swift +++ b/apps/ios/Sources/Litter/Views/SettingsView.swift @@ -146,21 +146,24 @@ struct SettingsView: View { title: "Internal Thinking", subtitle: "Reasoning and analysis blocks", systemImage: "brain.head.profile", - selection: $reasoningDisplayMode + selection: $reasoningDisplayMode, + modes: ConversationDetailDisplayMode.allCases ) transcriptDisplayPicker( title: "Commands", subtitle: "Shell commands and command output", systemImage: "terminal", - selection: $commandDisplayMode + selection: $commandDisplayMode, + modes: ConversationDetailDisplayMode.requiredActivityCases ) transcriptDisplayPicker( title: "Tools", subtitle: "MCP, web, image, and file-change cards", systemImage: "wrench.and.screwdriver", - selection: $toolDisplayMode + selection: $toolDisplayMode, + modes: ConversationDetailDisplayMode.requiredActivityCases ) } header: { Text("Conversation") @@ -172,10 +175,11 @@ struct SettingsView: View { title: String, subtitle: String, systemImage: String, - selection: Binding + selection: Binding, + modes: [ConversationDetailDisplayMode] ) -> some View { Picker(selection: selection) { - ForEach(ConversationDetailDisplayMode.allCases) { mode in + ForEach(modes) { mode in Text(mode.displayName).tag(mode.rawValue) } } label: { diff --git a/apps/ios/Sources/Litter/Views/ToolCallModels.swift b/apps/ios/Sources/Litter/Views/ToolCallModels.swift index 6753dd36d..4abc16515 100644 --- a/apps/ios/Sources/Litter/Views/ToolCallModels.swift +++ b/apps/ios/Sources/Litter/Views/ToolCallModels.swift @@ -29,6 +29,20 @@ enum ConversationDetailDisplayMode: String, CaseIterable, Identifiable, Equatabl ConversationDetailDisplayMode(rawValue: rawValue) ?? .collapsed } + /// Commands and tool results are part of the durable conversation record, + /// not optional decoration. Older builds allowed persisting `hidden` for + /// those rows; migrate that value to collapsed so reconnecting to a thread + /// can never make its work disappear from the transcript. + static func resolveRequiredActivity(_ rawValue: String) -> ConversationDetailDisplayMode { + let mode = resolve(rawValue) + return mode == .hidden ? .collapsed : mode + } + + static let requiredActivityCases: [ConversationDetailDisplayMode] = [ + .expanded, + .collapsed + ] + func defaultExpanded(isFailed: Bool = false) -> Bool { switch self { case .expanded: diff --git a/apps/ios/Tests/LitterTests/ConversationDisplayPreferenceTests.swift b/apps/ios/Tests/LitterTests/ConversationDisplayPreferenceTests.swift index d1ff864ed..cdaf57cac 100644 --- a/apps/ios/Tests/LitterTests/ConversationDisplayPreferenceTests.swift +++ b/apps/ios/Tests/LitterTests/ConversationDisplayPreferenceTests.swift @@ -8,6 +8,16 @@ final class ConversationDisplayPreferenceTests: XCTestCase { XCTAssertEqual(ConversationDetailDisplayMode.resolve("not-a-mode"), .collapsed) } + func testRequiredActivityMigratesHiddenToCollapsed() { + XCTAssertEqual(ConversationDetailDisplayMode.resolveRequiredActivity("expanded"), .expanded) + XCTAssertEqual(ConversationDetailDisplayMode.resolveRequiredActivity("collapsed"), .collapsed) + XCTAssertEqual(ConversationDetailDisplayMode.resolveRequiredActivity("hidden"), .collapsed) + XCTAssertEqual( + ConversationDetailDisplayMode.requiredActivityCases, + [.expanded, .collapsed] + ) + } + func testCollapsedModeOnlyExpandsFailuresByDefault() { XCTAssertTrue(ConversationDetailDisplayMode.expanded.defaultExpanded()) XCTAssertFalse(ConversationDetailDisplayMode.collapsed.defaultExpanded()) diff --git a/apps/ios/Tests/LitterUITests/LitterUITests.swift b/apps/ios/Tests/LitterUITests/LitterUITests.swift index 7cb797e3a..7ef5ff727 100644 --- a/apps/ios/Tests/LitterUITests/LitterUITests.swift +++ b/apps/ios/Tests/LitterUITests/LitterUITests.swift @@ -60,7 +60,7 @@ final class LitterUITests: XCTestCase { } @MainActor - func testConversationDisplayHiddenModeRemovesDetailRows() throws { + func testConversationDisplayHiddenLegacyValueStillShowsToolActivity() throws { let app = conversationDisplayHarnessApp(reasoning: "hidden", commands: "hidden", tools: "hidden") app.launch() @@ -68,11 +68,11 @@ final class LitterUITests: XCTestCase { XCTAssertTrue(app.staticTexts["UITEST_ASSISTANT_MESSAGE"].exists) XCTAssertFalse(app.staticTexts["Thinking"].exists) XCTAssertFalse(app.staticTexts["Internal reasoning"].exists) - XCTAssertFalse(app.staticTexts["printf UITEST_COMMAND_HEADER"].exists) - XCTAssertFalse(app.staticTexts["uiTest.fixtureTool"].exists) + XCTAssertTrue(app.staticTexts["printf UITEST_COMMAND_HEADER"].exists) + XCTAssertTrue(app.staticTexts["uiTest.fixtureTool"].exists) XCTAssertFalse(app.staticTexts["UITEST_REASONING_DETAIL"].exists) XCTAssertFalse(app.staticTexts["UITEST_COMMAND_OUTPUT"].exists) - XCTAssertFalse(app.staticTexts["UITEST_TOOL_DETAIL"].exists) + XCTAssertTrue(app.staticTexts["UITEST_TOOL_DETAIL"].exists) } @MainActor diff --git a/apps/ios/fastlane/metadata/en-US/release_notes.txt b/apps/ios/fastlane/metadata/en-US/release_notes.txt index 7844b92a5..558a8b30a 100644 --- a/apps/ios/fastlane/metadata/en-US/release_notes.txt +++ b/apps/ios/fastlane/metadata/en-US/release_notes.txt @@ -1,6 +1,7 @@ -Local Studio sessions now stay visible, correctly labeled, and connected to the right runtime. +Local Studio conversations are now faster and fully faithful across live turns and reconnects. -- Start every new Local Studio conversation through its built-in Pi runtime. -- Keep newly synchronized sessions visible even when older sessions are pinned. -- Resume and reconnect without sessions disappearing or changing into Codex sessions. -- Show the correct connected status without asking Local Studio users to sign in to OpenAI. +- Show command, tool-call, and tool-result activity in every conversation. +- Preserve session identity and ordering without replayed messages. +- Queue rapid follow-up messages reliably so a second message is never lost or sent twice. +- Restore exact tool output and completed turns after reconnecting. +- Batch streaming updates at display cadence for smoother, more responsive conversations. diff --git a/apps/ios/project.yml b/apps/ios/project.yml index 4702da467..f7ba76eb0 100644 --- a/apps/ios/project.yml +++ b/apps/ios/project.yml @@ -24,7 +24,7 @@ settings: SWIFT_VERSION: "5.9" SWIFT_STRICT_CONCURRENCY: targeted MARKETING_VERSION: "2.0.0" - CURRENT_PROJECT_VERSION: "200000254" + CURRENT_PROJECT_VERSION: "200000260" SWIFT_OBJC_BRIDGING_HEADER: Sources/Litter/Bridge/codex_bridge_objc.h OTHER_LDFLAGS: "$(inherited) -lc++ -lz -lresolv -lcodex_mobile_client" HEADER_SEARCH_PATHS: "$(inherited) $(PROJECT_DIR)/GeneratedRust/Headers" diff --git a/docs/releases/testflight-whats-new.md b/docs/releases/testflight-whats-new.md index 423fab9ec..00f8713bb 100644 --- a/docs/releases/testflight-whats-new.md +++ b/docs/releases/testflight-whats-new.md @@ -1,17 +1,18 @@ Summary -- Preserved Local Studio session identity across listing, streaming, hydration, and reconnect. -- Routed new Local Studio conversations through its built-in Pi runtime from the first request. -- Kept newly synchronized Local Studio sessions visible alongside legacy pinned sessions. -- Removed false OpenAI sign-in warnings from Local Studio connections. -- Enforced full filesystem access with no approval prompts for Pi and Local Studio Pi sessions. +- Restored Local Studio command, tool-call, and tool-result lifecycle events from legacy controllers. +- Made command and tool activity permanently visible, including for users with an older hidden preference. +- Closed the turn-start race that could lose, duplicate, or misorder a rapid second message. +- Preserved exact queued-turn model, mode, effort, and permission settings. +- Kept session identity, item ordering, and tool output stable across reconnect and hydration. +- Batched streaming deltas at display cadence to reduce native UI churn while preserving exact text. What to test -- Upgrade from 1.6 or 1.7 with existing Local Studio pins and confirm current sessions remain visible and every Local Studio row keeps its label. -- Start a Local Studio conversation from mobile, stream the reply, use shell and file tools, then reopen it in both Litter and Local Studio. -- Background and foreground the app during and after a turn; confirm the session and completed content survive reconnect and resume. -- Confirm Local Studio shows connected without an OpenAI sign-in warning. -- Confirm Pi and Local Studio Pi never request approval and can read and write the selected workspace. -- Confirm ordinary Codex, Pi, and generic KittyLitter connections retain their existing discovery and permission behavior. -- Verify reasoning, tool calls, streaming text, images, compaction, and older-turn hydration on iOS and Android. +- Pair with a Local Studio controller, run a shell command, and confirm the command card and completed output appear live. +- Send a second message immediately after the first tool-producing turn; confirm it appears immediately, runs once, and receives one response. +- Background and foreground during streaming, then reconnect; confirm no user, assistant, command, or tool rows replay or disappear. +- Reopen the same session from Litter and Local Studio and confirm the exact thread identity, order, and output match. +- Compare long streaming replies with 1.7.0 and confirm scrolling and token rendering remain smooth and responsive. +- Confirm Pi and Local Studio Pi continue to use no approvals with full workspace access. +- Verify the same session, tool, and streaming behavior on iOS and Android. diff --git a/shared/rust-bridge/codex-mobile-client/src/ffi/app_store.rs b/shared/rust-bridge/codex-mobile-client/src/ffi/app_store.rs index 1aafc5902..59c07d3a2 100644 --- a/shared/rust-bridge/codex-mobile-client/src/ffi/app_store.rs +++ b/shared/rust-bridge/codex-mobile-client/src/ffi/app_store.rs @@ -26,6 +26,12 @@ pub(crate) struct AppStoreSubscriptionState { } const MAX_COALESCED_STREAMING_TEXT_BYTES: usize = 8 * 1024; +/// Bound native/FFI streaming publishes to the display cadence. Without a +/// short collection window a fast producer and Swift/Kotlin consumer run in +/// lockstep, so `try_recv` sees only one tiny delta and every token copies the +/// platform snapshot. Sixteen milliseconds keeps first-token latency within a +/// single frame while capping that cross-boundary churn near 60 Hz. +const STREAMING_COALESCE_WINDOW: std::time::Duration = std::time::Duration::from_millis(16); #[cfg(test)] mod tests { @@ -41,6 +47,7 @@ mod tests { use codex_app_server_protocol as upstream; use serde_json::json; use std::collections::{HashMap, VecDeque}; + use std::sync::Arc; #[test] fn thread_item_parses_mcp_arguments_json() { @@ -172,6 +179,68 @@ mod tests { )); } + #[tokio::test] + async fn app_store_subscription_batches_interleaved_streaming_at_display_cadence() { + let reducer = Arc::new(AppStoreReducer::new()); + let key = ThreadKey { + server_id: "srv".to_string(), + thread_id: "thread-1".to_string(), + }; + let subscription = AppStoreSubscription { + state: std::sync::Mutex::new(Some(AppStoreSubscriptionState { + rx: reducer.subscribe(), + buffered: VecDeque::new(), + })), + }; + + const DELTA_COUNT: usize = 48; + reducer.emit_thread_streaming_delta( + &key, + "assistant-1", + ThreadStreamingDeltaKind::AssistantText, + "x", + ); + let producer = { + let reducer = Arc::clone(&reducer); + let key = key.clone(); + tokio::spawn(async move { + for _ in 1..DELTA_COUNT { + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + reducer.emit_thread_streaming_delta( + &key, + "assistant-1", + ThreadStreamingDeltaKind::AssistantText, + "x", + ); + } + }) + }; + + let mut delivered_text = String::new(); + let mut update_count = 0usize; + while delivered_text.len() < DELTA_COUNT { + let update = subscription + .next_update() + .await + .expect("streaming update"); + let AppStoreUpdateRecord::ThreadStreamingDelta { text, .. } = update else { + panic!("expected streaming delta"); + }; + delivered_text.push_str(&text); + update_count += 1; + } + producer.await.expect("producer"); + + eprintln!( + "batched {DELTA_COUNT} streaming deltas into {update_count} native updates" + ); + assert_eq!(delivered_text, "x".repeat(DELTA_COUNT)); + assert!( + update_count <= 12, + "expected display-cadence batching, got {update_count} native updates" + ); + } + #[test] fn app_store_subscription_coalesces_refresh_only_updates_into_full_resync() { let reducer = AppStoreReducer::new(); @@ -726,9 +795,37 @@ async fn receive_next_update( state.rx.recv().await? }; + let first = coalesce_streaming_window(state, first).await?; coalesce_ready_updates(state, first) } +async fn coalesce_streaming_window( + state: &mut AppStoreSubscriptionState, + mut update: AppStoreUpdateRecord, +) -> Result { + if !matches!(update, AppStoreUpdateRecord::ThreadStreamingDelta { .. }) { + return Ok(update); + } + + let deadline = tokio::time::Instant::now() + STREAMING_COALESCE_WINDOW; + loop { + let next = if let Some(update) = state.buffered.pop_front() { + update + } else { + match tokio::time::timeout_at(deadline, state.rx.recv()).await { + Ok(Ok(update)) => update, + Ok(Err(error)) => return Err(error), + Err(_) => return Ok(update), + } + }; + + if let Err(next) = merge_app_update(&mut update, next) { + state.buffered.push_front(next); + return Ok(update); + } + } +} + fn coalesce_ready_updates( state: &mut AppStoreSubscriptionState, mut update: AppStoreUpdateRecord, diff --git a/shared/rust-bridge/codex-mobile-client/src/mobile_client/mod.rs b/shared/rust-bridge/codex-mobile-client/src/mobile_client/mod.rs index 3c0f3cdc9..7acaf210e 100644 --- a/shared/rust-bridge/codex-mobile-client/src/mobile_client/mod.rs +++ b/shared/rust-bridge/codex-mobile-client/src/mobile_client/mod.rs @@ -2744,7 +2744,14 @@ impl MobileClient { || matches!(thread.info.status, ThreadSummaryStatus::Active) }); self.external_resume_thread_inner(server_id, thread_id, host_id, reconcile_active_items) - .await + .await?; + maybe_send_next_local_queued_follow_up( + Arc::clone(&self.app_store), + Arc::clone(&self.sessions), + key, + ) + .await; + Ok(()) } /// Force a fresh `thread/resume` against the server even if a direct @@ -2768,7 +2775,17 @@ impl MobileClient { thread_id: &str, ) -> Result<(), RpcError> { self.external_resume_thread_inner(server_id, thread_id, None, true) - .await + .await?; + maybe_send_next_local_queued_follow_up( + Arc::clone(&self.app_store), + Arc::clone(&self.sessions), + ThreadKey { + server_id: server_id.to_string(), + thread_id: thread_id.to_string(), + }, + ) + .await; + Ok(()) } async fn external_resume_thread_inner( @@ -3347,23 +3364,20 @@ impl MobileClient { params.approval_policy = None; params.sandbox_policy = None; } - let has_active_turn = thread_snapshot - .as_ref() - .is_some_and(|thread| thread.active_turn_id.is_some()); + let has_active_turn = thread_snapshot.as_ref().is_some_and(|thread| { + thread.active_turn_id.is_some() + || matches!(thread.info.status, ThreadSummaryStatus::Active) + }); let direct_params = params.clone(); - // Stage an optimistic local overlay so the user sees their message - // immediately, before the server echoes it back. - let optimistic_overlay_id = if !has_active_turn { - self.app_store - .stage_local_user_message_overlay(&thread_key, ¶ms.input) - } else { - None - }; let queued_draft = has_active_turn .then(|| { queued_follow_up_draft_from_inputs(¶ms.input, AppQueuedFollowUpKind::Message) }) - .flatten(); + .flatten() + .map(|mut draft| { + draft.turn_start_params = Some(direct_params.clone()); + draft + }); if let Some(draft) = queued_draft.clone() { self.app_store .enqueue_thread_follow_up_draft(&thread_key, draft.clone()); @@ -3410,13 +3424,31 @@ impl MobileClient { } } + let Some(reservation_turn_id) = self.app_store.reserve_local_turn_start(&thread_key) else { + // A TurnStarted/status event won the race after the snapshot was + // read. Preserve this input as the next message instead of firing + // a conflicting second turn/start. + if let Some(mut draft) = queued_follow_up_draft_from_inputs( + ¶ms.input, + AppQueuedFollowUpKind::Message, + ) { + draft.turn_start_params = Some(direct_params.clone()); + self.app_store + .enqueue_thread_follow_up_draft(&thread_key, draft); + return Ok(()); + } + return Err(RpcError::Deserialization( + "thread became active before the turn could be reserved".to_string(), + )); + }; + // Stage an optimistic local overlay so the user sees their message + // immediately, before the server echoes it back. + let optimistic_overlay_id = self + .app_store + .stage_local_user_message_overlay(&thread_key, ¶ms.input); let direct_command_id = self.app_store.begin_server_mutating_command( server_id, - if queued_draft.is_some() { - ServerMutatingCommandKind::SetQueuedFollowUpsState - } else { - ServerMutatingCommandKind::StartTurn - }, + ServerMutatingCommandKind::StartTurn, ¶ms.thread_id, ); let response_result = self @@ -3431,17 +3463,24 @@ impl MobileClient { let response = match response_result { Ok(response) => response, Err(error) => { - self.app_store - .finish_server_mutating_command_failure(server_id, &direct_command_id); - if let Some(overlay_id) = optimistic_overlay_id.as_ref() { - self.app_store - .remove_local_overlay_item(&thread_key, overlay_id); - } - if let Some(draft) = queued_draft.as_ref() { + if self + .app_store + .release_local_turn_start(&thread_key, &reservation_turn_id) + { self.app_store - .remove_thread_follow_up_draft(&thread_key, &draft.preview.id); + .finish_server_mutating_command_failure(server_id, &direct_command_id); + if let Some(overlay_id) = optimistic_overlay_id.as_ref() { + self.app_store + .remove_local_overlay_item(&thread_key, overlay_id); + } + return Err(RpcError::Deserialization(error)); } - return Err(RpcError::Deserialization(error)); + // A live TurnStarted/TurnCompleted event already replaced the + // reservation, so the server accepted the turn even though + // its response raced a transport failure. + self.app_store + .finish_server_mutating_command_success(server_id, &direct_command_id); + return Ok(()); } }; self.app_store @@ -3453,6 +3492,11 @@ impl MobileClient { &response.turn.id, ); } + self.app_store.resolve_local_turn_start( + &thread_key, + &reservation_turn_id, + &response.turn.id, + ); Ok(()) } diff --git a/shared/rust-bridge/codex-mobile-client/src/mobile_client/store_listener.rs b/shared/rust-bridge/codex-mobile-client/src/mobile_client/store_listener.rs index 7e48e9240..c5421bfdd 100644 --- a/shared/rust-bridge/codex-mobile-client/src/mobile_client/store_listener.rs +++ b/shared/rust-bridge/codex-mobile-client/src/mobile_client/store_listener.rs @@ -17,13 +17,12 @@ pub(super) fn spawn_store_listener( Arc::clone(&sessions), &event, ); - if let UiEvent::TurnCompleted { key, .. } = &event { - maybe_send_next_local_queued_follow_up( - Arc::clone(&app_store), - Arc::clone(&sessions), - key.clone(), - ) - .await; + if let Some(key) = queued_follow_up_dispatch_key(&event) { + let app_store = Arc::clone(&app_store); + let sessions = Arc::clone(&sessions); + MobileClient::spawn_detached(async move { + maybe_send_next_local_queued_follow_up(app_store, sessions, key).await; + }); } } Err(broadcast::error::RecvError::Closed) => break, @@ -35,6 +34,18 @@ pub(super) fn spawn_store_listener( }); } +fn queued_follow_up_dispatch_key(event: &UiEvent) -> Option { + match event { + UiEvent::TurnCompleted { key, .. } => Some(key.clone()), + UiEvent::ThreadStatusChanged { key, notification } + if matches!(¬ification.status, upstream::ThreadStatus::Idle) => + { + Some(key.clone()) + } + _ => None, + } +} + fn maybe_hydrate_collab_agent_metadata( app_store: Arc, sessions: Arc>>>, @@ -196,7 +207,10 @@ pub(super) async fn maybe_send_next_local_queued_follow_up( let Some(thread) = snapshot.threads.get(&key).cloned() else { return; }; - if thread.active_turn_id.is_some() || thread.queued_follow_up_drafts.is_empty() { + if thread.active_turn_id.is_some() + || matches!(thread.info.status, ThreadSummaryStatus::Active) + || thread.queued_follow_up_drafts.is_empty() + { return; } @@ -211,28 +225,146 @@ pub(super) async fn maybe_send_next_local_queued_follow_up( return; }; - let Some(draft) = app_store.claim_first_queued_follow_up_draft(&key) else { + let runtime_kind = thread.agent_runtime_kind.clone(); + let Some((draft, reservation_turn_id)) = + app_store.claim_queued_follow_up_for_dispatch(&key) + else { return; }; - let response = session.request( - "turn/start", - serde_json::json!({ - "threadId": key.thread_id, - "input": draft.inputs.clone(), - }), + let optimistic_overlay_id = app_store.stage_local_user_message_overlay(&key, &draft.inputs); + let mut turn_start_params = draft + .turn_start_params + .clone() + .unwrap_or_else(|| upstream::TurnStartParams { + thread_id: key.thread_id.clone(), + input: draft.inputs.clone(), + ..Default::default() + }); + // The controller owns permissions for Pi and Local Studio, but their + // mobile contract is always non-interactive full access. This path sends + // directly to the selected runtime, so apply the same normalization as + // MobileClient::request_typed_for_server_runtime. + if matches!(runtime_kind.as_str(), "pi" | "local-studio") { + turn_start_params.approval_policy = Some(upstream::AskForApproval::Never); + turn_start_params.sandbox_policy = Some(upstream::SandboxPolicy::DangerFullAccess); + } + turn_start_params.thread_id = key.thread_id.clone(); + turn_start_params.input = draft.inputs.clone(); + let command_id = app_store.begin_server_mutating_command( + &key.server_id, + ServerMutatingCommandKind::StartTurn, + &key.thread_id, ); - if let Err(error) = response.await { - app_store.restore_queued_follow_up_draft_front(&key, draft); - warn!( - "MobileClient: failed to autosend queued follow-up for {} thread {}: {}", - key.server_id, key.thread_id, error - ); + let response = session + .request_client_for_runtime( + runtime_kind, + upstream::ClientRequest::TurnStart { + request_id: upstream::RequestId::Integer(crate::next_request_id()), + params: turn_start_params, + }, + ) + .await + .and_then(|value| { + serde_json::from_value::(value) + .map_err(|error| RpcError::Deserialization(error.to_string())) + }); + match response { + Ok(response) => { + app_store.finish_server_mutating_command_success(&key.server_id, &command_id); + if let Some(overlay_id) = optimistic_overlay_id.as_deref() { + app_store.bind_local_user_message_overlay_to_turn( + &key, + overlay_id, + &response.turn.id, + ); + } + app_store.resolve_local_turn_start( + &key, + &reservation_turn_id, + &response.turn.id, + ); + } + Err(error) => { + app_store.finish_server_mutating_command_failure(&key.server_id, &command_id); + if app_store.release_local_turn_start(&key, &reservation_turn_id) { + if let Some(overlay_id) = optimistic_overlay_id.as_deref() { + app_store.remove_local_overlay_item(&key, overlay_id); + } + app_store.restore_queued_follow_up_draft_front(&key, draft); + } + warn!( + "MobileClient: failed to autosend queued follow-up for {} thread {}: {}", + key.server_id, key.thread_id, error + ); + } } } #[cfg(test)] mod tests { use super::*; + use crate::conversation_uniffi::HydratedConversationItemContent; + use crate::session::connection::TestRequestHandler; + use crate::store::AppQueuedFollowUpKind; + use std::sync::{Arc, Mutex as StdMutex}; + + fn make_thread_info(id: &str) -> ThreadInfo { + ThreadInfo { + id: id.to_string(), + title: Some("Thread".to_string()), + model: Some("glm-5.2".to_string()), + status: ThreadSummaryStatus::Idle, + preview: None, + cwd: Some("/tmp".to_string()), + path: Some("/tmp/thread".to_string()), + model_provider: Some("local-studio".to_string()), + agent_nickname: None, + agent_role: None, + parent_thread_id: None, + forked_from_id: None, + agent_status: None, + created_at: Some(1), + updated_at: Some(2), + } + } + + fn make_server_config(server_id: &str) -> ServerConfig { + ServerConfig { + server_id: server_id.to_string(), + display_name: "Local Studio".to_string(), + host: "127.0.0.1".to_string(), + port: 0, + websocket_url: Some("ws://127.0.0.1:0".to_string()), + is_local: false, + tls: false, + } + } + + fn queued_draft(text: &str) -> crate::store::QueuedFollowUpDraft { + queued_follow_up_draft_from_inputs( + &[upstream::UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }], + AppQueuedFollowUpKind::Message, + ) + .expect("queued draft") + } + + fn turn_start_response(turn_id: &str) -> serde_json::Value { + serde_json::json!({ + "turn": { + "id": turn_id, + "items": [], + "itemsView": "full", + "status": "inProgress", + "error": null, + "startedAt": 1, + "completedAt": null, + "durationMs": null + } + }) + } #[test] fn collab_receiver_thread_ids_extracts_spawn_agent_targets() { @@ -294,4 +426,148 @@ mod tests { )) ); } + + #[test] + fn idle_status_update_releases_a_queued_follow_up() { + let key = ThreadKey { + server_id: "srv".to_string(), + thread_id: "thread-1".to_string(), + }; + let event = UiEvent::ThreadStatusChanged { + key: key.clone(), + notification: upstream::ThreadStatusChangedNotification { + thread_id: key.thread_id.clone(), + status: upstream::ThreadStatus::Idle, + }, + }; + + assert_eq!(queued_follow_up_dispatch_key(&event), Some(key)); + } + + #[tokio::test] + async fn queued_follow_up_dispatch_is_exactly_once_and_preserves_next_message() { + let app_store = Arc::new(AppStoreReducer::new()); + let sessions = Arc::new(RwLock::new(HashMap::new())); + let server_id = "srv"; + let key = ThreadKey { + server_id: server_id.to_string(), + thread_id: "thread-1".to_string(), + }; + let config = make_server_config(server_id); + app_store.upsert_server(&config, ServerHealthSnapshot::Connected); + let mut thread = ThreadSnapshot::from_info(server_id, make_thread_info(&key.thread_id)); + thread.agent_runtime_kind = "local-studio".to_string(); + app_store.upsert_thread_snapshot(thread); + let mut second = queued_draft("second message"); + second.turn_start_params = Some(upstream::TurnStartParams { + thread_id: key.thread_id.clone(), + input: second.inputs.clone(), + model: Some("glm-5.2".to_string()), + ..Default::default() + }); + app_store.enqueue_thread_follow_up_draft(&key, second); + app_store.enqueue_thread_follow_up_draft(&key, queued_draft("third message")); + + let requests = Arc::new(StdMutex::new(Vec::new())); + let handler: TestRequestHandler = { + let requests = Arc::clone(&requests); + Arc::new(move |request| { + requests.lock().expect("request log").push(request.clone()); + match request { + upstream::ClientRequest::TurnStart { params, .. } => { + assert_eq!(params.model.as_deref(), Some("glm-5.2")); + assert_eq!(params.approval_policy, Some(upstream::AskForApproval::Never)); + assert_eq!( + params.sandbox_policy, + Some(upstream::SandboxPolicy::DangerFullAccess) + ); + Ok(turn_start_response("turn-queued")) + } + other => Err(RpcError::Deserialization(format!( + "unexpected request: {}", + other.method() + ))), + } + }) + }; + sessions.write().expect("sessions lock").insert( + server_id.to_string(), + Arc::new(ServerSession::test_stub_with_runtime_handlers( + config, + vec![("local-studio".to_string(), handler)], + )), + ); + + tokio::join!( + maybe_send_next_local_queued_follow_up( + Arc::clone(&app_store), + Arc::clone(&sessions), + key.clone() + ), + maybe_send_next_local_queued_follow_up( + Arc::clone(&app_store), + Arc::clone(&sessions), + key.clone() + ) + ); + + assert_eq!(requests.lock().expect("request log").len(), 1); + let snapshot = app_store.snapshot(); + let thread = snapshot.threads.get(&key).expect("thread"); + assert_eq!(thread.active_turn_id.as_deref(), Some("turn-queued")); + assert_eq!(thread.queued_follow_up_drafts.len(), 1); + assert_eq!(thread.queued_follow_up_drafts[0].preview.text, "third message"); + assert!(thread.local_overlay_items.iter().any(|item| { + item.source_turn_id.as_deref() == Some("turn-queued") + && matches!( + &item.content, + HydratedConversationItemContent::User(data) + if data.text == "second message" + ) + })); + } + + #[tokio::test] + async fn failed_queued_follow_up_dispatch_restores_message_without_a_ghost_turn() { + let app_store = Arc::new(AppStoreReducer::new()); + let sessions = Arc::new(RwLock::new(HashMap::new())); + let server_id = "srv"; + let key = ThreadKey { + server_id: server_id.to_string(), + thread_id: "thread-1".to_string(), + }; + let config = make_server_config(server_id); + app_store.upsert_server(&config, ServerHealthSnapshot::Connected); + let mut thread = ThreadSnapshot::from_info(server_id, make_thread_info(&key.thread_id)); + thread.agent_runtime_kind = "local-studio".to_string(); + app_store.upsert_thread_snapshot(thread); + app_store.enqueue_thread_follow_up_draft(&key, queued_draft("keep me")); + + let handler: TestRequestHandler = Arc::new(|request| match request { + upstream::ClientRequest::TurnStart { .. } => { + Err(RpcError::Transport(TransportError::Disconnected)) + } + other => Err(RpcError::Deserialization(format!( + "unexpected request: {}", + other.method() + ))), + }); + sessions.write().expect("sessions lock").insert( + server_id.to_string(), + Arc::new(ServerSession::test_stub_with_runtime_handlers( + config, + vec![("local-studio".to_string(), handler)], + )), + ); + + maybe_send_next_local_queued_follow_up(app_store.clone(), sessions, key.clone()).await; + + let snapshot = app_store.snapshot(); + let thread = snapshot.threads.get(&key).expect("thread"); + assert_eq!(thread.active_turn_id, None); + assert_eq!(thread.info.status, ThreadSummaryStatus::Idle); + assert_eq!(thread.queued_follow_up_drafts.len(), 1); + assert_eq!(thread.queued_follow_up_drafts[0].preview.text, "keep me"); + assert!(thread.local_overlay_items.is_empty()); + } } diff --git a/shared/rust-bridge/codex-mobile-client/src/mobile_client/tests.rs b/shared/rust-bridge/codex-mobile-client/src/mobile_client/tests.rs index 8ba367c41..b5bc93d1e 100644 --- a/shared/rust-bridge/codex-mobile-client/src/mobile_client/tests.rs +++ b/shared/rust-bridge/codex-mobile-client/src/mobile_client/tests.rs @@ -2031,6 +2031,151 @@ mod mobile_client_tests { ); } + #[tokio::test] + async fn active_status_without_turn_id_queues_instead_of_losing_the_message() { + let client = MobileClient::new(); + let server_id = "srv"; + let thread_id = "thread-1"; + let key = ThreadKey { + server_id: server_id.to_string(), + thread_id: thread_id.to_string(), + }; + let config = make_server_config(server_id); + client + .app_store + .upsert_server(&config, ServerHealthSnapshot::Connected); + let mut thread = ThreadSnapshot::from_info(server_id, make_thread_info(thread_id)); + thread.active_turn_id = None; + thread.info.status = ThreadSummaryStatus::Active; + thread.agent_runtime_kind = "local-studio".to_string(); + client.app_store.upsert_thread_snapshot(thread); + + let request_count = Arc::new(StdMutex::new(0usize)); + let handler: TestRequestHandler = { + let request_count = Arc::clone(&request_count); + Arc::new(move |request| { + *request_count.lock().expect("request count") += 1; + Err(RpcError::Deserialization(format!( + "unexpected request while active: {}", + request.method() + ))) + }) + }; + client.sessions.write().expect("sessions lock").insert( + server_id.to_string(), + Arc::new(ServerSession::test_stub_with_runtime_handlers( + config, + vec![("local-studio".to_string(), handler)], + )), + ); + + client + .start_turn( + server_id, + upstream::TurnStartParams { + thread_id: thread_id.to_string(), + input: vec![upstream::UserInput::Text { + text: "second message".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + ) + .await + .expect("active-status message should queue"); + + assert_eq!(*request_count.lock().expect("request count"), 0); + let snapshot = client.app_store.snapshot(); + let thread = snapshot.threads.get(&key).expect("thread"); + assert_eq!(thread.queued_follow_up_drafts.len(), 1); + assert_eq!(thread.queued_follow_up_drafts[0].preview.text, "second message"); + } + + #[tokio::test] + async fn turn_start_response_reserves_thread_before_a_rapid_second_send() { + let client = MobileClient::new(); + let server_id = "srv"; + let thread_id = "thread-1"; + let key = ThreadKey { + server_id: server_id.to_string(), + thread_id: thread_id.to_string(), + }; + let config = make_server_config(server_id); + client + .app_store + .upsert_server(&config, ServerHealthSnapshot::Connected); + let mut thread = ThreadSnapshot::from_info(server_id, make_thread_info(thread_id)); + thread.info.status = ThreadSummaryStatus::Idle; + thread.agent_runtime_kind = "local-studio".to_string(); + client.app_store.upsert_thread_snapshot(thread); + + let request_count = Arc::new(StdMutex::new(0usize)); + let handler: TestRequestHandler = { + let request_count = Arc::clone(&request_count); + Arc::new(move |request| match request { + upstream::ClientRequest::TurnStart { .. } => { + *request_count.lock().expect("request count") += 1; + Ok(json!({ + "turn": { + "id": "turn-first", + "items": [], + "itemsView": "full", + "status": "inProgress", + "error": null, + "startedAt": 1, + "completedAt": null, + "durationMs": null + } + })) + } + other => Err(RpcError::Deserialization(format!( + "unexpected request: {}", + other.method() + ))), + }) + }; + client.sessions.write().expect("sessions lock").insert( + server_id.to_string(), + Arc::new(ServerSession::test_stub_with_runtime_handlers( + config, + vec![("local-studio".to_string(), handler)], + )), + ); + + for text in ["first message", "second message"] { + client + .start_turn( + server_id, + upstream::TurnStartParams { + thread_id: thread_id.to_string(), + input: vec![upstream::UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }], + model: Some("glm-5.2".to_string()), + ..Default::default() + }, + ) + .await + .expect("turn send"); + } + + assert_eq!(*request_count.lock().expect("request count"), 1); + let snapshot = client.app_store.snapshot(); + let thread = snapshot.threads.get(&key).expect("thread"); + assert_eq!(thread.active_turn_id.as_deref(), Some("turn-first")); + assert_eq!(thread.info.status, ThreadSummaryStatus::Active); + assert_eq!(thread.queued_follow_up_drafts.len(), 1); + assert_eq!(thread.queued_follow_up_drafts[0].preview.text, "second message"); + assert_eq!( + thread.queued_follow_up_drafts[0] + .turn_start_params + .as_ref() + .and_then(|params| params.model.as_deref()), + Some("glm-5.2") + ); + } + #[test] fn queued_follow_up_message_json_round_trips_skill_inputs() { let inputs = vec![ @@ -2065,4 +2210,238 @@ mod mobile_client_tests { assert_eq!(preview.kind, AppQueuedFollowUpKind::PendingSteer); assert_eq!(preview.text, "Please try the same search again."); } + + #[tokio::test] + #[ignore = "requires LITTER_LIVE_LOCAL_STUDIO_PAIR and LITTER_LIVE_LOCAL_STUDIO_THREAD_ID"] + async fn live_local_studio_resume_preserves_unique_command_results() { + let pair_json = std::env::var("LITTER_LIVE_LOCAL_STUDIO_PAIR") + .expect("LITTER_LIVE_LOCAL_STUDIO_PAIR"); + let thread_id = std::env::var("LITTER_LIVE_LOCAL_STUDIO_THREAD_ID") + .expect("LITTER_LIVE_LOCAL_STUDIO_THREAD_ID"); + let pair = crate::alleycat::parse_pair_payload(&pair_json).expect("valid pair payload"); + let server_id = format!("alleycat:local-studio:{}", pair.node_id); + let client = MobileClient::new(); + + let acceptance = async { + client + .connect_remote_over_alleycat( + server_id.clone(), + "Local Studio live acceptance".to_string(), + pair, + "local-studio".to_string(), + vec!["local-studio".to_string()], + AlleycatAgentWire::Jsonl, + ) + .await + .map_err(|error| format!("connect to Local Studio: {error}"))?; + client + .external_resume_thread(&server_id, &thread_id, None) + .await + .map_err(|error| format!("resume Local Studio thread: {error}"))?; + client + .load_thread_turns_page(&server_id, &thread_id, None, Some(5)) + .await + .map_err(|error| format!("load authoritative turn page: {error}"))?; + + let key = ThreadKey { + server_id: server_id.clone(), + thread_id: thread_id.clone(), + }; + let snapshot = client.app_store.snapshot(); + let thread = snapshot + .threads + .get(&key) + .ok_or_else(|| "resumed thread snapshot missing".to_string())?; + let unique_count = thread + .items + .iter() + .map(|item| item.id.as_str()) + .collect::>() + .len(); + let has_command_output = thread.items.iter().any(|item| { + matches!( + &item.content, + crate::conversation_uniffi::HydratedConversationItemContent::CommandExecution(data) + if data.output.as_deref().is_some_and(|output| !output.trim().is_empty()) + ) + }); + Ok::<_, String>((unique_count, thread.items.len(), has_command_output)) + }; + let acceptance = acceptance.await; + client.disconnect_server(&server_id); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + client.shutdown_alleycat_endpoint().await; + let (unique_count, item_count, has_command_output) = acceptance.expect("live acceptance"); + assert_eq!( + unique_count, + item_count, + "resumed timeline must not replay duplicate item ids" + ); + assert!( + has_command_output, + "resumed timeline must retain completed command output" + ); + } + + #[tokio::test] + #[ignore = "requires LITTER_LIVE_LOCAL_STUDIO_PAIR"] + async fn live_local_studio_rapid_second_send_runs_once_after_tool_turn() { + let pair_json = std::env::var("LITTER_LIVE_LOCAL_STUDIO_PAIR") + .expect("LITTER_LIVE_LOCAL_STUDIO_PAIR"); + let pair = crate::alleycat::parse_pair_payload(&pair_json).expect("valid pair payload"); + let server_id = format!("alleycat:local-studio:{}", pair.node_id); + let client = MobileClient::new(); + + let acceptance = async { + client + .connect_remote_over_alleycat( + server_id.clone(), + "Local Studio live queued-turn acceptance".to_string(), + pair, + "local-studio".to_string(), + vec!["local-studio".to_string()], + AlleycatAgentWire::Jsonl, + ) + .await + .map_err(|error| format!("connect to Local Studio: {error}"))?; + let start_params: upstream::ThreadStartParams = crate::types::AppStartThreadRequest { + agent_runtime_kind: Some("local-studio".to_string()), + model: Some("glm-5.2".to_string()), + cwd: Some("/tmp".to_string()), + approval_policy: Some(crate::types::AppAskForApproval::Never), + sandbox: Some(crate::types::AppSandboxMode::DangerFullAccess), + developer_instructions: None, + persist_extended_history: false, + dynamic_tools: None, + ephemeral: Some(false), + } + .try_into() + .map_err(|error: crate::RpcClientError| error.to_string())?; + let thread_response: upstream::ThreadStartResponse = client + .request_typed_for_server_runtime( + &server_id, + "local-studio".to_string(), + upstream::ClientRequest::ThreadStart { + request_id: upstream::RequestId::Integer(crate::next_request_id()), + params: start_params, + }, + ) + .await?; + let key = client + .apply_thread_start_response(&server_id, &thread_response) + .map_err(|error| error.to_string())?; + client.note_thread_runtime(key.clone(), "local-studio".to_string()); + + for text in [ + "Use the bash tool exactly once to run sleep 1; printf LITTER_FIRST_TOOL_OK, then reply exactly LITTER_FIRST_FINAL_OK.", + "Reply exactly LITTER_SECOND_FINAL_OK.", + ] { + client + .start_turn( + &server_id, + upstream::TurnStartParams { + thread_id: key.thread_id.clone(), + input: vec![upstream::UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }], + model: Some("glm-5.2".to_string()), + ..Default::default() + }, + ) + .await + .map_err(|error| error.to_string())?; + } + + let (unique_count, item_count, live_has_tool_output) = tokio::time::timeout( + std::time::Duration::from_secs(90), + async { + loop { + let snapshot = client.app_store.snapshot(); + if let Some(thread) = snapshot.threads.get(&key) { + let assistant_text = thread + .items + .iter() + .filter_map(|item| match &item.content { + crate::conversation_uniffi::HydratedConversationItemContent::Assistant(data) => { + Some(data.text.as_str()) + } + _ => None, + }) + .collect::>() + .join("\n"); + let has_tool_output = thread.items.iter().any(|item| { + matches!( + &item.content, + crate::conversation_uniffi::HydratedConversationItemContent::CommandExecution(data) + if data.output.as_deref().is_some_and(|output| output.contains("LITTER_FIRST_TOOL_OK")) + ) + }); + if thread.active_turn_id.is_none() + && thread.queued_follow_up_drafts.is_empty() + && assistant_text.contains("LITTER_FIRST_FINAL_OK") + && assistant_text.contains("LITTER_SECOND_FINAL_OK") + { + let unique_count = thread + .items + .iter() + .map(|item| item.id.as_str()) + .collect::>() + .len(); + return (unique_count, thread.items.len(), has_tool_output); + } + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + }, + ) + .await + .map_err(|_| "timed out waiting for both queued turns".to_string())?; + if !live_has_tool_output { + client + .load_thread_turns_page(&server_id, &key.thread_id, None, Some(5)) + .await + .map_err(|error| format!("authoritative tool refresh: {error}"))?; + } + let authoritative_has_tool_output = client + .app_store + .snapshot() + .threads + .get(&key) + .is_some_and(|thread| { + thread.items.iter().any(|item| { + matches!( + &item.content, + crate::conversation_uniffi::HydratedConversationItemContent::CommandExecution(data) + if data.output.as_deref().is_some_and(|output| output.contains("LITTER_FIRST_TOOL_OK")) + ) + }) + }); + Ok::<_, String>(( + unique_count, + item_count, + live_has_tool_output, + authoritative_has_tool_output, + )) + } + .await; + + client.disconnect_server(&server_id); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + client.shutdown_alleycat_endpoint().await; + let (unique_count, item_count, live_has_tool_output, authoritative_has_tool_output) = + acceptance.expect("live queued-turn acceptance"); + assert_eq!( + unique_count, item_count, + "live two-turn timeline must not replay duplicate item ids" + ); + assert!( + live_has_tool_output, + "live reducer must retain the streamed command result" + ); + assert!( + authoritative_has_tool_output, + "authoritative refresh must retain the command result" + ); + } } diff --git a/shared/rust-bridge/codex-mobile-client/src/mobile_client/thread_projection.rs b/shared/rust-bridge/codex-mobile-client/src/mobile_client/thread_projection.rs index 80c2eccb0..a9afb4171 100644 --- a/shared/rust-bridge/codex-mobile-client/src/mobile_client/thread_projection.rs +++ b/shared/rust-bridge/codex-mobile-client/src/mobile_client/thread_projection.rs @@ -86,6 +86,7 @@ pub(super) fn queued_follow_up_draft_from_inputs( }, inputs: inputs.to_vec(), source_message_json: queued_follow_up_message_json_from_inputs(inputs), + turn_start_params: None, }) } diff --git a/shared/rust-bridge/codex-mobile-client/src/store/reducer.rs b/shared/rust-bridge/codex-mobile-client/src/store/reducer.rs index 83728f125..27929d755 100644 --- a/shared/rust-bridge/codex-mobile-client/src/store/reducer.rs +++ b/shared/rust-bridge/codex-mobile-client/src/store/reducer.rs @@ -51,6 +51,7 @@ use crate::terminal::TerminalBackendKind; const USER_INPUT_NOTE_PREFIX: &str = "user_note: "; const USER_INPUT_OTHER_OPTION_LABEL: &str = "None of the above"; const LOCAL_USER_MESSAGE_ITEM_PREFIX: &str = "local-user-message:"; +const LOCAL_QUEUED_TURN_RESERVATION_PREFIX: &str = "local-queued-turn:"; const DESKTOP_FILE_CONTEXT_HEADER: &str = "# Files mentioned by the user:"; const DESKTOP_FILE_CONTEXT_REQUEST_HEADER: &str = "## My request for Codex:"; @@ -699,6 +700,7 @@ impl AppStoreReducer { preview, inputs: Vec::new(), source_message_json: None, + turn_start_params: None, }, ); } @@ -719,6 +721,7 @@ impl AppStoreReducer { } } + #[cfg(test)] pub(crate) fn claim_first_queued_follow_up_draft( &self, key: &ThreadKey, @@ -739,6 +742,121 @@ impl AppStoreReducer { draft } + /// Atomically claim one queued message and reserve the thread's next turn. + /// + /// The reservation closes the response-before-notification window in + /// which two terminal notifications (or a reconnect refresh racing a + /// terminal notification) could both observe an idle thread and start two + /// queued messages. It also makes the composer reflect that work is in + /// flight immediately, before `turn/started` crosses the network. + pub(crate) fn claim_queued_follow_up_for_dispatch( + &self, + key: &ThreadKey, + ) -> Option<(QueuedFollowUpDraft, String)> { + let reservation_turn_id = format!( + "{LOCAL_QUEUED_TURN_RESERVATION_PREFIX}{}", + uuid::Uuid::new_v4() + ); + let result = self + .mutate_thread_with_result(key, |thread| { + if thread.active_turn_id.is_some() + || matches!(thread.info.status, ThreadSummaryStatus::Active) + { + return None; + } + let position = thread.queued_follow_up_drafts.iter().position(|draft| { + draft.preview.kind == super::snapshot::AppQueuedFollowUpKind::Message + })?; + let draft = thread.queued_follow_up_drafts.remove(position); + thread.active_turn_id = Some(reservation_turn_id.clone()); + thread.info.status = ThreadSummaryStatus::Active; + sync_thread_follow_up_projection(thread); + Some((draft, reservation_turn_id.clone())) + }) + .flatten(); + if result.is_some() { + self.emit_thread_metadata_changed(key); + } + result + } + + /// Reserve an idle thread before the turn/start request crosses the + /// network. This closes the response-before-TurnStarted window where a + /// rapid second send could otherwise observe the thread as idle and race + /// a second turn/start against the first one. + pub(crate) fn reserve_local_turn_start(&self, key: &ThreadKey) -> Option { + let reservation_turn_id = format!( + "{LOCAL_QUEUED_TURN_RESERVATION_PREFIX}{}", + uuid::Uuid::new_v4() + ); + let reserved = self + .mutate_thread_with_result(key, |thread| { + if thread.active_turn_id.is_some() + || matches!(thread.info.status, ThreadSummaryStatus::Active) + { + return None; + } + thread.active_turn_id = Some(reservation_turn_id.clone()); + thread.info.status = ThreadSummaryStatus::Active; + Some(reservation_turn_id.clone()) + }) + .flatten(); + if reserved.is_some() { + self.emit_thread_metadata_changed(key); + } + reserved + } + + /// Replace a still-pending local reservation with the authoritative turn + /// id returned by `turn/start`. A live `turn/started` or `turn/completed` + /// event may already have won the race; in that case this is intentionally + /// a no-op so an old response can never resurrect a completed turn. + pub(crate) fn resolve_local_turn_start( + &self, + key: &ThreadKey, + reservation_turn_id: &str, + turn_id: &str, + ) { + let changed = self + .mutate_thread_with_result(key, |thread| { + if thread.active_turn_id.as_deref() != Some(reservation_turn_id) { + return false; + } + thread.active_turn_id = Some(turn_id.to_string()); + thread.info.status = ThreadSummaryStatus::Active; + true + }) + .unwrap_or(false); + if changed { + self.emit_thread_metadata_changed(key); + } + } + + /// Release a failed local reservation. Returns true only when the + /// reservation was still authoritative, which tells the caller whether it + /// is safe to restore the draft. If a live event already advanced the + /// thread, restoring would send the user's message twice. + pub(crate) fn release_local_turn_start( + &self, + key: &ThreadKey, + reservation_turn_id: &str, + ) -> bool { + let released = self + .mutate_thread_with_result(key, |thread| { + if thread.active_turn_id.as_deref() != Some(reservation_turn_id) { + return false; + } + thread.active_turn_id = None; + thread.info.status = ThreadSummaryStatus::Idle; + true + }) + .unwrap_or(false); + if released { + self.emit_thread_metadata_changed(key); + } + released + } + pub(crate) fn restore_queued_follow_up_draft_front( &self, key: &ThreadKey, @@ -984,6 +1102,7 @@ impl AppStoreReducer { preview, inputs: Vec::new(), source_message_json: None, + turn_start_params: None, }) .collect(); self.set_thread_follow_up_drafts(key, drafts); diff --git a/shared/rust-bridge/codex-mobile-client/src/store/snapshot.rs b/shared/rust-bridge/codex-mobile-client/src/store/snapshot.rs index c3a864b84..85a3ca398 100644 --- a/shared/rust-bridge/codex-mobile-client/src/store/snapshot.rs +++ b/shared/rust-bridge/codex-mobile-client/src/store/snapshot.rs @@ -260,6 +260,10 @@ pub(crate) struct QueuedFollowUpDraft { pub preview: AppQueuedFollowUpPreview, pub inputs: Vec, pub source_message_json: Option, + /// Preserve the exact turn settings selected when the user pressed send. + /// A queued second message must not silently lose its model, effort, mode, + /// or permission policy when it is dispatched after the active turn ends. + pub turn_start_params: Option, } impl ThreadSnapshot { diff --git a/shared/rust-bridge/codex-slingshot/src/json_line_wire.rs b/shared/rust-bridge/codex-slingshot/src/json_line_wire.rs index 00cc2a08f..4b6390fe0 100644 --- a/shared/rust-bridge/codex-slingshot/src/json_line_wire.rs +++ b/shared/rust-bridge/codex-slingshot/src/json_line_wire.rs @@ -15,6 +15,43 @@ use codex_app_server_client::{JsonRpcWire, RemoteAppServerClient, RemoteAppServe use codex_app_server_protocol::JSONRPCMessage; use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader}; +/// Non-Codex bridges can intentionally trail the upstream app-server schema. +/// v0.129 made lifecycle timestamps mandatory; older Pi/Local Studio bridges +/// still emit otherwise-valid item notifications without them. Normalize at +/// the raw JSONL boundary, before `RemoteAppServerClient` strict-decodes and +/// silently drops the entire notification (including tool calls/results). +fn normalize_legacy_item_lifecycle_notification(value: &mut serde_json::Value) { + let Some(message) = value.as_object_mut() else { + return; + }; + let timestamp_field = match message.get("method").and_then(serde_json::Value::as_str) { + Some("item/started") => "startedAtMs", + Some("item/completed") => "completedAtMs", + _ => return, + }; + let Some(params) = message + .get_mut("params") + .and_then(serde_json::Value::as_object_mut) + else { + return; + }; + params + .entry(timestamp_field.to_string()) + .or_insert_with(|| serde_json::Value::Number(0.into())); + if let Some(item) = params + .get_mut("item") + .and_then(serde_json::Value::as_object_mut) + && item.get("type").and_then(serde_json::Value::as_str) == Some("commandExecution") + && item + .get("cwd") + .and_then(serde_json::Value::as_str) + .is_none_or(str::is_empty) + { + // AbsolutePathBuf rejects the empty cwd emitted by older Pi bridges. + item.insert("cwd".to_string(), serde_json::Value::String("/".to_string())); + } +} + struct JsonLineWire { reader: BufReader, writer: W, @@ -64,7 +101,13 @@ where if read == 0 { return Ok(None); } - serde_json::from_str::(&line) + let mut value = serde_json::from_str::(&line).map_err(|err| { + IoError::other(format!( + "remote app server at `{label}` sent invalid JSON-RPC: {err}" + )) + })?; + normalize_legacy_item_lifecycle_notification(&mut value); + serde_json::from_value::(value) .map(Some) .map_err(|err| { IoError::other(format!( @@ -85,6 +128,89 @@ where } } +#[cfg(test)] +mod tests { + use super::normalize_legacy_item_lifecycle_notification; + use codex_app_server_protocol::{JSONRPCMessage, ServerNotification}; + use serde_json::json; + + #[test] + fn legacy_item_lifecycle_notifications_survive_strict_upstream_decode() { + for (method, timestamp_field) in [ + ("item/started", "startedAtMs"), + ("item/completed", "completedAtMs"), + ] { + let mut value = json!({ + "jsonrpc": "2.0", + "method": method, + "params": { + "threadId": "thread-1", + "turnId": "turn-1", + "item": { + "type": "userMessage", + "id": "user-1", + "content": [] + } + } + }); + normalize_legacy_item_lifecycle_notification(&mut value); + assert_eq!(value["params"][timestamp_field], 0); + let message: JSONRPCMessage = serde_json::from_value(value).expect("json-rpc"); + let JSONRPCMessage::Notification(notification) = message else { + panic!("expected notification"); + }; + ServerNotification::try_from(notification) + .expect("upstream should accept normalized lifecycle notification"); + } + } + + #[test] + fn explicit_lifecycle_timestamp_is_never_overwritten() { + let mut value = json!({ + "jsonrpc": "2.0", + "method": "item/completed", + "params": { + "threadId": "thread-1", + "turnId": "turn-1", + "completedAtMs": 42, + "item": { "type": "userMessage", "id": "user-1", "content": [] } + } + }); + normalize_legacy_item_lifecycle_notification(&mut value); + assert_eq!(value["params"]["completedAtMs"], 42); + } + + #[test] + fn legacy_command_item_with_empty_cwd_survives_strict_upstream_decode() { + let mut value = json!({ + "jsonrpc": "2.0", + "method": "item/completed", + "params": { + "threadId": "thread-1", + "turnId": "turn-1", + "item": { + "type": "commandExecution", + "id": "command-1", + "command": "printf OK", + "cwd": "", + "source": "agent", + "status": "completed", + "commandActions": [], + "aggregatedOutput": "OK" + } + } + }); + normalize_legacy_item_lifecycle_notification(&mut value); + assert_eq!(value["params"]["item"]["cwd"], "/"); + let message: JSONRPCMessage = serde_json::from_value(value).expect("json-rpc"); + let JSONRPCMessage::Notification(notification) = message else { + panic!("expected notification"); + }; + ServerNotification::try_from(notification) + .expect("upstream should accept normalized command lifecycle notification"); + } +} + /// Connect a [`RemoteAppServerClient`] over an arbitrary line-delimited JSON-RPC /// stream. Mirrors the API shape of upstream's `connect_websocket_stream`. pub async fn connect_json_line_stream(