mirror of
https://github.com/openai/codex.git
synced 2026-10-20 17:38:45 +00:00
jif
960e878df4
## Why Unread queue-only messages prevented idle agents from unloading, and sending a message to an evicted agent reloaded its session. Pending mail should not require an idle recipient to occupy a loaded thread slot. ## What changed - Retain queue-only mail in an in-memory runtime mailbox when a session unloads. Queue new messages for locally evicted agents without reloading them or starting a turn. - Add `AgentControl::take_mailbox` and `AgentControl::watch_mailbox` so sessions can consume retained mail and react to host mailbox notifications. - Preserve message order across eviction, reload, and subsequent follow-ups. Keep agents with outstanding durable sleep or turn-triggering mail loaded, and discard retained mail when an agent is removed or its tree shuts down. ## Testing Add integration coverage for retaining mail across eviction, sending while unloaded, and ordering messages around a follow-up after reload. Add a host controller test verifying that mailbox notifications wake `wait_agent` and that the message is delivered exactly once. GitOrigin-RevId: 4b55647f9ee9d1aaf3753f0688863bda81617327
82 lines
2.2 KiB
Rust
82 lines
2.2 KiB
Rust
use codex_protocol::ThreadId;
|
|
use codex_protocol::protocol::InterAgentCommunication;
|
|
|
|
pub(crate) static PENDING_MAILBOX_MESSAGES: codex_diagnostics::Gauge =
|
|
codex_diagnostics::Gauge::new("core.mailbox.pending");
|
|
|
|
const AGENT_COMMUNICATION_TARGET: &str = "codex_otel.agent_communication";
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
pub(crate) enum AgentCommunicationKind {
|
|
Spawn,
|
|
Message,
|
|
Followup,
|
|
Result,
|
|
}
|
|
|
|
impl AgentCommunicationKind {
|
|
fn as_str(self) -> &'static str {
|
|
match self {
|
|
Self::Spawn => "spawn",
|
|
Self::Message => "message",
|
|
Self::Followup => "followup",
|
|
Self::Result => "result",
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub(crate) struct AgentCommunicationContext {
|
|
kind: AgentCommunicationKind,
|
|
sender_thread_id: ThreadId,
|
|
}
|
|
|
|
impl AgentCommunicationContext {
|
|
pub(crate) fn new(kind: AgentCommunicationKind, sender_thread_id: ThreadId) -> Self {
|
|
Self {
|
|
kind,
|
|
sender_thread_id,
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(crate) fn logging_enabled() -> bool {
|
|
tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::INFO)
|
|
}
|
|
|
|
pub(crate) fn emit_agent_communication_send(
|
|
communication_id: &str,
|
|
context: &AgentCommunicationContext,
|
|
communication: &InterAgentCommunication,
|
|
receiver_thread_id: ThreadId,
|
|
) {
|
|
tracing::info!(
|
|
target: AGENT_COMMUNICATION_TARGET,
|
|
{
|
|
event.name = "codex.agent_communication",
|
|
communication_id,
|
|
kind = context.kind.as_str(),
|
|
state = "send",
|
|
sender_thread_id = %context.sender_thread_id,
|
|
receiver_thread_id = %receiver_thread_id,
|
|
content = communication
|
|
.encrypted_content
|
|
.as_deref()
|
|
.unwrap_or("[plaintext]"),
|
|
},
|
|
"agent communication"
|
|
);
|
|
}
|
|
|
|
pub(crate) fn emit_agent_communication_receive(communication_id: &str) {
|
|
tracing::info!(
|
|
target: AGENT_COMMUNICATION_TARGET,
|
|
{
|
|
event.name = "codex.agent_communication",
|
|
communication_id,
|
|
state = "receive",
|
|
},
|
|
"agent communication"
|
|
);
|
|
}
|