hanzoai/dev
2
0
Fork 0
mirror of https://github.com/hanzoai/dev.git synced 2026-10-25 03:32:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(request_user_input): route answers via UserInputAnswer

Browse Source
This commit is contained in:
James Peter
2026-01-22 10:21:32 +10:00
parent 7aad17a511
commit b972185d73
7 changed files with 220 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
code-rs/core/src/codex/session.rs
+19 -12
View File
@@ -132,7 +132,7 @@ pub(super) struct State {
pub(super) approved_commands: HashSet<ApprovedCommandPattern>,
pub(super) current_task: Option<AgentTask>,
pub(super) pending_approvals: HashMap<String, oneshot::Sender<ReviewDecision>>,
pub(super) pending_tool_user_input: VecDeque<oneshot::Sender<Vec<InputItem>>>,
pub(super) pending_request_user_input: HashMap<String, oneshot::Sender<crate::protocol::RequestUserInputResponse>>,
pub(super) pending_input: Vec<ResponseInputItem>,
pub(super) pending_user_input: Vec<QueuedUserInput>,
pub(super) history: ConversationHistory,
@@ -1120,25 +1120,32 @@ impl Session {
}
}
pub async fn request_tool_user_input(&self) -> Option<Vec<InputItem>> {
pub fn register_pending_user_input(
&self,
turn_id: String,
) -> std::result::Result<oneshot::Receiver<crate::protocol::RequestUserInputResponse>, String> {
let (tx, rx) = oneshot::channel();
{
let mut state = self.state.lock().unwrap();
state.pending_tool_user_input.push_back(tx);
let mut state = self.state.lock().unwrap();
if state.pending_request_user_input.contains_key(&turn_id) {
return Err(format!("request_user_input already pending for turn_id={turn_id}"));
}
rx.await.ok()
state.pending_request_user_input.insert(turn_id, tx);
Ok(rx)
}
pub fn notify_tool_user_input(&self, items: Vec<InputItem>) -> Option<Vec<InputItem>> {
pub fn notify_user_input_response(
&self,
turn_id: &str,
response: crate::protocol::RequestUserInputResponse,
) {
let pending = {
let mut state = self.state.lock().unwrap();
state.pending_tool_user_input.pop_front()
state.pending_request_user_input.remove(turn_id)
};
if let Some(tx) = pending {
let _ = tx.send(items);
None
let _ = tx.send(response);
} else {
Some(items)
tracing::warn!("no pending request_user_input found for turn_id={turn_id}");
}
}
@@ -1855,7 +1862,7 @@ impl Session {
let mut state = self.state.lock().unwrap();
state.pending_approvals.clear();
state.pending_tool_user_input.clear();
state.pending_request_user_input.clear();
// Do not clear `pending_input` here. When a user submits a new message
// immediately after an interrupt, it may have been routed to
// `pending_input` by an earlier code path. Clearing it would drop the
code-rs/core/src/codex/streaming.rs
+36 -86
View File
@@ -804,14 +804,6 @@ pub(super) async fn submission_loop(
}
};
let items = match sess.notify_tool_user_input(items) {
Some(items) => items,
None => {
// Consumed as input for a pending `request_user_input` tool call.
continue;
}
};
// Clean up old status items when new user input arrives
// This prevents token buildup from old screenshots/status messages
sess.cleanup_old_status_items().await;
@@ -835,14 +827,6 @@ pub(super) async fn submission_loop(
}
};
let items = match sess.notify_tool_user_input(items) {
Some(items) => items,
None => {
// Consumed as input for a pending `request_user_input` tool call.
continue;
}
};
if sess.has_running_task() {
let mut response_item = response_input_from_core_items(items.clone());
sess.enforce_user_message_limits(&sub.id, &mut response_item);
@@ -877,6 +861,16 @@ pub(super) async fn submission_loop(
other => sess.notify_approval(&id, other),
}
}
Op::UserInputAnswer { id, response } => {
let sess = match sess.as_ref() {
Some(sess) => sess,
None => {
send_no_session_event(sub.id).await;
continue;
}
};
sess.notify_user_input_response(&id, response);
}
Op::RegisterApprovedCommand {
command,
match_kind,
@@ -3035,9 +3029,8 @@ async fn handle_request_user_input(
ctx: &ToolCallCtx,
arguments: String,
) -> ResponseInputItem {
use code_protocol::request_user_input::RequestUserInputAnswer;
use code_protocol::request_user_input::RequestUserInputArgs;
use code_protocol::request_user_input::RequestUserInputResponse;
use code_protocol::request_user_input::RequestUserInputEvent;
let args: RequestUserInputArgs = match serde_json::from_str(&arguments) {
Ok(args) => args,
@@ -3062,84 +3055,41 @@ async fn handle_request_user_input(
};
}
let mut prompt = String::from("The model requested user input:\n");
for (idx, q) in args.questions.iter().enumerate() {
prompt.push_str(&format!(
"\n{}. {} ({})\n{}\n",
idx + 1,
q.header,
q.id,
q.question
));
if let Some(options) = &q.options {
prompt.push_str("Options:\n");
for option in options {
prompt.push_str(&format!("- {}: {}\n", option.label, option.description));
}
let rx_response = match sess.register_pending_user_input(ctx.sub_id.clone()) {
Ok(rx) => rx,
Err(err) => {
return ResponseInputItem::FunctionCallOutput {
call_id: ctx.call_id.clone(),
output: FunctionCallOutputPayload {
content: err,
success: Some(false),
},
};
}
}
if args.questions.len() == 1 {
prompt.push_str("\nReply with your answer.");
} else {
prompt.push_str("\nReply with one line per question (in order). Extra lines are attached to the final question.");
}
};
sess.send_ordered_from_ctx(
ctx,
EventMsg::AgentMessage(AgentMessageEvent {
message: prompt,
EventMsg::RequestUserInput(RequestUserInputEvent {
call_id: ctx.call_id.clone(),
turn_id: ctx.sub_id.clone(),
questions: args.questions,
}),
)
.await;
let user_items = sess.request_tool_user_input().await;
let Some(user_items) = user_items else {
return ResponseInputItem::FunctionCallOutput {
call_id: ctx.call_id.clone(),
output: FunctionCallOutputPayload {
content: "request_user_input was cancelled before receiving a response".to_string(),
success: Some(false),
},
};
};
let user_text = user_items
.iter()
.filter_map(|item| match item {
InputItem::Text { text } => Some(text.trim().to_string()),
_ => None,
})
.collect::<Vec<_>>()
.join("\n")
.trim()
.to_string();
let response: RequestUserInputResponse = serde_json::from_str(&user_text).unwrap_or_else(|_| {
let mut answers = std::collections::HashMap::new();
let mut lines: Vec<String> = user_text
.lines()
.map(|line| line.trim_end().to_string())
.collect();
if args.questions.len() == 1 {
lines = vec![user_text.clone()];
} else if lines.len() > args.questions.len() {
let tail = lines.split_off(args.questions.len() - 1);
lines.push(tail.join("\n"));
}
for (idx, q) in args.questions.iter().enumerate() {
let value = lines.get(idx).cloned().unwrap_or_default();
answers.insert(
q.id.clone(),
RequestUserInputAnswer {
answers: vec![value],
let response = match rx_response.await {
Ok(response) => response,
Err(_) => {
return ResponseInputItem::FunctionCallOutput {
call_id: ctx.call_id.clone(),
output: FunctionCallOutputPayload {
content: "request_user_input was cancelled before receiving a response".to_string(),
success: Some(false),
},
);
};
}
RequestUserInputResponse { answers }
});
};
let content = match serde_json::to_string(&response) {
Ok(content) => content,
code-rs/core/src/protocol.rs
+13
View File
@@ -46,6 +46,8 @@ pub use code_protocol::skills::Skill;
pub use code_protocol::protocol::ENVIRONMENT_CONTEXT_OPEN_TAG;
pub use code_protocol::protocol::ExitedReviewModeEvent;
pub use code_protocol::protocol::ReviewSnapshotInfo;
pub use code_protocol::request_user_input::RequestUserInputEvent;
pub use code_protocol::request_user_input::RequestUserInputResponse;
/// Submission Queue Entry - requests from user
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -195,6 +197,15 @@ pub enum Op {
decision: ReviewDecision,
},
/// Resolve a request_user_input tool call.
#[serde(rename = "user_input_answer", alias = "request_user_input_response")]
UserInputAnswer {
/// Turn id for the in-flight request.
id: String,
/// User-provided answers.
response: RequestUserInputResponse,
},
/// Update a specific validation tool toggle for the session.
UpdateValidationTool {
name: String,
@@ -787,6 +798,8 @@ pub enum EventMsg {
ExecApprovalRequest(ExecApprovalRequestEvent),
RequestUserInput(RequestUserInputEvent),
ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent),
BackgroundEvent(BackgroundEventEvent),
code-rs/exec/src/event_processor_with_human_output.rs
+7
View File
@@ -202,6 +202,13 @@ impl EventProcessor for EventProcessorWithHumanOutput {
EventMsg::BackgroundEvent(BackgroundEventEvent { message }) => {
ts_println!(self, "{}", message.style(self.dimmed));
}
EventMsg::RequestUserInput(ev) => {
let question_count = ev.questions.len();
ts_println!(
self,
"Model requested user input ({question_count} question(s)); interactive prompts are not supported in this runner.",
);
}
EventMsg::EnvironmentContextFull(_)
| EventMsg::EnvironmentContextDelta(_)
| EventMsg::BrowserSnapshot(_)
code-rs/mcp-server/src/code_tool_runner.rs
+20
View File
@@ -232,6 +232,26 @@ async fn run_code_tool_session_inner(
.await;
continue;
}
EventMsg::RequestUserInput(ev) => {
let question_count = ev.questions.len();
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_string(),
text: format!(
"Codex requested user input ({question_count} question(s)) but interactive prompts are not supported in this MCP tool session."
),
annotations: None,
})],
is_error: Some(true),
structured_content: None,
};
outgoing.send_response(request_id.clone(), result).await;
running_requests_id_to_code_uuid
.lock()
.await
.remove(&request_id);
break;
}
EventMsg::TaskComplete(TaskCompleteEvent { last_agent_message }) => {
let text = match last_agent_message {
Some(msg) => msg.clone(),
code-rs/protocol/src/protocol.rs
+13
View File
@@ -21,8 +21,10 @@ use crate::models::ResponseItem;
use crate::num_format::format_with_separators;
use crate::parse_command::ParsedCommand;
use crate::plan_tool::UpdatePlanArgs;
use crate::request_user_input::RequestUserInputResponse;
use mcp_types::CallToolResult;
use mcp_types::Tool as McpTool;
pub use crate::request_user_input::RequestUserInputEvent;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
@@ -167,6 +169,15 @@ pub enum Op {
decision: ReviewDecision,
},
/// Resolve a request_user_input tool call.
#[serde(rename = "user_input_answer", alias = "request_user_input_response")]
UserInputAnswer {
/// Turn id for the in-flight request.
id: String,
/// User-provided answers.
response: RequestUserInputResponse,
},
/// Append an entry to the persistent cross-session message history.
///
/// Note the entry is not guaranteed to be logged if the user has
@@ -573,6 +584,8 @@ pub enum EventMsg {
ExecApprovalRequest(ExecApprovalRequestEvent),
RequestUserInput(RequestUserInputEvent),
ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent),
BackgroundEvent(BackgroundEventEvent),
code-rs/tui/src/chatwidget.rs
+112 -1
View File
@@ -1463,6 +1463,12 @@ pub(crate) enum TurnOrigin {
Developer,
}
#[derive(Clone, Debug)]
struct PendingRequestUserInput {
turn_id: String,
questions: Vec<code_protocol::request_user_input::RequestUserInputQuestion>,
}
#[derive(Clone)]
struct RenderRequestSeed {
history_id: HistoryId,
@@ -1545,6 +1551,7 @@ pub(crate) struct ChatWidget<'a> {
// Cache of the last developer/system note we injected (hidden messages)
last_developer_message: Option<String>,
pending_turn_origin: Option<TurnOrigin>,
pending_request_user_input: Option<PendingRequestUserInput>,
current_turn_origin: Option<TurnOrigin>,
// Tracks whether lingering running exec/tool cells have been cleared for the
// current turn. Reset on TaskStarted; set after the first assistant message
@@ -6439,6 +6446,7 @@ impl ChatWidget<'_> {
last_user_message: None,
last_developer_message: None,
pending_turn_origin: None,
pending_request_user_input: None,
current_turn_origin: None,
cleared_lingering_execs_this_turn: true,
exec: ExecState {
@@ -6794,6 +6802,7 @@ impl ChatWidget<'_> {
last_user_message: None,
last_developer_message: None,
pending_turn_origin: None,
pending_request_user_input: None,
current_turn_origin: None,
cleared_lingering_execs_this_turn: true,
exec: ExecState {
@@ -8065,6 +8074,10 @@ impl ChatWidget<'_> {
match input_result {
InputResult::Submitted(text) => {
if let Some(pending) = self.pending_request_user_input.take() {
self.submit_request_user_input_answer(pending, text);
return;
}
self.pending_turn_origin = Some(TurnOrigin::User);
let cleaned = Self::strip_context_sections(&text);
self.last_user_message = (!cleaned.trim().is_empty()).then_some(cleaned);
@@ -10882,6 +10895,63 @@ impl ChatWidget<'_> {
}
}
fn submit_request_user_input_answer(&mut self, pending: PendingRequestUserInput, raw: String) {
use code_protocol::request_user_input::RequestUserInputAnswer;
use code_protocol::request_user_input::RequestUserInputResponse;
let display_text = raw.trim();
if !display_text.is_empty() {
let key = self.near_time_key_current_req(None);
let state = history_cell::new_user_prompt(display_text.to_string());
let _ =
self.history_insert_plain_state_with_key(state, key, "request_user_input_answer");
self.restore_reasoning_in_progress_if_streaming();
}
let response = serde_json::from_str::<RequestUserInputResponse>(&raw).unwrap_or_else(|_| {
let question_count = pending.questions.len();
let mut lines: Vec<String> = raw
.lines()
.map(|line| line.trim_end().to_string())
.collect();
if question_count <= 1 {
lines = vec![raw.trim().to_string()];
} else if lines.len() > question_count {
let tail = lines.split_off(question_count - 1);
lines.push(tail.join("\n"));
}
while lines.len() < question_count {
lines.push(String::new());
}
let mut answers = std::collections::HashMap::new();
for (idx, question) in pending.questions.iter().enumerate() {
let value = lines.get(idx).cloned().unwrap_or_default();
answers.insert(
question.id.clone(),
RequestUserInputAnswer {
answers: vec![value],
},
);
}
RequestUserInputResponse { answers }
});
if let Err(e) = self.code_op_tx.send(Op::UserInputAnswer {
id: pending.turn_id,
response,
}) {
tracing::error!("failed to send Op::UserInputAnswer: {e}");
}
self.clear_composer();
self.bottom_pane
.update_status_text("waiting for model".to_string());
self.request_redraw();
}
fn submit_user_message(&mut self, user_message: UserMessage) {
if self.layout.scroll_offset > 0 {
layout_scroll::to_bottom(self);
@@ -13309,6 +13379,7 @@ impl ChatWidget<'_> {
// This begins the new turn; clear the pending prompt anchor count
// so subsequent background events use standard placement.
self.pending_user_prompts_for_next_turn = 0;
self.pending_request_user_input = None;
// Reset stream headers for new turn
self.stream.reset_headers_for_new_turn();
self.stream_state.current_kind = None;
@@ -13348,6 +13419,7 @@ impl ChatWidget<'_> {
}
EventMsg::TaskComplete(TaskCompleteEvent { last_agent_message }) => {
self.clear_reconnecting();
self.pending_request_user_input = None;
let had_running_execs = !self.exec.running_commands.is_empty();
// Finalize any active streams
let finalizing_streams = self.stream.is_write_cycle_active();
@@ -13626,6 +13698,43 @@ impl ChatWidget<'_> {
},
);
}
EventMsg::RequestUserInput(ev) => {
let key = self.near_time_key_current_req(event.order.as_ref());
let mut lines: Vec<String> = Vec::new();
lines.push("Model requested user input".to_string());
for question in &ev.questions {
let header = &question.header;
let id = &question.id;
let question_text = &question.question;
lines.push(format!("\n{header} ({id})\n{question_text}"));
if let Some(options) = &question.options {
lines.push("Options:".to_string());
for option in options {
let label = &option.label;
let description = &option.description;
lines.push(format!("- {label}: {description}"));
}
}
}
lines.push("\nReply in the composer to continue.".to_string());
let role = history_cell::plain_role_for_kind(PlainMessageKind::Notice);
let state =
history_cell::plain_message_state_from_paragraphs(PlainMessageKind::Notice, role, lines);
let _ = self.history_insert_plain_state_with_key(state, key, "request_user_input");
self.restore_reasoning_in_progress_if_streaming();
self.pending_request_user_input = Some(PendingRequestUserInput {
turn_id: ev.turn_id.clone(),
questions: ev.questions.clone(),
});
self.bottom_pane
.update_status_text("waiting for user input".to_string());
self.bottom_pane.set_task_running(true);
self.bottom_pane.ensure_input_focus();
self.request_redraw();
}
EventMsg::ApplyPatchApprovalRequest(ev) => {
let id2 = id.clone();
let ev2 = ev.clone();
@@ -14729,7 +14838,9 @@ impl ChatWidget<'_> {
}
// Newer protocol variants we currently ignore in the TUI
EventMsg::UserMessage(_) => {}
EventMsg::TurnAborted(_) => {}
EventMsg::TurnAborted(_) => {
self.pending_request_user_input = None;
}
EventMsg::ConversationPath(_) => {}
EventMsg::EnteredReviewMode(review_request) => {
if self.auto_resolve_enabled() {