Plan与RequestInput协议

本文承接Permissions协议，也会用到UserInput类型体系。Plan 和 request_user_input 都会影响用户界面，但它们的所有权完全不同：Plan 是模型文本流中的标记，被解析为计划 item 和增量事件；request_user_input 是 Core 工具主动创建的 pending waiter，必须等客户端回答后才能继续。

阅读时不要把“计划更新”和“用户回答”都看成普通 EventMsg。Plan 需要流式解析和完成时重建，问题请求需要 call/turn ID、oneshot 生命周期、模式限制和客户端 JSON-RPC 回传。

1. 两类交互工具

源码位置：codex-rs/protocol/src/plan_tool.rs :: StepStatus、PlanItemArg、UpdatePlanArgs

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
pub enum StepStatus {
    Pending,
    InProgress,
    Completed,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, TS)]
#[serde(deny_unknown_fields)]
pub struct PlanItemArg {
    pub step: String,
    pub status: StepStatus,
}

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, TS)]
#[serde(deny_unknown_fields)]
pub struct UpdatePlanArgs {
    #[serde(default)]
    pub explanation: Option<String>,
    pub plan: Vec<PlanItemArg>,
}

这里的 update_plan 是 TODO/checklist 工具，不是 plan mode 的模型提案。它产生 EventMsg::PlanUpdate，由客户端显示结构化步骤；模型输出中的 <proposed_plan> 则走另一条流式 parser。

源码位置：codex-rs/protocol/src/request_user_input.rs :: RequestUserInputQuestion、RequestUserInputArgs、RequestUserInputResponse、RequestUserInputEvent

#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
pub struct RequestUserInputQuestion {
    pub id: String,
    pub header: String,
    pub question: String,
    #[serde(rename = "isOther", default)]
    pub is_other: bool,
    #[serde(rename = "isSecret", default)]
    pub is_secret: bool,
    pub options: Option<Vec<RequestUserInputQuestionOption>>,
}

#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
pub struct RequestUserInputArgs {
    pub questions: Vec<RequestUserInputQuestion>,
    #[serde(rename = "isBlocking")]
    pub is_blocking: bool,
    #[serde(rename = "autoResolutionMs", skip_serializing_if = "Option::is_none")]
    pub auto_resolution_ms: Option<u64>,
}

#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
pub struct RequestUserInputResponse {
    pub answers: HashMap<String, RequestUserInputAnswer>,
}

问题协议按 question id 建立答案映射，is_secret 是展示语义，is_blocking 是工具/turn 行为语义，不能把二者混为一谈。

2. Plan标签解析

源码位置：codex-rs/utils/stream-parser/src/proposed_plan.rs :: ProposedPlanSegment、ProposedPlanParser

const OPEN_TAG: &str = "<proposed_plan>";
const CLOSE_TAG: &str = "</proposed_plan>";

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProposedPlanSegment {
    Normal(String),
    ProposedPlanStart,
    ProposedPlanDelta(String),
    ProposedPlanEnd,
}

#[derive(Debug)]
pub struct ProposedPlanParser {
    parser: TaggedLineParser<PlanTag>,
}

impl ProposedPlanParser {
    pub fn new() -> Self {
        Self {
            parser: TaggedLineParser::new(vec![TagSpec {
                open: OPEN_TAG,
                close: CLOSE_TAG,
                tag: PlanTag::ProposedPlan,
            }]),
        }
    }
}

parser 跨 chunk 保持状态，因此开标签被拆在两个网络片段中仍能识别。它同时返回 visible_text 和有序 segments：普通文本保留给 assistant message，计划文本转成 Plan segment。

源码位置：codex-rs/utils/stream-parser/src/proposed_plan.rs :: map_segments、extract_proposed_plan_text

fn map_segments(segments: Vec<TaggedLineSegment<PlanTag>>) -> StreamTextChunk<ProposedPlanSegment> {
    let mut out = StreamTextChunk::default();
    for segment in segments {
        let mapped = match segment {
            TaggedLineSegment::Normal(text) => ProposedPlanSegment::Normal(text),
            TaggedLineSegment::TagStart(PlanTag::ProposedPlan) => ProposedPlanSegment::ProposedPlanStart,
            TaggedLineSegment::TagDelta(PlanTag::ProposedPlan, text) => {
                ProposedPlanSegment::ProposedPlanDelta(text)
            }
            TaggedLineSegment::TagEnd(PlanTag::ProposedPlan) => ProposedPlanSegment::ProposedPlanEnd,
        };
        if let ProposedPlanSegment::Normal(text) = &mapped {
            out.visible_text.push_str(text);
        }
        out.extracted.push(mapped);
    }
    out
}

pub fn extract_proposed_plan_text(text: &str) -> Option<String> {
    let mut parser = ProposedPlanParser::new();
    let mut plan_text = String::new();
    let mut saw_plan_block = false;
    for segment in parser
        .push_str(text)
        .extracted
        .into_iter()
        .chain(parser.finish().extracted)
    {
        match segment {
            ProposedPlanSegment::ProposedPlanStart => {
                saw_plan_block = true;
                plan_text.clear();
            }
            ProposedPlanSegment::ProposedPlanDelta(delta) => plan_text.push_str(&delta),
            ProposedPlanSegment::ProposedPlanEnd | ProposedPlanSegment::Normal(_) => {}
        }
    }
    saw_plan_block.then_some(plan_text)
}

未闭合标签在 finish() 时被视为结束，这保证 response stream 结束时仍能得到可解释的计划文本；同时，visible_text 不包含计划块，因此不会再生成一份重复的 assistant 文本。

3. Plan item状态

源码位置：codex-rs/core/src/session/turn.rs :: ProposedPlanItemState

struct ProposedPlanItemState {
    item_id: String,
    started: bool,
    completed: bool,
}

impl ProposedPlanItemState {
    fn new(turn_id: &str) -> Self {
        Self {
            item_id: format!("{turn_id}-plan"),
            started: false,
            completed: false,
        }
    }

    async fn start(&mut self, sess: &Session, turn_context: &TurnContext) {
        if self.started || self.completed {
            return;
        }
        self.started = true;
        let item = TurnItem::Plan(PlanItem {
            id: self.item_id.clone(),
            text: String::new(),
        });
        sess.emit_turn_item_started(turn_context, &item).await;
    }
}

Plan item ID 从 turn sub id 派生，并由 started/completed 防止重复生命周期事件。这个状态只在单次 response streaming 期间存在，不是 session 全局持久状态。

源码位置：codex-rs/core/src/session/turn.rs :: push_delta、complete_with_text

async fn push_delta(&mut self, sess: &Session, turn_context: &TurnContext, delta: &str) {
    if self.completed || delta.is_empty() {
        return;
    }
    let event = PlanDeltaEvent {
        thread_id: sess.thread_id.to_string(),
        turn_id: turn_context.sub_id.clone(),
        item_id: self.item_id.clone(),
        delta: delta.to_string(),
    };
    sess.send_event(turn_context, EventMsg::PlanDelta(event)).await;
}

async fn complete_with_text(
    &mut self,
    sess: &Session,
    turn_context: &TurnContext,
    text: String,
) {
    if self.completed || !self.started {
        return;
    }
    self.completed = true;
    let item = TurnItem::Plan(PlanItem {
        id: self.item_id.clone(),
        text,
    });
    sess.emit_turn_item_completed(turn_context, item).await;
}

PlanDelta 是流式观察信号，完成 item 的文本则在 assistant item 完成后重新提取。客户端不能简单假设“把所有 delta 拼起来”一定等于最终 item 内容。

4. Plan mode文本

源码位置：codex-rs/core/src/session/turn.rs :: handle_plan_segments

async fn handle_plan_segments(
    sess: &Session,
    turn_context: &TurnContext,
    state: &mut PlanModeStreamState,
    item_id: &str,
    segments: Vec<ProposedPlanSegment>,
) {
    for segment in segments {
        match segment {
            ProposedPlanSegment::Normal(delta) => {
                if delta.is_empty() {
                    continue;
                }
                let has_non_whitespace = delta.chars().any(|ch| !ch.is_whitespace());
                if !has_non_whitespace && !state.started_agent_message_items.contains(item_id) {
                    state
                        .leading_whitespace_by_item
                        .entry(item_id.to_string())
                        .or_default()
                        .push_str(&delta);
                    continue;
                }
                maybe_emit_pending_agent_message_start(sess, turn_context, state, item_id).await;
                let event = AgentMessageContentDeltaEvent {
                    thread_id: sess.thread_id.to_string(),
                    turn_id: turn_context.sub_id.clone(),
                    item_id: item_id.to_string(),
                    delta,
                };
                sess.send_event(turn_context, EventMsg::AgentMessageContentDelta(event))
                    .await;
            }
            ProposedPlanSegment::ProposedPlanStart => {
                if !state.plan_item_state.completed {
                    state.plan_item_state.start(sess, turn_context).await;
                }
            }
            ProposedPlanSegment::ProposedPlanDelta(delta) => {
                if !state.plan_item_state.completed {
                    if !state.plan_item_state.started {
                        state.plan_item_state.start(sess, turn_context).await;
                    }
                    state.plan_item_state.push_delta(sess, turn_context, &delta).await;
                }
            }
            ProposedPlanSegment::ProposedPlanEnd => {}
        }
    }
}

普通文本在出现非空白内容前会被缓冲，避免计划-only response 先显示一个空 assistant item。计划 segment 则独立发送 PlanDelta，二者共享 turn/item 关联但不共享展示语义。

5. Request工具规范

源码位置：codex-rs/core/src/tools/handlers/request_user_input_spec.rs :: create_request_user_input_tool、normalize_request_user_input_tool_args

pub(crate) fn normalize_request_user_input_tool_args(
    mut args: RequestUserInputToolArgs,
) -> Result<RequestUserInputToolArgs, String> {
    let missing_options = args
        .questions
        .iter()
        .any(|question| question.options.as_ref().is_none_or(Vec::is_empty));
    if missing_options {
        return Err("request_user_input requires non-empty options for every question".to_string());
    }

    for question in &mut args.questions {
        question.is_other = true;
    }

    Ok(args)
}

Core 强制每个问题有非空 options，并统一打开 free-form Other 选项。这个规范化发生在工具 handler 中，早于 session pending waiter。

源码位置：codex-rs/core/src/tools/handlers/request_user_input.rs :: RequestUserInputHandler::handle_call

if turn.session_source.is_non_root_agent() {
    return Err(FunctionCallError::RespondToModel(
        "request_user_input can only be used by the root thread".to_string(),
    ));
}

let mode = turn.collaboration_mode().mode;
if let Some(message) = request_user_input_unavailable_message(mode, &self.available_modes) {
    return Err(FunctionCallError::RespondToModel(message));
}

let args: RequestUserInputToolArgs = parse_arguments(&arguments)?;
let args = normalize_request_user_input_tool_args(args)
    .map_err(FunctionCallError::RespondToModel)?;
let args = RequestUserInputArgs {
    questions: args.questions,
    is_blocking: mode == ModeKind::Plan,
    auto_resolution_ms: None,
};

request tool 只允许 root thread；是否可用由当前 collaboration mode 和注册时的 available_modes 决定。当前实现把 Plan mode 请求标记为 blocking，其他允许模式标记为 non-blocking。

6. Session注册等待

源码位置：codex-rs/core/src/session/mod.rs :: request_user_input

let _elicitation = self.services.elicitations.register();
let sub_id = turn_context.sub_id.clone();
let (tx_response, rx_response) = oneshot::channel();
let prev_entry = {
    let mut active = self.active_turn.lock().await;
    match active.as_mut() {
        Some(at) => {
            let mut ts = at.turn_state.lock().await;
            ts.insert_pending_user_input(sub_id, tx_response)
        }
        None => None,
    }
};
let event = EventMsg::RequestUserInput(RequestUserInputEvent {
    call_id,
    turn_id: turn_context.sub_id.clone(),
    questions: args.questions,
    is_blocking: args.is_blocking,
    auto_resolution_ms: args.auto_resolution_ms,
});
turn_context.turn_metadata_state.mark_user_input_requested_during_turn();
self.send_event(turn_context, event).await;
rx_response.await.ok()

pending key 使用 sub_id，不是 question id。插入旧 entry 时会覆盖前一个等待者并记录 warning；没有 active turn 则不能注册。事件先发送，future 后等待，保证客户端先看到请求再处理回答。

7. 回答回传

源码位置：codex-rs/core/src/session/handlers.rs :: request_user_input_response；codex-rs/core/src/session/mod.rs :: notify_user_input_response

pub async fn request_user_input_response(
    sess: &Arc<Session>,
    id: String,
    response: RequestUserInputResponse,
) {
    sess.notify_user_input_response(&id, response).await;
}

handler 只负责把协议 response 交给 Session。Session 按 id 删除 pending sender，再把答案发送给工具 future；未知 id 只记录 warning，不会创建新的 waiter。

源码位置：codex-rs/app-server/src/bespoke_event_handling.rs :: EventMsg::RequestUserInput 分支

let params = ToolRequestUserInputParams {
    thread_id: conversation_id.to_string(),
    turn_id: request.turn_id,
    item_id: request.call_id,
    questions,
    is_blocking: request.is_blocking,
    auto_resolution_ms: request.auto_resolution_ms,
};
let (pending_request_id, rx) = outgoing
    .send_request(ServerRequestPayload::ToolRequestUserInput(params))
    .await;

App Server 把 Core event 投影为带 thread/turn/item 关联的 JSON-RPC request，并保存自己的 request id。客户端响应回来后，on_request_user_input_response 将答案转换成 Core Op::UserInputAnswer。

8. App Server计划

源码位置：codex-rs/app-server-protocol/src/protocol/event_mapping.rs :: EventMsg::PlanDelta

EventMsg::PlanDelta(event) => ServerNotification::PlanDelta(PlanDeltaNotification {
    thread_id,
    turn_id,
    item_id: event.item_id,
    delta: event.delta,
}),

PlanDeltaNotification 是单独的 item progress notification。update_plan 工具产生的结构化 checklist 则通过 TurnPlanUpdatedNotification，两者都含 plan 字样，但一个来自模型 proposed-plan stream，一个来自控制工具。

源码位置：codex-rs/app-server/src/bespoke_event_handling.rs :: handle_turn_plan_update

async fn handle_turn_plan_update(
    conversation_id: ThreadId,
    event_turn_id: &str,
    plan_update_event: UpdatePlanArgs,
    outgoing: &ThreadScopedOutgoingMessageSender,
) {
    let notification = TurnPlanUpdatedNotification {
        thread_id: conversation_id.to_string(),
        turn_id: event_turn_id.to_string(),
        explanation: plan_update_event.explanation,
        plan: plan_update_event
            .plan
            .into_iter()
            .map(TurnPlanStep::from)
            .collect(),
    };
    outgoing
        .send_server_notification(ServerNotification::TurnPlanUpdated(notification))
        .await;
}

因此客户端至少要区分 item/plan/delta 和 turn/plan/updated：前者是文本流增量，后者是 checklist 全量更新。

9. 失败与取消

计划流可能在标签未闭合、模型 stream 中断或 Turn abort 时结束；parser 的 finish() 会封闭尚未结束的 plan block，但 Core 只有在 assistant item 完成时才提交最终 Plan item。问题请求则可能因非 root thread、不可用 mode、缺少 options、错误的 pending id、Turn 关闭或客户端断开而失败或取消。它们都应释放临时状态，不能把未回答的问题当成空答案，也不能把未完成的计划当成正常完成。

这些失败路径说明了两条协议的边界：Plan 的异常输入通常退化为可解释的文本/生命周期状态，RequestUserInput 的异常输入则必须让等待 future 结束。下面的测试只覆盖部分本地边界，不证明所有远端断开时序。

10. 测试与边界

源码位置：codex-rs/utils/stream-parser/src/proposed_plan.rs :: streams_proposed_plan_segments_and_visible_text、closes_unterminated_plan_block_on_finish；codex-rs/protocol/src/request_user_input_tests.rs :: request_user_input_event_defaults_legacy_missing_is_blocking_to_true

源码位置：codex-rs/core/src/tools/handlers/request_user_input_tests.rs :: multi_agent_v2_request_user_input_rejects_subagent_threads、request_user_input_sets_non_blocking_outside_plan_mode、request_user_input_sets_blocking_from_turn_mode

源码位置：codex-rs/app-server-protocol/src/protocol/v2/tests.rs :: tool_request_user_input_params_default_legacy_missing_is_blocking_to_true

cd codex-rs
cargo test -p codex-utils-stream-parser streams_proposed_plan_segments_and_visible_text -- --nocapture --test-threads=1
cargo test -p codex-utils-stream-parser closes_unterminated_plan_block_on_finish -- --nocapture --test-threads=1
cargo test -p codex-protocol request_user_input_event_defaults_legacy_missing_is_blocking_to_true -- --nocapture --test-threads=1
cargo test -p codex-core multi_agent_v2_request_user_input_rejects_subagent_threads -- --nocapture --test-threads=1
cargo test -p codex-core request_user_input_sets_non_blocking_outside_plan_mode -- --nocapture --test-threads=1
cargo test -p codex-core request_user_input_sets_blocking_from_turn_mode -- --nocapture --test-threads=1
cargo test -p codex-app-server-protocol tool_request_user_input_params_default_legacy_missing_is_blocking_to_true -- --nocapture --test-threads=1

这些测试说明跨 chunk 计划解析、legacy blocking 默认、root-only 限制、模式到 blocking 的映射和 App Server 参数兼容；不能证明客户端一定及时回答、工具业务一定接受答案，或计划 delta 与最终 item 文本永远逐字相等。

11. 源码定位练习

遇到“计划被显示成普通消息”，先查 ProposedPlanParser 是否识别标签，再查 handle_plan_segments 是否延迟 assistant message start；遇到 checklist 没更新，检查是否走了 PlanHandler 的 EventMsg::PlanUpdate 路径。

遇到“问题已显示但工具不继续”，按 call_id、turn_id、pending sub_id 和 App Server request id 逐层核对；遇到子 agent 无法提问，先确认 handler 的 root-thread 限制和当前 mode 的 available modes。

Plan 是模型输出的结构化可见投影，RequestUserInput 是 Core 挂起的控制协议。只有同时理解 parser、生命周期状态、pending waiter 和客户端回传，才能准确定位“计划丢失”和“答案没有生效”这两类看似相近的问题。