Solo's Blog

切换主题

MultiAgent实时上下文

分析协作模式、Realtime状态和子代理完成通知如何沿不同时间线进入模型上下文。

基于rust-v0.150.0
CodexRustContextMultiAgentRealtime
4153
19.9m

MultiAgent实时上下文 ​

协作模式、Realtime 会话和子代理通知都会改变模型当前应如何行动，但它们不是同一种“多 Agent 上下文”。协作模式是 Turn 配置的持久 developer 指令；Realtime 是会话活跃状态的开始/结束转换；子代理通知则是另一个 Thread 进入最终状态后追加到父 Thread 的异步消息。三者的所有者、差分键和失效时机都不同。

本文只分析这些状态怎样进入普通模型请求，不展开 agent registry、音频编码、WebRTC 信令或工具 schema。读者应先了解 Context变更语义 的 WorldState snapshot，也可结合 TurnContext字段 理解 Turn 配置为何能在后续请求生效。读完后应能从 build_world_state_for_step 分别追到协作指令、Realtime 转换和子代理完成消息，并判断一次“没有新增消息”究竟是稳定状态、未知历史还是异步通知尚未到达。

1. 三条时间线 ​

本篇涉及的状态分布在 Session、Turn、Realtime conversation 和子 Thread 上。把它们都称为“当前模式”会掩盖真正的生命周期。

上下文所有者差分身份注入角色变化来源
协作模式Session/Turn 配置mode + modeldeveloperthread settings 更新
RealtimeRealtimeConversationactivedeveloper会话开始、关闭、错误
V1 子代理通知父 Thread history完整消息追加user子 Thread 最终状态
V2 完成消息inter-agent mailboxassistant fragment（无 marker）agent 通信子 Thread 最终状态

协作模式和 Realtime 都进入 WorldState，但语义不同：前者描述稳定工作方式，后者描述一次有方向的状态转换。子代理通知不属于 WorldState section，因为它记录的是发生过的异步事件，而不是每个 Step 都能重新计算的当前配置。

当前实现还把 MultiAgent V2 的“使用提示”拆成独立 section：MultiAgentModeState 描述有效模式， MultiAgentUsageHintState 描述 root/subagent 如何使用协作工具。提示文本参与自己的 snapshot/hash，模式本身 不变也可能因为提示迁移而重新发出；这与协作模式只按 mode + model 判断差分的规则不同。

源码位置：codex-rs/core/src/session/world_state.rs :: build_world_state_for_step； codex-rs/core/src/context/world_state/multi_agent_usage_hint.rs :: MultiAgentUsageHintState

rust
let mut multi_agent_mode = MultiAgentModeState::new(
    super::multi_agents::effective_multi_agent_mode(turn_context),
);
if let Some(usage_hint) = super::multi_agents::usage_hint_text(
    turn_context,
    &turn_context.session_source,
) {
    multi_agent_mode = multi_agent_mode.with_usage_hint(
        &MultiAgentUsageHintState::new(usage_hint),
    );
}
world_state.add_section(multi_agent_mode);

2. 模式对象 ​

CollaborationMode 不只有 Default 和 Plan 枚举值，还携带模型、推理强度和可选 developer instructions。模式更新因此可能同时改变执行模型和模型可见指令。

源码位置：codex-rs/protocol/src/config_types.rs :: CollaborationMode、Settings。

rust
#[derive(Clone, PartialEq, Eq, Hash, Debug, Serialize, Deserialize, JsonSchema, TS)]
#[serde(rename_all = "lowercase")]
pub struct CollaborationMode {
    pub mode: ModeKind,
    pub settings: Settings,
}

impl CollaborationMode {
    pub fn model(&self) -> &str {
        self.settings_ref().model.as_str()
    }

    pub fn reasoning_effort(&self) -> Option<ReasoningEffort> {
        self.settings_ref().reasoning_effort.clone()
    }

    pub fn with_updates(
        &self,
        model: Option<String>,
        effort: Option<Option<ReasoningEffort>>,
        developer_instructions: Option<Option<String>>,
    ) -> Self {
        let settings = self.settings_ref();
        let updated_settings = Settings {
            model: model.unwrap_or_else(|| settings.model.clone()),
            reasoning_effort: effort.unwrap_or_else(|| settings.reasoning_effort.clone()),
            developer_instructions: developer_instructions
                .unwrap_or_else(|| settings.developer_instructions.clone()),
        };

        CollaborationMode {
            mode: self.mode,
            settings: updated_settings,
        }
    }
}

#[derive(Clone, PartialEq, Eq, Hash, Debug, Serialize, Deserialize, JsonSchema, TS)]
pub struct Settings {
    pub model: String,
    pub reasoning_effort: Option<ReasoningEffort>,
    pub developer_instructions: Option<String>,
}

with_updates 对可选字段使用双层 Option：外层 None 表示保持原值，Some(None) 表示清除 reasoning effort 或 developer instructions。这个区别让一次局部 settings 更新不必重建完整模式对象。

ModeKind::allows_request_user_input 只允许 Plan 模式使用专用提问工具。这是模式行为的一部分，但不是 <collaboration_mode> 文本本身提供的能力；工具是否注册还要看上层运行模式和工具配置。

3. 文案选择 ​

CollaborationModeState 先按 mode 从模型 catalog 选择文案，找不到对应项时才回退 settings.developer_instructions。catalog 中显式空字符串不是“缺失”，而是要求生成一对空 marker，以清除旧模式文本。

源码位置：codex-rs/core/src/context/world_state/collaboration_mode.rs :: from_collaboration_mode。

rust
pub(crate) fn from_collaboration_mode(
    collaboration_mode: &CollaborationMode,
    catalog_messages: Option<&CollaborationModeMessages>,
) -> Self {
    let catalog_instructions =
        catalog_messages.and_then(|messages| match collaboration_mode.mode {
            ModeKind::Default => messages.default.as_ref(),
            ModeKind::Plan => messages.plan.as_ref(),
        });

    Self {
        mode: collaboration_mode.mode,
        model: collaboration_mode.settings.model.clone(),
        instructions: catalog_instructions.cloned().or_else(|| {
            collaboration_mode
                .settings
                .developer_instructions
                .clone()
                .filter(|instructions| !instructions.is_empty())
        }),
    }
}

这段 or_else 的顺序意味着只要 catalog 对当前 mode 有值，即使值为空，也不会读取 legacy developer instructions。若 catalog 对当前 mode 为 None，非空 legacy 文案才成为 fallback。

4. 模式差分 ​

协作模式 snapshot 故意只保存 mode 和 model，不保存 instructions 文本或 reasoning effort。

源码位置：codex-rs/core/src/context/world_state/collaboration_mode.rs :: CollaborationModeSnapshot、snapshot。

rust
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub(crate) enum CollaborationModeSnapshot {
    Current { mode: ModeKind, model: String },
    Legacy(ModeKind),
}

impl WorldStateSection for CollaborationModeState {
    const ID: &'static str = "collaboration_mode";
    type Snapshot = CollaborationModeSnapshot;

    fn snapshot(&self) -> Self::Snapshot {
        CollaborationModeSnapshot::Current {
            mode: self.mode,
            model: self.model.clone(),
        }
    }

    fn should_persist(&self) -> bool {
        self.instructions.is_some()
    }

因此，同一 mode、同一 model 内仅修改 developer instructions，不会触发新的 WorldState fragment；reasoning effort 变化也不会因为这个 section 本身而产生新文案。mode 或 model 改变才重新评估指令。

源码位置：codex-rs/core/src/context/world_state/collaboration_mode.rs :: render_diff。

rust
fn render_diff(
    &self,
    previous: PreviousSectionState<'_, Self::Snapshot>,
) -> Option<Box<dyn ContextualUserFragment>> {
    if matches!(
        previous,
        PreviousSectionState::Known(CollaborationModeSnapshot::Current { mode, model })
            if *mode == self.mode && model == &self.model
    ) || matches!(previous, PreviousSectionState::Unknown)
        || (self.instructions.is_none() && matches!(previous, PreviousSectionState::Absent))
    {
        return None;
    }

    Some(Box::new(CollaborationModeInstructions {
        instructions: self.instructions.clone().unwrap_or_default(),
    }))
}

如果上一状态已知且 mode/model 相同，返回 None；previous 为 Unknown 时也不盲目补发，由 retained-history matcher 负责判断历史中是否已有 fragment。若从有文案切换到没有文案，且 mode 或 model 已变化，则发送空 <collaboration_mode></collaboration_mode>，让当前状态显式覆盖旧说明。

5. Step组装 ​

每个 Step 构造 WorldState 时读取当前 Turn 的 collaboration mode 和当前模型 catalog。配置开关关闭时，这个 section 根本不加入 WorldState。

源码位置：codex-rs/core/src/session/world_state.rs :: build_world_state_for_step。

rust
if turn_context.config.include_collaboration_mode_instructions {
    world_state.add_section(CollaborationModeState::from_collaboration_mode(
        &turn_context.collaboration_mode(),
        turn_context
            .model_info
            .model_messages
            .as_ref()
            .and_then(|messages| messages.collaboration_modes.as_ref()),
    ));
}

这解释了 model 为什么属于 snapshot：切换模型后，即使 mode 仍为 Default，新模型 catalog 可能拥有不同的 Default 文案，必须重新注入。相反，同一模型 catalog 内容在后台发生变化，但 mode/model 未变时，这个 section 不会自动感知文本变化。

6. Realtime转换 ​

Realtime WorldState 只保存 active: bool。start/end instructions 是渲染当前转换所需的载荷，不进入 snapshot。

源码位置：codex-rs/core/src/context/world_state/realtime.rs :: RealtimeState、RealtimeSnapshot。

rust
#[derive(Clone, Debug)]
pub(crate) struct RealtimeState {
    snapshot: RealtimeSnapshot,
    start_instructions: Option<String>,
    end_instructions: Option<String>,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub(crate) struct RealtimeSnapshot {
    active: bool,
}

impl RealtimeState {
    pub(crate) fn new(
        active: bool,
        start_instructions: Option<&str>,
        end_instructions: Option<&str>,
    ) -> Self {
        Self {
            snapshot: RealtimeSnapshot { active },
            start_instructions: start_instructions.map(str::to_string),
            end_instructions: end_instructions.map(str::to_string),
        }
    }

false→true 发送开始指令，true→false 发送结束指令，相同状态不发送。active 状态下修改 custom start instructions 不会重新注入，因为模型已经处于 Realtime 模式，当前 section 只表达边沿转换。

源码位置：codex-rs/core/src/context/world_state/realtime.rs :: render_transition、render_diff。

rust
fn render_transition(&self, previous_active: bool) -> Option<Box<dyn ContextualUserFragment>> {
    match (previous_active, self.snapshot.active) {
        (false, true) => Some(self.render_start()),
        (true, false) => Some(match self.end_instructions.as_deref() {
            Some(instructions) => {
                Box::new(RealtimeEndInstructions::with_instructions(instructions))
            }
            None => Box::new(RealtimeEndInstructions::new()),
        }),
        (false, false) | (true, true) => None,
    }
}

fn render_diff(
    &self,
    previous: PreviousSectionState<'_, Self::Snapshot>,
) -> Option<Box<dyn ContextualUserFragment>> {
    match previous {
        PreviousSectionState::Known(previous) if previous == &self.snapshot => None,
        PreviousSectionState::Known(previous) => self.render_transition(previous.active),
        PreviousSectionState::Absent | PreviousSectionState::Unknown
            if self.snapshot.active =>
        {
            Some(self.render_start())
        }
        PreviousSectionState::Absent | PreviousSectionState::Unknown => None,
    }
}

与协作模式不同，Realtime 在 previous 为 Unknown 且当前 active 时会补发开始指令。原因是 active 会话必须立刻改变模型对语音 transcript 的解释；宁可重新声明开始，也不能让模型继续按普通输入处理。

7. 会话所有权 ​

RealtimeConversation 用 Mutex<Option<ConversationState>> 持有当前连接；每次 start 创建独立的 Arc<AtomicBool>。后续注册 fanout task 或结束会话时使用 Arc::ptr_eq，防止旧会话的异步清理误操作新会话。

源码位置：codex-rs/core/src/realtime_conversation.rs :: register_fanout_task、finish_if_active。

rust
pub(crate) async fn register_fanout_task(
    &self,
    realtime_active: &Arc<AtomicBool>,
    fanout_task: JoinHandle<()>,
) {
    let mut fanout_task = Some(fanout_task);
    {
        let mut guard = self.state.lock().await;
        if let Some(state) = guard.as_mut()
            && Arc::ptr_eq(&state.realtime_active, realtime_active)
        {
            state.fanout_task = fanout_task.take();
        }
    }

    if let Some(fanout_task) = fanout_task {
        fanout_task.abort();
        let _ = fanout_task.await;
    }
}

pub(crate) async fn finish_if_active(&self, realtime_active: &Arc<AtomicBool>) {
    let state = {
        let mut guard = self.state.lock().await;
        match guard.as_ref() {
            Some(state) if Arc::ptr_eq(&state.realtime_active, realtime_active) => guard.take(),
            _ => None,
        }
    };

    if let Some(state) = state {
        stop_conversation_state(state, RealtimeFanoutTaskStop::Detach).await;
    }
}

如果注册 fanout 时 active token 已不匹配，传入的 task 会被立即 abort；如果旧 fanout 尝试结束会话但 token 不匹配，什么也不做。这是异步重连场景中的所有权屏障。

真正停止时先把 active 设为 false，再 cancel stop token，等待 input task，最后按调用场景 await 或 detach fanout。

源码位置：codex-rs/core/src/realtime_conversation.rs :: stop_conversation_state。

rust
async fn stop_conversation_state(
    mut state: ConversationState,
    fanout_task_stop: RealtimeFanoutTaskStop,
) {
    state.realtime_active.store(false, Ordering::Relaxed);
    state.stop_token.cancel();
    let _ = state.input_task.await;

    if let Some(fanout_task) = state.fanout_task.take() {
        match fanout_task_stop {
            RealtimeFanoutTaskStop::Await => {
                let _ = fanout_task.await;
            }
            RealtimeFanoutTaskStop::Detach => {}
        }
    }
}

active 的原子变化随后会被新的 Step 捕获，WorldState 才在普通模型请求中生成结束指令。关闭 transport 本身和模型看到 <realtime_conversation> 结束说明不是同一个同步动作。

8. Handoff路由 ​

Realtime fanout 收到 HandoffRequested 时，把语音侧请求转换为带 <realtime_delegation> 标记的普通文本，再调用 route_realtime_text_input 进入 Codex Turn。事件流关闭后，还会尝试路由 transcript tail。

源码位置：codex-rs/core/src/realtime_conversation.rs :: fanout task。

rust
while let Ok(event) = events_rx.recv().await {
    if let RealtimeEvent::Error(_) = &event {
        end = RealtimeConversationEnd::Error;
    }
    let maybe_routed_text = match &event {
        RealtimeEvent::HandoffRequested(handoff) => {
            realtime_delegation_from_handoff(handoff)
        }
        _ => None,
    };
    if let Some(text) = maybe_routed_text {
        let sess_for_routed_text = Arc::clone(&sess_clone);
        sess_for_routed_text.route_realtime_text_input(text).await;
    }
    sess_clone
        .send_event_raw(ev(EventMsg::RealtimeConversationRealtime(
            RealtimeConversationRealtimeEvent {
                payload: event.clone(),
            },
        )))
        .await;
}
if let Ok(text) = transcript_tail_rx.recv().await {
    sess_clone.route_realtime_text_input(text).await;
}

handoff 输入和 transcript delta 在包装时进行 XML 转义，避免语音文本中的 <、>、& 破坏上下文边界。Realtime 指令告诉模型输入可能缺少标点或存在识别错误；handoff wrapper 则保存这段文本来自 Realtime 委派，而不是用户键盘输入。

9. 完成监听 ​

子代理完成通知由 detached watcher 负责。它只为 SubAgentSource::ThreadSpawn 启动，订阅子 Thread 状态直到 is_final；订阅关闭时回退主动读取一次状态。

源码位置：codex-rs/core/src/agent/control.rs :: maybe_start_completion_watcher。

rust
fn maybe_start_completion_watcher(
    &self,
    child_thread_id: ThreadId,
    session_source: Option<SessionSource>,
    child_reference: String,
    child_agent_path: Option<AgentPath>,
) {
    let Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
        parent_thread_id, ..
    })) = session_source
    else {
        return;
    };
    let control = self.clone();
    tokio::spawn(async move {
        let status = match control.subscribe_status(child_thread_id).await {
            Ok(mut status_rx) => {
                let mut status = status_rx.borrow().clone();
                while !is_final(&status) {
                    if status_rx.changed().await.is_err() {
                        status = control.get_status(child_thread_id).await;
                        break;
                    }
                    status = status_rx.borrow().clone();
                }
                status
            }
            Err(_) => control.get_status(child_thread_id).await,
        };
        if !is_final(&status) {
            return;
        }

watcher 不轮询模型请求，也不触发父 Thread 新 Turn。它只在最终状态到达后把消息排入父侧；父模型何时消费，取决于父 Thread 下一次请求或当前调度路径。

10. V1与V2 ​

完成通知随后按 MultiAgent 版本分流。V2 有 agent_path 时构造 InterAgentCompletionMessage，通过 InterAgentCommunication 发给直接父路径，且 trigger_turn=false；V1 则生成 user-role <subagent_notification> 并直接追加父 Thread history。

源码位置：codex-rs/core/src/agent/control.rs :: completion watcher 分流。

rust
let child_thread = state.get_thread(child_thread_id).await.ok();
let child_uses_multi_agent_v2 = match child_thread.as_ref() {
    Some(child_thread) => {
        child_thread.multi_agent_version() == Some(MultiAgentVersion::V2)
    }
    None => true,
};
if child_agent_path.is_some() && child_uses_multi_agent_v2 {
    let Some(child_agent_path) = child_agent_path.clone() else {
        return;
    };
    let Some(parent_agent_path) = child_agent_path
        .as_str()
        .rsplit_once('/')
        .and_then(|(parent, _)| AgentPath::try_from(parent).ok())
    else {
        return;
    };
    let Some(message) = format_inter_agent_completion_message(
        parent_agent_path.clone(),
        child_agent_path.clone(),
        &status,
    ) else {
        return;
    };
    let communication = InterAgentCommunication::new(
        child_agent_path,
        parent_agent_path,
        Vec::new(),
        message,
        /*trigger_turn*/ false,
    );
    let context =
        AgentCommunicationContext::new(AgentCommunicationKind::Result, child_thread_id);
    let _ = control
        .send_inter_agent_communication(
            parent_thread_id,
            communication,
            context,
            /*parent_turn_id*/ None,
        )
        .await;
    return;
}
let message = format_subagent_notification_message(child_reference.as_str(), &status);
let Ok(parent_thread) = state.get_thread(parent_thread_id).await else {
    return;
};
parent_thread
    .inject_user_message_without_turn(message)
    .await;

V1 fragment 只含 agent reference 和完整 AgentStatus JSON：

源码位置：codex-rs/core/src/context/subagent_notification.rs :: SubagentNotification。

rust
impl ContextualUserFragment for SubagentNotification {
    fn role(&self) -> &'static str {
        "user"
    }

    fn markers(&self) -> (&'static str, &'static str) {
        Self::type_markers()
    }

    fn type_markers() -> (&'static str, &'static str) {
        ("<subagent_notification>", "</subagent_notification>")
    }

    fn body(&self) -> String {
        format!(
            "\n{}\n",
            serde_json::json!({
                "agent_path": &self.agent_reference,
                "status": &self.status,
            })
        )
    }
}

V2 completion formatter 只为 Completed、Errored、Shutdown 和 NotFound 生成正文；PendingInit、Running、Interrupted 返回 None。这个结果作为无 marker 的 assistant fragment 通过 mailbox 传递；错误正文最多保留约 900 token，并追加如何重新分配任务的行动建议。

11. 故障边界 ​

这三条时间线的非正常路径也不同：

现象实际边界首查位置
改了同模式文案却未追加snapshot 只比较 mode/modelcollaboration_mode.rs
Realtime 文案改变却未重发snapshot 只比较 activerealtime.rs
旧 Realtime task 立即 abortactive token 已不属于当前会话register_fanout_task
transport 错误后仍无结束说明关闭事件与下一次 Step 注入异步分离fanout、WorldState
子代理 Running 没有完成消息watcher 只处理最终状态is_final、formatter
V2 完成未出现在根 history发给直接父 agent mailboxcompletion watcher
子 Thread 已不存在状态回退为 NotFound，V1 仍可通知get_status fallback

取消和清理在本主题中不是统一动作。Realtime 有 stop token、task await/detach 和 active 原子位；子代理 watcher 自身是 detached task，父 Thread 不存在或 control 无法 upgrade 时直接结束；协作模式没有独立清理任务，只由后续 settings 和 WorldState 差分覆盖。

12. 时间线测试 ​

协作模式测试构造 Default 和 Plan 两轮请求，模型 catalog 同时提供两种文案。第一轮断言只出现 Default catalog 文案且 legacy 文案不存在；切换 Plan 后断言历史保留 Default，同时新增一份 Plan 文案。这证明 catalog 优先级和 mode 转换，不证明模型会按 Plan 行为执行。

源码位置：codex-rs/core/tests/suite/collaboration_instructions.rs :: catalog_collaboration_messages_track_mode_changes。

rust
let first_dev_texts = developer_texts(&req1.single_request().input());
assert_eq!(
    count_messages_containing(&first_dev_texts, &collab_xml(default_text)),
    1
);
assert_eq!(
    count_messages_containing(&first_dev_texts, "legacy default instructions"),
    0
);

core_test_support::submit_thread_settings(
    &test.codex,
    codex_protocol::protocol::ThreadSettingsOverrides {
        collaboration_mode: Some(collab_mode_for_model(
            ModeKind::Plan,
            model_slug,
            Some("legacy plan instructions"),
        )),
        ..Default::default()
    },
)
.await?;
test.submit_text_turn("plan turn").await?;

let second_dev_texts = developer_texts(&req2.single_request().input());
assert_eq!(
    count_messages_containing(&second_dev_texts, &collab_xml(default_text)),
    1
);
assert_eq!(
    count_messages_containing(&second_dev_texts, &collab_xml(plan_text)),
    1
);

Realtime snapshot 测试覆盖 Absent、Unknown、inactive、active 和 custom instructions。关键断言由 snapshot 固定：active→active 即使 custom 文案改变仍无输出，active→inactive 才生成结束 fragment。它证明转换语义，不证明真实音频连接已经建立或关闭。

源码位置：codex-rs/core/src/context/world_state/realtime_tests.rs :: snapshots。

rust
insta::assert_snapshot!(render_section_cases(&[
    (Absent, Absent),
    (Absent, Known(&inactive)),
    (Absent, Known(&active)),
    (Known(&inactive), Known(&active)),
    (Known(&inactive), Known(&custom_active)),
    (Known(&active), Known(&active)),
    (Known(&custom_active), Known(&changed_custom_active)),
    (Known(&active), Known(&inactive)),
    (Unknown, Known(&active)),
    (Unknown, Known(&inactive)),
]));

完成 watcher 测试创建一个不存在的 child ID，等待父 Thread 收到通知，再断言 history 含该 ID。这证明 subscribe 失败或 child 缺失时仍能走 NotFound 通知，不证明任意父子树路径都会被路由到根节点。

源码位置：codex-rs/core/src/agent/control_tests.rs :: completion_watcher_notifies_parent_when_child_is_missing。

rust
let child_thread_id = ThreadId::new();

harness.control.maybe_start_completion_watcher(
    child_thread_id,
    Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
        parent_thread_id,
        depth: 1,
        agent_path: None,
        agent_nickname: None,
        agent_role: Some("explorer".to_string()),
    })),
    child_thread_id.to_string(),
    /*child_agent_path*/ None,
);

assert_eq!(wait_for_subagent_notification(&parent_thread).await, true);

let history_items = parent_thread
    .session
    .clone_history()
    .await
    .raw_items()
    .to_vec();
assert_eq!(
    history_contains_text(
        &history_items,
        &format!("\"agent_path\":\"{child_thread_id}\"")
    ),
    true
);

要验证自己是否真正掌握本篇，可以分别预测三种变化的下一条模型上下文：同 model 的 Default 文案被替换、Realtime active 从 true 变 false、V2 孙 agent 完成。正确答案分别是协作 section 不追加、生成 Realtime 结束 fragment、只向直接父 agent 发送 result message。

bash
rg -n "CollaborationModeState|RealtimeState|maybe_start_completion_watcher|SubagentNotification" \
  codex-rs/core/src \
  codex-rs/protocol/src
cargo test -p codex-core context::world_state::collaboration_mode --lib
cargo test -p codex-core context::world_state::realtime --lib
cargo test -p codex-core completion_watcher_notifies_parent_when_child_is_missing --lib
cargo test -p codex-core realtime_conversation::tests --lib