diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index ce8b78a176..fba95e8b9e 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -3050,7 +3050,10 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet } if persistence_succeeded { - let status = if execution_result.success && execution_result.has_final_response { + let status = if execution_result.success + && (execution_result.has_final_response + || execution_result.effective_finish_reason == "user_steering") + { AgentTurnSettlementStatus::Completed } else { AgentTurnSettlementStatus::Failed @@ -3060,7 +3063,8 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet turn_id, AgentTurnSettlementResult { status, - final_response: (status == AgentTurnSettlementStatus::Completed) + final_response: (status == AgentTurnSettlementStatus::Completed + && execution_result.has_final_response) .then_some(final_response.clone()), finish_reason: Some(execution_result.effective_finish_reason.clone()), }, @@ -17448,6 +17452,87 @@ mod tests { .expect("clean up persisted test session"); } + #[tokio::test] + async fn steering_handoff_settles_without_fabricating_a_final_response() { + let workspace = tempfile::tempdir().expect("workspace"); + crate::service::workspace::legacy_compat::register_local_fixture_blocking(workspace.path()); + let (coordinator, session_manager) = test_persistent_coordinator(); + let session = session_manager + .create_session( + "Durable completion".to_string(), + "Standard".to_string(), + SessionConfig { + workspace_path: Some(workspace.path().to_string_lossy().into_owned()), + ..Default::default() + }, + ) + .await + .expect("create session"); + let turn_id = session_manager + .start_dialog_turn( + &session.session_id, + "Standard".to_string(), + "finish".to_string(), + Some("turn-durable-fence".to_string()), + None, + None, + ) + .await + .expect("start turn"); + let message = Message::assistant("complete response".to_string()) + .with_turn_id(turn_id.clone()) + .with_round_id("round-final".to_string()); + + ConversationCoordinator::persist_completed_dialog_turn( + coordinator.event_queue.as_ref(), + session_manager.as_ref(), + None, + &session.session_id, + &turn_id, + &ExecutionResult { + final_message: message.clone(), + total_rounds: 1, + success: true, + new_messages: vec![message], + finish_reason: FinishReason::Complete, + total_tools: 0, + duration_ms: 1, + partial_recovery_reason: None, + effective_finish_reason: "user_steering".to_string(), + has_final_response: false, + }, + None, + ) + .await; + + assert_eq!( + session_manager + .turn_settlement_result(&session.session_id, &turn_id) + .and_then(|result| result.final_response), + None + ); + + assert_eq!( + session_manager + .turn_settlement_result(&session.session_id, &turn_id) + .unwrap() + .status, + AgentTurnSettlementStatus::Completed + ); + let events = coordinator.event_queue.dequeue_batch(10).await; + assert!(events.iter().any(|envelope| matches!( + &envelope.event, + AgenticEvent::SessionHistoryChanged { + session_id, + settled_turn_id: Some(settled_turn_id), + } if session_id == &session.session_id && settled_turn_id == &turn_id + ))); + session_manager + .delete_session_by_id(&session.session_id) + .await + .expect("clean up persisted test session"); + } + #[tokio::test] async fn load_relay_session_turns_reads_history_after_the_session_is_unloaded() { let workspace = tempfile::tempdir().expect("workspace"); diff --git a/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs b/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs index d620e29800..565faecea1 100644 --- a/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs +++ b/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs @@ -52,6 +52,22 @@ fn fingerprint(value: &impl serde::Serialize) -> PortResult { Ok(format!("{:x}", Sha256::digest(bytes))) } impl HostQueueState { + pub(super) fn pending_steering_ids( + &self, + session: &str, + target: &str, + ) -> std::collections::HashSet { + self.sessions + .get(session) + .into_iter() + .flat_map(|queue| queue.entries.values()) + .filter(|entry| { + entry.view.status == DialogQueueStatus::SteeringPending + && entry.view.target_turn_id.as_deref() == Some(target) + }) + .filter_map(|entry| entry.view.steering_id.clone()) + .collect() + } pub(super) fn contains(&self, session: &str, turn: &str) -> bool { self.sessions .get(session) @@ -205,6 +221,54 @@ impl HostQueueState { } } impl DialogScheduler { + /// Transfer accepted inputs in acceptance order, preserving their original + /// turn IDs, images, metadata, routing and exactly-once queue receipts. + pub(super) fn release_steering_turns(&self, session: &str, target: &str) -> bool { + let managed = self.queue_state().pending_steering_ids(session, target); + let injections = + self.round_injection_buffer + .drain_matching_for_turn(session, target, |message| managed.contains(&message.id)); + let mut turns = Vec::new(); + { + let mut state = self.queue_state(); + if let Some(s) = state.sessions.get_mut(session) { + for injection in injections { + let entry = s.entries.values_mut().find(|entry| { + entry.view.status == DialogQueueStatus::SteeringPending + && entry.view.target_turn_id.as_deref() == Some(target) + && entry.view.steering_id.as_deref() == Some(&injection.id) + }); + if let Some(entry) = entry { + if let Some(turn) = entry.held.take() { + entry.view.status = DialogQueueStatus::Queued; + entry.view.target_turn_id = None; + entry.view.steering_id = None; + entry.view.reason = None; + s.revision += 1; + turns.push(turn); + } + } + } + } + } + let released = !turns.is_empty(); + for turn in turns.into_iter().rev() { + self.requeue_front(session, turn); + } + released + } + pub(super) fn has_queued_host_message(&self, session: &str) -> bool { + self.queue_state() + .sessions + .get(session) + .is_some_and(|queue| { + queue + .entries + .values() + .any(|entry| entry.view.status == DialogQueueStatus::Queued) + }) + } + pub(super) fn hold_managed_queue(&self, session: &str, reason: &str) { self.hold_managed_queue_for_outcome(session, reason, None); } @@ -580,14 +644,7 @@ impl DialogScheduler { steering_id.clone(), SystemTime::now(), ); - if let DialogSteeringAction::Buffer { mut injection, .. } = decision { - if let Err(reason) = self - .prepare_goal_steering(session, target, &mut injection) - .await - { - self.queue_state().hold(session, &turn, &reason); - return Err(error(&reason)); - } + if let DialogSteeringAction::Buffer { injection, .. } = decision { { let mut state = self.queue_state(); let e = state diff --git a/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs b/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs index e47a70363c..d0ded77f4e 100644 --- a/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs +++ b/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs @@ -103,7 +103,7 @@ async fn host_queue_cancel_is_idempotent_and_never_cancels_active_turn() { } #[tokio::test] -async fn host_queue_steering_retains_payload_until_consumption_and_rejects_cancel() { +async fn host_queue_steering_retains_payload_until_formal_turn_handoff_and_rejects_cancel() { let (scheduler, _, _root, epoch) = fixture().await; scheduler .manage_host_queue(request( @@ -128,7 +128,13 @@ async fn host_queue_steering_retains_payload_until_consumption_and_rejects_cance let injections = scheduler .round_injection_source .take_pending("host-queue-session", "active-turn"); - assert_eq!(injections.len(), 1); + assert!( + injections.is_empty(), + "human inputs must not become inline injections" + ); + assert!(scheduler + .round_injection_source + .should_yield_to_user_turn("host-queue-session", "active-turn")); assert_eq!( scheduler.queue_depth("host-queue-session"), 1, @@ -146,13 +152,8 @@ async fn host_queue_steering_retains_payload_until_consumption_and_rejects_cance .unwrap_err() .message .contains("too_late")); - let injection = &injections[0]; - scheduler.round_injection_source.acknowledge_consumed( - "host-queue-session", - "active-turn", - &injection.id, - injection.kind, - ); + assert!(scheduler.release_steering_turns("host-queue-session", "active-turn")); + assert!(!scheduler.release_steering_turns("host-queue-session", "active-turn")); let result = scheduler .manage_host_queue(request( Some(&epoch), @@ -162,8 +163,17 @@ async fn host_queue_steering_retains_payload_until_consumption_and_rejects_cance )) .await .unwrap(); - assert_eq!(result.receipt.unwrap().status, Status::Steered); - assert_eq!(result.used, 0); + assert_eq!(result.receipt.unwrap().status, Status::Queued); + assert_eq!(result.used, 1); + assert_eq!(scheduler.queues.depth("host-queue-session"), 1); + assert_eq!( + scheduler + .dequeue_next("host-queue-session") + .unwrap() + .turn_id + .as_deref(), + Some("queued-a") + ); } #[tokio::test] @@ -1123,7 +1133,7 @@ fn host_queue_new_prompt_after_error_survives_delayed_failure_cleanup() { } #[tokio::test] -async fn thread_goal_host_queue_promote_activates_once() { +async fn thread_goal_host_queue_promote_preserves_one_unmodified_user_turn() { let (scheduler, sessions, _, root) = test_scheduler_with_persistence(true); mark_session_processing(&sessions, &root, "host-queue-session", "active-turn").await; scheduler @@ -1159,20 +1169,254 @@ async fn thread_goal_host_queue_promote_activates_once() { .effective_session_storage_path("host-queue-session") .await .unwrap(); - let goal = scheduler + assert!(scheduler .coordinator .get_thread_goal("host-queue-session", &storage) .await .unwrap() + .is_none()); + assert!(scheduler.release_steering_turns("host-queue-session", "active-turn")); + let turn = scheduler.dequeue_next("host-queue-session").unwrap(); + assert_eq!(turn.turn_id.as_deref(), Some("queued-goal")); + assert_eq!(turn.user_input, "/goal finish queued work"); + assert!(!scheduler.release_steering_turns("host-queue-session", "active-turn")); +} + +#[test] +fn steering_creates_persisted_user_turns_in_order_without_a_connected_caller() { + run_host_queue_lifecycle_test(|| async { + let (scheduler, sessions, _, root) = test_scheduler_with_persistence(true); + let id = "host-queue-session"; + let workspace = fixture_workspace_dir(root.path().join("steering-history")); + sessions + .create_session_with_id( + Some(id.into()), + "Steering history".into(), + "Standard".into(), + SessionConfig { + workspace_path: Some(workspace.to_string_lossy().into_owned()), + ..Default::default() + }, + ) + .await + .unwrap(); + sessions + .start_dialog_turn( + id, + "Standard".into(), + "first question".into(), + Some("original-turn".into()), + None, + None, + ) + .await + .unwrap(); + scheduler + .active_turns + .insert(id, desktop_active_turn("original-turn")); + TEST_MODEL_RESOLUTION_AI_CONFIG + .scope( + host_queue_lifecycle_model(), + sessions.update_session_model_id(id, "queue-lifecycle-model"), + ) + .await + .unwrap(); + let epoch = scheduler + .manage_host_queue(request(None, Action::List)) + .await + .unwrap() + .queue_epoch; + for turn_id in ["waiting", "steer-a", "steer-b"] { + let mut input = message(turn_id); + input.content = format!("content {turn_id}"); + input.metadata.insert( + "context".into(), + serde_json::json!({"path":"/workspace/source.rs"}), + ); + scheduler + .manage_host_queue(request(Some(&epoch), Action::Submit { message: input })) + .await + .unwrap(); + } + for turn_id in ["steer-a", "steer-b"] { + let promote = request( + Some(&epoch), + Action::Promote { + turn_id: turn_id.into(), + operation_id: format!("promote-{turn_id}"), + expected_active_turn_id: Some("original-turn".into()), + }, + ); + scheduler.manage_host_queue(promote.clone()).await.unwrap(); + scheduler.manage_host_queue(promote).await.unwrap(); + } + assert!(scheduler + .round_injection_source + .should_yield_to_user_turn(id, "original-turn")); + assert!(scheduler + .round_injection_source + .take_pending(id, "original-turn") + .is_empty()); + let storage = sessions.effective_session_storage_path(id).await.unwrap(); + scheduler + .coordinator + .create_thread_goal(id, &storage, "finish the requested work".into(), Some(1000)) + .await + .unwrap(); + scheduler + .coordinator + .thread_goal_runtime(id) + .record_round_billable_tokens("original-turn", 25); + + // Settle real persistent turns through the same owner used by execution. + // No further mutation from the submitting mobile/desktop caller is needed. + for (finished, next, index) in [("original-turn", "steer-a", 1), ("steer-a", "steer-b", 2)] + { + let answer = + Message::assistant(format!("response {finished}")).with_turn_id(finished.into()); + sessions + .complete_dialog_turn( + id, + finished, + format!("response {finished}"), + &[answer], + crate::agentic::core::TurnStats::default(), + Some("user_steering".into()), + Some(false), + ) + .await + .unwrap(); + sessions + .update_session_state(id, SessionState::Idle) + .await + .unwrap(); + process_host_queue_outcome( + &scheduler, + TurnOutcome::Completed { + turn_id: finished.into(), + final_response: String::new(), + }, + ) + .await; + assert!(scheduler.active_turns.matches_turn(id, next)); + let goal = scheduler + .coordinator + .get_thread_goal(id, &storage) + .await + .unwrap() + .unwrap(); + assert_eq!( + goal.tokens_used, 25, + "handoff preserves old-turn accounting" + ); + assert_eq!( + goal.auto_continuation_count, 0, + "human turn needs no automatic continuation" + ); + + assert_eq!(sessions.get_turn_count(id), index + 1); + let persisted = sessions + .persistence_manager() + .load_dialog_turn(&storage, id, index) + .await + .unwrap() + .unwrap(); + assert_eq!(persisted.turn_id, next); + assert_eq!(persisted.user_message.content, format!("content {next}")); + assert_eq!( + persisted.user_message.metadata.as_ref().unwrap()["context"]["path"], + "/workspace/source.rs" + ); + let preceding = sessions + .persistence_manager() + .load_dialog_turn(&storage, id, index - 1) + .await + .unwrap() + .unwrap(); + assert_eq!(preceding.turn_id, finished); + assert!( + !preceding.model_rounds.is_empty(), + "previous output stays with the previous user message" + ); + // Duplicate old outcomes must not retire or duplicate the new execution. + process_host_queue_outcome( + &scheduler, + TurnOutcome::Completed { + turn_id: finished.into(), + final_response: String::new(), + }, + ) + .await; + assert!(scheduler.active_turns.matches_turn(id, next)); + assert_eq!(sessions.get_turn_count(id), index + 1); + } + assert_eq!( + scheduler.dequeue_next(id).unwrap().turn_id.as_deref(), + Some("waiting") + ); + let _ = scheduler + .coordinator + .cancel_dialog_turn(id, "steer-b") + .await; + }); +} + +#[tokio::test] +async fn host_queue_handoff_preserves_turn_scoped_sdk_steering() { + let (scheduler, _, _root, epoch) = fixture().await; + scheduler + .buffer_steering( + "host-queue-session".into(), + "active-turn".into(), + "SDK input".into(), + None, + Vec::new(), + serde_json::Map::new(), + ) + .await .unwrap(); - assert!(goal.is_active()); - assert_eq!(goal.objective, "finish queued work"); - let injections = scheduler + assert!(!scheduler + .round_injection_source + .should_yield_to_user_turn("host-queue-session", "active-turn")); + scheduler + .manage_host_queue(request( + Some(&epoch), + Action::Submit { + message: message("queued-user"), + }, + )) + .await + .unwrap(); + scheduler + .manage_host_queue(request( + Some(&epoch), + Action::Promote { + turn_id: "queued-user".into(), + operation_id: "promote-user".into(), + expected_active_turn_id: Some("active-turn".into()), + }, + )) + .await + .unwrap(); + let inline = scheduler .round_injection_source .take_pending("host-queue-session", "active-turn"); - assert_eq!(injections.len(), 1); - assert_eq!(injections[0].display_content, "/goal finish queued work"); - assert!(injections[0] - .content - .contains("\nfinish queued work")); + assert_eq!(inline.len(), 1); + assert_eq!(inline[0].content, "SDK input"); + assert!(scheduler + .round_injection_source + .should_yield_to_user_turn("host-queue-session", "active-turn")); + assert!(scheduler.release_steering_turns("host-queue-session", "active-turn")); + assert!(!scheduler + .round_injection_source + .should_yield_to_user_turn("host-queue-session", "active-turn")); + assert_eq!( + scheduler + .dequeue_next("host-queue-session") + .unwrap() + .turn_id + .as_deref(), + Some("queued-user") + ); + assert!(!scheduler.queues.has_items("host-queue-session")); } diff --git a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs index 2392341cac..490c3d3189 100644 --- a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs +++ b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs @@ -321,7 +321,27 @@ impl DialogRoundInjectionSource for SchedulerRoundInjectionSource { } fn take_pending(&self, session_id: &str, turn_id: &str) -> Vec { - self.buffer.drain_for_turn(session_id, turn_id) + // Hold receipt state through the drain: promotion registers its receipt + // before publishing the injection. A racing promotion must never be + // mistaken for the turn-scoped SDK's legacy inline input. + let queue = self + .host_queue + .lock() + .unwrap_or_else(|error| error.into_inner()); + let managed = queue.pending_steering_ids(session_id, turn_id); + self.buffer + .drain_matching_for_turn(session_id, turn_id, |message| { + !managed.contains(&message.id) + }) + } + + fn should_yield_to_user_turn(&self, session_id: &str, turn_id: &str) -> bool { + !self + .host_queue + .lock() + .unwrap_or_else(|error| error.into_inner()) + .pending_steering_ids(session_id, turn_id) + .is_empty() } fn acknowledge_consumed( @@ -2671,7 +2691,7 @@ impl DialogScheduler { mut outcome_rx: mpsc::UnboundedReceiver<(String, TurnOutcome)>, ) { while let Some((session_id, outcome)) = outcome_rx.recv().await { - let (active_turn, active_internal_turn, lifecycle_plan) = { + let (active_turn, active_internal_turn, lifecycle_plan, has_user_successor) = { let _operation_guard = self.lock_session_operation(&session_id).await; let Some(active_turn_result) = take_active_turn_for_outcome( &self.active_turns, @@ -2714,6 +2734,12 @@ impl DialogScheduler { }); let lifecycle_plan = resolve_turn_outcome_lifecycle_plan(&outcome, active_turn.is_some()); + if active_turn.is_some() && lifecycle_plan.status == TurnOutcomeStatus::Completed { + self.release_steering_turns(&session_id, outcome.turn_id()); + } + // Include inputs already released by an earlier handoff. Each + // queued human turn takes priority over automatic goal follow-ups. + let has_user_successor = self.has_queued_host_message(&session_id); let retired_injections = self .host_queue .lock() @@ -2755,7 +2781,12 @@ impl DialogScheduler { self.requeue_front(&session_id, turn); } } - (active_turn, active_internal_turn, lifecycle_plan) + ( + active_turn, + active_internal_turn, + lifecycle_plan, + has_user_successor, + ) }; let status = lifecycle_plan.status; let queue_action = lifecycle_plan.queue_action; @@ -2827,7 +2858,9 @@ impl DialogScheduler { outcome.turn_id(), active_turn.user_input(), active_turn.user_message_metadata(), - turn_completed, + // Account usage, but the accepted human turn + // replaces automatic goal continuation. + turn_completed && !has_user_successor, ) .await { diff --git a/src/crates/assembly/core/src/agentic/execution/execution_engine.rs b/src/crates/assembly/core/src/agentic/execution/execution_engine.rs index 13f62ca7cc..85774e633f 100644 --- a/src/crates/assembly/core/src/agentic/execution/execution_engine.rs +++ b/src/crates/assembly/core/src/agentic/execution/execution_engine.rs @@ -4201,11 +4201,8 @@ impl ExecutionEngine { } } - // User-steering messages submitted while this turn is running: drain and inject - // them as user messages into the working history before starting the next round - // (Codex-style mid-turn injection). This does NOT end the current turn: if the - // model wanted to finish but the user steered, we keep the turn running so the - // steering message gets a response. + // Human steering becomes the next persisted user turn. Runtime + // reminders still enter this turn through the injection channel. let mut injection_applied = false; if let Some(source) = context.round_injection.as_ref() { let pending = source.take_pending(&context.session_id, &context.dialog_turn_id); @@ -4301,6 +4298,17 @@ impl ExecutionEngine { injection_applied = true; } } + if source.should_yield_to_user_turn(&context.session_id, &context.dialog_turn_id) + && !self + .round_executor + .is_dialog_turn_cancelled(&dialog_turn_id) + { + // Persist runtime reminders before handing off, so background + // results arriving at this boundary stay in session context. + // Already-started tools and results keep their original turn. + finalization_reason = Some("user_steering"); + break; + } } // P0-1: Decide whether to end the turn here. @@ -4678,7 +4686,7 @@ impl ExecutionEngine { let success = has_final_response || matches!( effective_finish_reason, - "max_rounds" | "repeated_tool_failures" + "max_rounds" | "repeated_tool_failures" | "user_steering" ); // Post-processing hook: when a DeepResearch dialog turn finishes @@ -4689,7 +4697,7 @@ impl ExecutionEngine { { if openbitfun_agent_workflows::deep_research::should_post_process_research_report( &agent_type, - success, + success && effective_finish_reason != "user_steering", ) { if let Some(workspace) = context.workspace.as_ref() { if let Some(workspace_services) = context.workspace_services.as_ref() { diff --git a/src/crates/assembly/core/src/agentic/execution/types.rs b/src/crates/assembly/core/src/agentic/execution/types.rs index 267e69fec0..7fda23e24d 100644 --- a/src/crates/assembly/core/src/agentic/execution/types.rs +++ b/src/crates/assembly/core/src/agentic/execution/types.rs @@ -43,8 +43,8 @@ pub struct ExecutionContext { pub terminal_port: Option>, /// Remote execution provider injected by product assembly. pub remote_exec_port: Option>, - /// When set, engine drains pending round injections at each round boundary - /// and injects them into the dialog history without ending the turn. + /// At round boundaries, yield to a queued user turn when requested; other + /// pending runtime injections remain within this execution. pub round_injection: Option>, /// When false, the execution loop suppresses user-facing turn lifecycle events. pub emit_lifecycle_events: bool, diff --git a/src/crates/contracts/runtime-ports/src/agent_api.rs b/src/crates/contracts/runtime-ports/src/agent_api.rs index f2f5bc87b9..252c0b0767 100644 --- a/src/crates/contracts/runtime-ports/src/agent_api.rs +++ b/src/crates/contracts/runtime-ports/src/agent_api.rs @@ -1041,6 +1041,13 @@ pub trait DialogRoundInjectionSource: Send + Sync { ) -> RoundInjectionToolPreemption; fn take_pending(&self, session_id: &str, turn_id: &str) -> Vec; + /// End the current execution at an atomic boundary so the scheduler can + /// start an accepted user message as a normal, persisted dialog turn. + /// Legacy providers retain their inline-injection behavior by default. + fn should_yield_to_user_turn(&self, _session_id: &str, _turn_id: &str) -> bool { + false + } + fn acknowledge_consumed( &self, _session_id: &str, diff --git a/src/crates/execution/agent-runtime/src/scheduler.rs b/src/crates/execution/agent-runtime/src/scheduler.rs index 180bd5b6f8..904cbea149 100644 --- a/src/crates/execution/agent-runtime/src/scheduler.rs +++ b/src/crates/execution/agent-runtime/src/scheduler.rs @@ -580,12 +580,25 @@ impl SessionRoundInjectionBuffer { /// injections that target a different turn are retained until the targeted /// turn consumes them or the session is cleared. pub fn drain_for_turn(&self, session_id: &str, turn_id: &str) -> Vec { + self.drain_matching_for_turn(session_id, turn_id, |_| true) + } + + pub fn drain_matching_for_turn( + &self, + session_id: &str, + turn_id: &str, + matches: impl Fn(&RoundInjection) -> bool, + ) -> Vec { let Some(mut entry) = self.inner.get_mut(session_id) else { return Vec::new(); }; let mut taken = Vec::new(); let mut keep = Vec::new(); for msg in entry.drain(..) { + if !matches(&msg) { + keep.push(msg); + continue; + } match &msg.target { RoundInjectionTarget::ExactTurn(target_turn_id) if target_turn_id == turn_id => { taken.push(msg); diff --git a/src/crates/execution/agent-runtime/tests/agent_session_contracts/scheduler_contracts.rs b/src/crates/execution/agent-runtime/tests/agent_session_contracts/scheduler_contracts.rs index 29589ecf34..960eaf015f 100644 --- a/src/crates/execution/agent-runtime/tests/agent_session_contracts/scheduler_contracts.rs +++ b/src/crates/execution/agent-runtime/tests/agent_session_contracts/scheduler_contracts.rs @@ -654,6 +654,30 @@ fn round_injection_buffer_drains_only_messages_for_the_active_turn() { assert_eq!(buffer.pending_count("s1"), 0); } +#[test] +fn human_turn_handoff_leaves_runtime_reminders_and_other_targets_intact() { + let buffer = SessionRoundInjectionBuffer::default(); + buffer.push("s1", exact_turn_msg("turn-a", "first")); + buffer.push("s1", current_turn_msg("background")); + buffer.push("s1", exact_turn_msg("turn-b", "other")); + buffer.push("s1", exact_turn_msg("turn-a", "second")); + let human = buffer.drain_matching_for_turn("s1", "turn-a", |message| { + message.kind == RoundInjectionKind::UserSteering + }); + assert_eq!( + human + .iter() + .map(|message| message.content.as_str()) + .collect::>(), + vec!["first", "second"] + ); + assert_eq!( + buffer.drain_for_turn("s1", "turn-a")[0].content, + "background" + ); + assert_eq!(buffer.drain_for_turn("s1", "turn-b")[0].content, "other"); +} + #[test] fn round_injection_buffer_removes_one_exact_delivery_by_id() { let buffer = SessionRoundInjectionBuffer::default(); diff --git a/src/mobile-web/README.md b/src/mobile-web/README.md index df0d423a55..1f626e5fdc 100644 --- a/src/mobile-web/README.md +++ b/src/mobile-web/README.md @@ -80,8 +80,14 @@ turn runs. Accepted follow-ups appear in the host message queue, shared with the desktop and supported Peer Device controllers. Closing this page, disconnecting the phone, or leaving the session does not stop host-side dispatch. -- **Send now** starts the selected message when idle or steers it into the - observed active turn. Waiting for steering is distinct from being consumed. +- **Send now** starts the selected message when idle. While a turn runs, the + host finishes the current atomic action and starts the selected message as + the next regular user turn. It has its own history and navigation entry on + desktop and mobile; existing tool results stay with the preceding turn. + Acceptance is distinct from the new turn actually starting. Older hosts may + still render steering inline within the active turn. The turn-scoped SDK + steering API used by CLI/Dispatch retains its inline contract; this handoff + belongs to host queue promotion. - **Remove from queue** only removes an unstarted message. It never stops the active turn. An operation that lost a race with dispatch is rejected. - A failed turn or unconsumed steering retains the message as blocked on the diff --git a/src/mobile-web/src/components/ChatComposerBar.tsx b/src/mobile-web/src/components/ChatComposerBar.tsx index 7f1c322cb0..ec6857a7e3 100644 --- a/src/mobile-web/src/components/ChatComposerBar.tsx +++ b/src/mobile-web/src/components/ChatComposerBar.tsx @@ -82,19 +82,6 @@ export default function ChatComposerBar({ size="sm" /> ) : null} - {streaming && ( -