openai/codex · error · std::io::Error

BrokenPipe

BrokenPipe

Error message

unknown process

What it means

ExecutorProcessTransport.send serializes an MCP message and writes it to a managed child process via a process registry. When the registry answers WriteStatus::UnknownProcess - the targeted process ID is not currently registered - send maps it to ErrorKind::BrokenPipe with message 'unknown process', signalling the write targeted a process the registry no longer knows.

Source

Thrown at codex-rs/rmcp-client/src/executor_process_transport.rs:253

        &mut self,
        item: TxJsonRpcMessage<RoleClient>,
    ) -> impl Future<Output = std::result::Result<(), Self::Error>> + Send + 'static {
        let process = Arc::clone(&self.process);
        let stdin_write_semaphore = Arc::clone(&self.stdin_write_semaphore);
        async move {
            let _stdin_write_permit = stdin_write_semaphore
                .acquire()
                .await
                .map_err(io::Error::other)?;
            // rmcp hands us a structured JSON-RPC message. Stdio transport on
            // the wire is JSON plus one newline delimiter.
            let mut bytes = to_vec(&item).map_err(io::Error::other)?;
            bytes.push(b'\n');
            let response = process.write(bytes).await.map_err(io::Error::other)?;
            match response.status {
                WriteStatus::Accepted => Ok(()),
                WriteStatus::UnknownProcess => {
                    Err(io::Error::new(io::ErrorKind::BrokenPipe, "unknown process"))
                }
                WriteStatus::StdinClosed => {
                    Err(io::Error::new(io::ErrorKind::BrokenPipe, "stdin closed"))
                }
                WriteStatus::Starting => Err(io::Error::new(
                    io::ErrorKind::WouldBlock,
                    "process is starting",
                )),
            }
        }
    }

    fn receive(&mut self) -> impl Future<Output = Option<RxJsonRpcMessage<RoleClient>>> + Send {
        self.receive_message()
    }

    async fn close(&mut self) -> std::result::Result<(), Self::Error> {
        self.process.terminate().await.map_err(io::Error::other)?;

View on GitHub (pinned to 339751715c)

Solutions

  1. Treat BrokenPipe 'unknown process' as terminal for the transport: drop it and build a new one bound to the current process
  2. Re-resolve the process ID from the registry before retrying the send
  3. Do not cache transports across restarts - re-create them on process lifecycle events

Example fix

// before
transport.send(item).await?; // BrokenPipe: unknown process
// after
match transport.send(item).await {
    Ok(()) => {}
    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
        let transport = rebuild_transport_for_current_process().await?;
        transport.send(item).await?;
    }
    Err(e) => return Err(e),
}
Defensive patterns

Strategy: fallback

Validate before calling

// if the registry is reachable, confirm the target exists before sending:
if !registry.contains(process_id) {
    return Err(stale_transport(process_id));
}

Try / catch

match transport.send(item).await {
    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe
        && e.to_string().contains("unknown process") =>
    {
        // stale transport: rebuild against the live registry and resend once
    }
    other => other,
}

Prevention

When it happens

Trigger: Calling send() on a transport after the target process was removed from (or never registered in) the registry: the process restarted with a new ID, the registry was cleared, or a stale transport handle was retained across a reconnect.

Common situations: Caching an ExecutorProcessTransport across a child restart; races between process teardown and in-flight sends; reconnect logic that reuses the pre-restart transport.

Related errors

AI-assisted analysis of openai/codex@339751715c (2026-08-25). Data as JSON: /api/errors/862e62ea6e8af36f. Report an issue: GitHub.