Merge 4ba46e794f into sapling-pr-archive-bolinfest

2026-11-14 21:40:39 +00:00 · 2026-06-23 15:51:47 -07:00
parent 25fd8fff77 4ba46e794f
commit 9d45d676b9
89 changed files with 1459 additions and 372 deletions
@@ -1,5 +1,6 @@
 # Core crate ownership.
 /codex-rs/core/ @openai/codex-core-agent-team
+/codex-rs/exec-server-protocol/ @openai/codex-core-agent-team
 /codex-rs/ext/extension-api/ @openai/codex-core-agent-team
 /codex-rs/prompts/ @openai/codex-core-agent-team
 /codex-rs/utils/path-uri/ @openai/codex-core-agent-team
@@ -2883,13 +2883,12 @@ dependencies = [
 "bytes",
 "clatter",
 "codex-api",
- "codex-app-server-protocol",
 "codex-client",
+ "codex-exec-server-protocol",
 "codex-file-system",
 "codex-network-proxy",
 "codex-protocol",
 "codex-sandboxing",
- "codex-shell-command",
 "codex-test-binary-support",
 "codex-utils-absolute-path",
 "codex-utils-path-uri",
@@ -2918,6 +2917,21 @@ dependencies = [
 "wiremock",
 ]

+[[package]]
+name = "codex-exec-server-protocol"
+version = "0.0.0"
+dependencies = [
+ "base64 0.22.1",
+ "codex-file-system",
+ "codex-network-proxy",
+ "codex-protocol",
+ "codex-shell-command",
+ "codex-utils-path-uri",
+ "pretty_assertions",
+ "serde",
+ "serde_json",
+]
+
 [[package]]
 name = "codex-execpolicy"
 version = "0.0.0"
@@ -44,6 +44,7 @@ members = [
    "secrets",
    "exec",
    "file-system",
+    "exec-server-protocol",
    "exec-server",
    "execpolicy",
    "execpolicy-legacy",
@@ -171,6 +172,7 @@ codex-core-plugins = { path = "core-plugins" }
 codex-core-skills = { path = "core-skills" }
 codex-exec = { path = "exec" }
 codex-file-system = { path = "file-system" }
+codex-exec-server-protocol = { path = "exec-server-protocol" }
 codex-exec-server = { path = "exec-server" }
 codex-execpolicy = { path = "execpolicy" }
 codex-extension-api = { path = "ext/extension-api" }
@@ -139,6 +139,7 @@ impl TurnCodexErrorFact {
 #[serde(rename_all = "snake_case")]
 pub enum CodexErrKind {
    TurnAborted,
+    RolloutBudgetExceeded,
    Stream,
    ContextWindowExceeded,
    ThreadNotFound,
@@ -195,6 +196,7 @@ impl From<&CodexErr> for CodexErrKind {
    fn from(error: &CodexErr) -> Self {
        match error {
            CodexErr::TurnAborted => CodexErrKind::TurnAborted,
+            CodexErr::RolloutBudgetExceeded => CodexErrKind::RolloutBudgetExceeded,
            CodexErr::Stream(..) => CodexErrKind::Stream,
            CodexErr::ContextWindowExceeded => CodexErrKind::ContextWindowExceeded,
            CodexErr::ThreadNotFound(_) => CodexErrKind::ThreadNotFound,
@@ -6834,4 +6834,4 @@
    }
  ],
  "title": "ClientRequest"
-}
+}
@@ -573,6 +573,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -7042,10 +7042,10 @@
              "environmentId": {
                "type": "string"
              },
-                "path": {
-                  "description": "Absolute path for the root in the selected environment.",
-                  "type": "string"
-                },
+              "path": {
+                "description": "Absolute path for the root in the selected environment.",
+                "type": "string"
+              },
              "type": {
                "enum": [
                  "environment"
@@ -7070,6 +7070,7 @@
          {
            "enum": [
              "contextWindowExceeded",
+              "rolloutBudgetExceeded",
              "usageLimitExceeded",
              "serverOverloaded",
              "cyberPolicy",
@@ -20690,4 +20691,4 @@
  },
  "title": "CodexAppServerProtocol",
  "type": "object"
-}
+}
@@ -3310,6 +3310,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -18468,4 +18469,4 @@
  },
  "title": "CodexAppServerProtocolV2",
  "type": "object"
-}
+}
@@ -7,6 +7,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -30,6 +30,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -111,6 +111,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -33,6 +33,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -33,6 +33,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -33,6 +33,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -111,6 +111,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -33,6 +33,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -388,4 +388,4 @@
  },
  "title": "ThreadStartParams",
  "type": "object"
-}
+}
@@ -111,6 +111,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -33,6 +33,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -33,6 +33,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -30,6 +30,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -30,6 +30,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -30,6 +30,7 @@
        {
          "enum": [
            "contextWindowExceeded",
+            "rolloutBudgetExceeded",
            "usageLimitExceeded",
            "serverOverloaded",
            "cyberPolicy",
@@ -9,4 +9,4 @@ import type { NonSteerableTurnKind } from "./NonSteerableTurnKind";
 * When an upstream HTTP status is available (for example, from the Responses API or a provider),
 * it is forwarded in `httpStatusCode` on the relevant `codexErrorInfo` variant.
 */
-export type CodexErrorInfo = "contextWindowExceeded" | "usageLimitExceeded" | "serverOverloaded" | "cyberPolicy" | { "httpConnectionFailed": { httpStatusCode: number | null, } } | { "responseStreamConnectionFailed": { httpStatusCode: number | null, } } | "internalServerError" | "unauthorized" | "badRequest" | "threadRollbackFailed" | "sandboxError" | { "responseStreamDisconnected": { httpStatusCode: number | null, } } | { "responseTooManyFailedAttempts": { httpStatusCode: number | null, } } | { "activeTurnNotSteerable": { turnKind: NonSteerableTurnKind, } } | "other";
+export type CodexErrorInfo = "contextWindowExceeded" | "rolloutBudgetExceeded" | "usageLimitExceeded" | "serverOverloaded" | "cyberPolicy" | { "httpConnectionFailed": { httpStatusCode: number | null, } } | { "responseStreamConnectionFailed": { httpStatusCode: number | null, } } | "internalServerError" | "unauthorized" | "badRequest" | "threadRollbackFailed" | "sandboxError" | { "responseStreamDisconnected": { httpStatusCode: number | null, } } | { "responseTooManyFailedAttempts": { httpStatusCode: number | null, } } | { "activeTurnNotSteerable": { turnKind: NonSteerableTurnKind, } } | "other";
@@ -1,7 +1,7 @@
 mod experimental_api;
 mod export;
-mod jsonrpc_lite;
 mod protocol;
+pub mod rpc;
 mod schema_fixtures;

 pub use experimental_api::*;
@@ -12,7 +12,6 @@ pub use export::generate_json_with_experimental;
 pub use export::generate_ts;
 pub use export::generate_ts_with_options;
 pub use export::generate_types;
-pub use jsonrpc_lite::*;
 pub use protocol::common::*;
 pub use protocol::event_mapping::*;
 pub use protocol::item_builders::*;
@@ -40,6 +39,7 @@ pub use protocol::v1::SandboxSettings;
 pub use protocol::v1::Tools;
 pub use protocol::v1::UserSavedConfig;
 pub use protocol::v2::*;
+pub use rpc::*;
 pub use schema_fixtures::SchemaFixtureOptions;
 #[doc(hidden)]
 pub use schema_fixtures::generate_typescript_schema_fixture_subtree_for_tests;
@@ -70,6 +70,7 @@ pub enum NonSteerableTurnKind {
 #[ts(export_to = "v2/")]
 pub enum CodexErrorInfo {
    ContextWindowExceeded,
+    RolloutBudgetExceeded,
    UsageLimitExceeded,
    ServerOverloaded,
    CyberPolicy,
@@ -115,6 +116,7 @@ impl From<CoreCodexErrorInfo> for CodexErrorInfo {
    fn from(value: CoreCodexErrorInfo) -> Self {
        match value {
            CoreCodexErrorInfo::ContextWindowExceeded => CodexErrorInfo::ContextWindowExceeded,
+            CoreCodexErrorInfo::RolloutBudgetExceeded => CodexErrorInfo::RolloutBudgetExceeded,
            CoreCodexErrorInfo::UsageLimitExceeded => CodexErrorInfo::UsageLimitExceeded,
            CoreCodexErrorInfo::ServerOverloaded => CodexErrorInfo::ServerOverloaded,
            CoreCodexErrorInfo::CyberPolicy => CodexErrorInfo::CyberPolicy,
@@ -1423,6 +1423,7 @@ There are additional item-specific events:
 `codexErrorInfo` maps to the `CodexErrorInfo` enum. Common values:

 - `ContextWindowExceeded`
+- `RolloutBudgetExceeded`
 - `UsageLimitExceeded`
 - `HttpConnectionFailed { httpStatusCode? }`: upstream HTTP failures including 4xx/5xx
 - `ResponseStreamConnectionFailed { httpStatusCode? }`: failure to connect to the response SSE stream
@@ -0,0 +1,19 @@
+use serde::Deserialize;
+use serde::Serialize;
+
+use super::Capability;
+use super::SupportedProtocolVersions;
+
+/// Explains why connection negotiation was rejected before any session opened.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, tag = "type", rename_all_fields = "camelCase")]
+pub enum HandshakeRejectReason {
+    #[serde(rename = "noCompatibleVersion")]
+    NoCompatibleVersion {
+        supported_versions: SupportedProtocolVersions,
+    },
+    #[serde(rename = "missingRequiredCapability")]
+    MissingRequiredCapability { capability: Capability },
+    #[serde(rename = "invalidHello")]
+    InvalidHello { message: String },
+}
@@ -0,0 +1,207 @@
+use pretty_assertions::assert_eq;
+use serde_json::json;
+
+use super::Capability;
+use super::CapabilitySet;
+use super::ClientHello;
+use super::ClientToHost;
+use super::HandshakeRejectReason;
+use super::HostHello;
+use super::HostToClient;
+use super::ProtocolVersion;
+use super::SessionId;
+use super::SupportedProtocolVersions;
+
+fn session_id() -> SessionId {
+    SessionId::new("session-1").expect("valid session ID")
+}
+
+fn capability(value: &str) -> Capability {
+    Capability::new(value).expect("valid capability")
+}
+
+fn supported_versions() -> SupportedProtocolVersions {
+    SupportedProtocolVersions::try_new([ProtocolVersion::V1])
+        .expect("nonempty unique protocol versions")
+}
+
+#[test]
+fn handshake_wire_contract_is_explicit_and_round_trips() {
+    let client_hello = ClientToHost::ClientHello(
+        ClientHello::new(
+            supported_versions(),
+            CapabilitySet::try_new([capability("required")]).expect("valid required set"),
+            CapabilitySet::try_new([capability("optional")]).expect("valid optional set"),
+        )
+        .expect("disjoint capabilities"),
+    );
+    let client_hello_json = json!({
+        "type": "connection/hello",
+        "supportedVersions": [1],
+        "requiredCapabilities": ["required"],
+        "optionalCapabilities": ["optional"],
+    });
+    assert_eq!(
+        serde_json::to_value(&client_hello).expect("serialize"),
+        client_hello_json
+    );
+    assert_eq!(
+        serde_json::from_value::<ClientToHost>(client_hello_json).expect("deserialize"),
+        client_hello
+    );
+
+    let host_hello = HostToClient::HostHello(HostHello::new(
+        ProtocolVersion::V1,
+        CapabilitySet::try_new([capability("required")]).expect("valid capabilities"),
+    ));
+    let host_hello_json = json!({
+        "type": "connection/ready",
+        "selectedVersion": 1,
+        "capabilities": ["required"],
+    });
+    assert_eq!(
+        serde_json::to_value(&host_hello).expect("serialize"),
+        host_hello_json
+    );
+    assert_eq!(
+        serde_json::from_value::<HostToClient>(host_hello_json).expect("deserialize"),
+        host_hello
+    );
+
+    let rejected = HostToClient::HandshakeRejected {
+        reason: HandshakeRejectReason::NoCompatibleVersion {
+            supported_versions: supported_versions(),
+        },
+    };
+    let rejected_json = json!({
+        "type": "connection/rejected",
+        "reason": {
+            "type": "noCompatibleVersion",
+            "supportedVersions": [1],
+        },
+    });
+    assert_eq!(
+        serde_json::to_value(&rejected).expect("serialize"),
+        rejected_json
+    );
+    assert_eq!(
+        serde_json::from_value::<HostToClient>(rejected_json).expect("deserialize"),
+        rejected
+    );
+}
+
+#[test]
+fn session_lifecycle_wire_contract_is_explicit_and_round_trips() {
+    let client_messages = [
+        (
+            ClientToHost::OpenSession {
+                session_id: session_id(),
+            },
+            json!({ "type": "session/open", "sessionId": "session-1" }),
+        ),
+        (
+            ClientToHost::CloseSession {
+                session_id: session_id(),
+            },
+            json!({ "type": "session/close", "sessionId": "session-1" }),
+        ),
+    ];
+    for (message, encoded) in client_messages {
+        assert_eq!(serde_json::to_value(&message).expect("serialize"), encoded);
+        assert_eq!(
+            serde_json::from_value::<ClientToHost>(encoded).expect("deserialize"),
+            message
+        );
+    }
+
+    let host_messages = [
+        (
+            HostToClient::SessionReady {
+                session_id: session_id(),
+            },
+            json!({ "type": "session/ready", "sessionId": "session-1" }),
+        ),
+        (
+            HostToClient::SessionClosed {
+                session_id: session_id(),
+            },
+            json!({ "type": "session/closed", "sessionId": "session-1" }),
+        ),
+    ];
+    for (message, encoded) in host_messages {
+        assert_eq!(serde_json::to_value(&message).expect("serialize"), encoded);
+        assert_eq!(
+            serde_json::from_value::<HostToClient>(encoded).expect("deserialize"),
+            message
+        );
+    }
+}
+
+#[test]
+fn invalid_protocol_states_cannot_be_constructed_or_decoded() {
+    assert!(SessionId::new("").is_err());
+    assert!(Capability::new("   ").is_err());
+    assert!(ProtocolVersion::new(/*value*/ 0).is_none());
+    assert!(SupportedProtocolVersions::try_new([]).is_err());
+    assert!(
+        SupportedProtocolVersions::try_new([ProtocolVersion::V1, ProtocolVersion::V1]).is_err()
+    );
+    assert!(CapabilitySet::try_new([capability("same"), capability("same")]).is_err());
+
+    let version_two = ProtocolVersion::new(/*value*/ 2).expect("valid protocol version");
+    let versions = SupportedProtocolVersions::try_new([ProtocolVersion::V1, version_two])
+        .expect("valid versions");
+    assert!(versions.contains(ProtocolVersion::V1));
+    assert_eq!(
+        versions.iter().collect::<Vec<_>>(),
+        vec![ProtocolVersion::V1, version_two]
+    );
+
+    let overlapping = capability("overlapping");
+    assert!(
+        ClientHello::new(
+            supported_versions(),
+            CapabilitySet::try_new([overlapping.clone()]).expect("valid required set"),
+            CapabilitySet::try_new([overlapping]).expect("valid optional set"),
+        )
+        .is_err()
+    );
+
+    for invalid in [
+        json!({ "type": "session/open", "sessionId": "" }),
+        json!({
+            "type": "connection/hello",
+            "supportedVersions": [],
+            "requiredCapabilities": [],
+            "optionalCapabilities": [],
+        }),
+        json!({
+            "type": "connection/hello",
+            "supportedVersions": [1],
+            "requiredCapabilities": ["overlapping"],
+            "optionalCapabilities": ["overlapping"],
+        }),
+    ] {
+        assert!(serde_json::from_value::<ClientToHost>(invalid).is_err());
+    }
+}
+
+#[test]
+fn unknown_fields_are_rejected() {
+    assert!(
+        serde_json::from_value::<ClientToHost>(json!({
+            "type": "session/open",
+            "sessionId": "session-1",
+            "unexpected": true,
+        }))
+        .is_err()
+    );
+    assert!(
+        serde_json::from_value::<HostToClient>(json!({
+            "type": "session/ready",
+            "sessionId": "session-1",
+            "unexpected": true,
+        }))
+        .is_err()
+    );
+}
@@ -0,0 +1,142 @@
+use std::fmt;
+
+use serde::Deserialize;
+use serde::Serialize;
+
+use super::Capability;
+use super::CapabilitySet;
+use super::HandshakeRejectReason;
+use super::ProtocolVersion;
+use super::SessionId;
+use super::SupportedProtocolVersions;
+
+#[derive(Clone, Debug, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ClientHello {
+    supported_versions: SupportedProtocolVersions,
+    required_capabilities: CapabilitySet,
+    optional_capabilities: CapabilitySet,
+}
+
+impl ClientHello {
+    pub fn new(
+        supported_versions: SupportedProtocolVersions,
+        required_capabilities: CapabilitySet,
+        optional_capabilities: CapabilitySet,
+    ) -> Result<Self, ClientHelloError> {
+        if let Some(capability) = required_capabilities
+            .iter()
+            .find(|capability| optional_capabilities.contains(capability))
+        {
+            return Err(ClientHelloError::OverlappingCapability(capability.clone()));
+        }
+        Ok(Self {
+            supported_versions,
+            required_capabilities,
+            optional_capabilities,
+        })
+    }
+
+    pub fn supported_versions(&self) -> &SupportedProtocolVersions {
+        &self.supported_versions
+    }
+
+    pub fn required_capabilities(&self) -> &CapabilitySet {
+        &self.required_capabilities
+    }
+
+    pub fn optional_capabilities(&self) -> &CapabilitySet {
+        &self.optional_capabilities
+    }
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+struct ClientHelloWire {
+    supported_versions: SupportedProtocolVersions,
+    required_capabilities: CapabilitySet,
+    optional_capabilities: CapabilitySet,
+}
+
+impl<'de> Deserialize<'de> for ClientHello {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let wire = ClientHelloWire::deserialize(deserializer)?;
+        Self::new(
+            wire.supported_versions,
+            wire.required_capabilities,
+            wire.optional_capabilities,
+        )
+        .map_err(serde::de::Error::custom)
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ClientHelloError {
+    OverlappingCapability(Capability),
+}
+
+impl fmt::Display for ClientHelloError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::OverlappingCapability(capability) => write!(
+                formatter,
+                "capability `{capability}` cannot be both required and optional"
+            ),
+        }
+    }
+}
+
+impl std::error::Error for ClientHelloError {}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub struct HostHello {
+    selected_version: ProtocolVersion,
+    capabilities: CapabilitySet,
+}
+
+impl HostHello {
+    pub fn new(selected_version: ProtocolVersion, capabilities: CapabilitySet) -> Self {
+        Self {
+            selected_version,
+            capabilities,
+        }
+    }
+
+    pub fn selected_version(&self) -> ProtocolVersion {
+        self.selected_version
+    }
+
+    pub fn capabilities(&self) -> &CapabilitySet {
+        &self.capabilities
+    }
+}
+
+/// Messages sent from a client to the code-mode host.
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, tag = "type", rename_all_fields = "camelCase")]
+pub enum ClientToHost {
+    #[serde(rename = "connection/hello")]
+    ClientHello(ClientHello),
+    #[serde(rename = "session/open")]
+    OpenSession { session_id: SessionId },
+    #[serde(rename = "session/close")]
+    CloseSession { session_id: SessionId },
+}
+
+/// Messages sent from the code-mode host to a client.
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, tag = "type", rename_all_fields = "camelCase")]
+pub enum HostToClient {
+    #[serde(rename = "connection/ready")]
+    HostHello(HostHello),
+    #[serde(rename = "connection/rejected")]
+    HandshakeRejected { reason: HandshakeRejectReason },
+    #[serde(rename = "session/ready")]
+    SessionReady { session_id: SessionId },
+    #[serde(rename = "session/closed")]
+    SessionClosed { session_id: SessionId },
+}
@@ -0,0 +1,29 @@
+//! Transport-neutral messages for the callback-only code-mode host boundary.
+//!
+//! Protocol version 1 relies on ordered framing and connection-scoped
+//! fail-stop behavior rather than message sequence numbers. It defines no
+//! optional capabilities yet; capability names provide an extension point for
+//! later versions without weakening the v1 decoder.
+
+mod error;
+mod message;
+mod types;
+
+pub use error::HandshakeRejectReason;
+pub use message::ClientHello;
+pub use message::ClientHelloError;
+pub use message::ClientToHost;
+pub use message::HostHello;
+pub use message::HostToClient;
+pub use types::Capability;
+pub use types::CapabilitySet;
+pub use types::DuplicateCapability;
+pub use types::InvalidIdentifier;
+pub use types::InvalidSupportedProtocolVersions;
+pub use types::ProtocolVersion;
+pub use types::SessionId;
+pub use types::SupportedProtocolVersions;
+
+#[cfg(test)]
+#[path = "host_tests.rs"]
+mod tests;
@@ -0,0 +1,226 @@
+use std::collections::BTreeSet;
+use std::fmt;
+use std::num::NonZeroU32;
+
+use serde::Deserialize;
+use serde::Deserializer;
+use serde::Serialize;
+use serde::Serializer;
+use serde::de::Error as _;
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
+#[serde(transparent)]
+pub struct ProtocolVersion(NonZeroU32);
+
+impl ProtocolVersion {
+    pub const V1: Self = Self(NonZeroU32::MIN);
+
+    pub const fn new(value: u32) -> Option<Self> {
+        match NonZeroU32::new(value) {
+            Some(value) => Some(Self(value)),
+            None => None,
+        }
+    }
+
+    pub const fn get(self) -> u32 {
+        self.0.get()
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct InvalidIdentifier;
+
+impl fmt::Display for InvalidIdentifier {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("identifier must not be empty")
+    }
+}
+
+impl std::error::Error for InvalidIdentifier {}
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+struct NonEmptyString(String);
+
+impl NonEmptyString {
+    fn new(value: impl Into<String>) -> Result<Self, InvalidIdentifier> {
+        let value = value.into();
+        if value.trim().is_empty() {
+            Err(InvalidIdentifier)
+        } else {
+            Ok(Self(value))
+        }
+    }
+
+    fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl Serialize for NonEmptyString {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        self.0.serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for NonEmptyString {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Self::new(String::deserialize(deserializer)?).map_err(D::Error::custom)
+    }
+}
+
+/// A named protocol feature advertised during connection negotiation.
+#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
+#[serde(transparent)]
+pub struct Capability(NonEmptyString);
+
+impl Capability {
+    pub fn new(value: impl Into<String>) -> Result<Self, InvalidIdentifier> {
+        NonEmptyString::new(value).map(Self)
+    }
+
+    pub fn as_str(&self) -> &str {
+        self.0.as_str()
+    }
+}
+
+impl fmt::Display for Capability {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(self.as_str())
+    }
+}
+
+/// Identifies one logical code-mode session on a connection.
+#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
+#[serde(transparent)]
+pub struct SessionId(NonEmptyString);
+
+impl SessionId {
+    pub fn new(value: impl Into<String>) -> Result<Self, InvalidIdentifier> {
+        NonEmptyString::new(value).map(Self)
+    }
+
+    pub fn as_str(&self) -> &str {
+        self.0.as_str()
+    }
+}
+
+impl fmt::Display for SessionId {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(self.as_str())
+    }
+}
+
+#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
+#[serde(transparent)]
+pub struct CapabilitySet(BTreeSet<Capability>);
+
+impl CapabilitySet {
+    pub fn empty() -> Self {
+        Self::default()
+    }
+
+    pub fn try_new(
+        capabilities: impl IntoIterator<Item = Capability>,
+    ) -> Result<Self, DuplicateCapability> {
+        let mut unique = BTreeSet::new();
+        for capability in capabilities {
+            if !unique.insert(capability.clone()) {
+                return Err(DuplicateCapability { capability });
+            }
+        }
+        Ok(Self(unique))
+    }
+
+    pub fn contains(&self, capability: &Capability) -> bool {
+        self.0.contains(capability)
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &Capability> {
+        self.0.iter()
+    }
+}
+
+impl<'de> Deserialize<'de> for CapabilitySet {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Self::try_new(Vec::<Capability>::deserialize(deserializer)?).map_err(D::Error::custom)
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct DuplicateCapability {
+    capability: Capability,
+}
+
+impl fmt::Display for DuplicateCapability {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(formatter, "duplicate capability `{}`", self.capability)
+    }
+}
+
+impl std::error::Error for DuplicateCapability {}
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+#[serde(transparent)]
+pub struct SupportedProtocolVersions(BTreeSet<ProtocolVersion>);
+
+impl SupportedProtocolVersions {
+    pub fn try_new(
+        versions: impl IntoIterator<Item = ProtocolVersion>,
+    ) -> Result<Self, InvalidSupportedProtocolVersions> {
+        let mut unique = BTreeSet::new();
+        for version in versions {
+            if !unique.insert(version) {
+                return Err(InvalidSupportedProtocolVersions::Duplicate(version));
+            }
+        }
+        if unique.is_empty() {
+            return Err(InvalidSupportedProtocolVersions::Empty);
+        }
+        Ok(Self(unique))
+    }
+
+    pub fn contains(&self, version: ProtocolVersion) -> bool {
+        self.0.contains(&version)
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = ProtocolVersion> + '_ {
+        self.0.iter().copied()
+    }
+}
+
+impl<'de> Deserialize<'de> for SupportedProtocolVersions {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Self::try_new(Vec::<ProtocolVersion>::deserialize(deserializer)?).map_err(D::Error::custom)
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum InvalidSupportedProtocolVersions {
+    Empty,
+    Duplicate(ProtocolVersion),
+}
+
+impl fmt::Display for InvalidSupportedProtocolVersions {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Empty => formatter.write_str("at least one protocol version is required"),
+            Self::Duplicate(version) => {
+                write!(formatter, "duplicate protocol version {}", version.get())
+            }
+        }
+    }
+}
+
+impl std::error::Error for InvalidSupportedProtocolVersions {}
@@ -1,4 +1,5 @@
 mod description;
+pub mod host;
 mod response;
 mod runtime;
 mod session;
@@ -110,12 +110,6 @@ pub trait CodeModeSessionDelegate: Send + Sync {
 /// must keep those values isolated. Implementations may execute cells
 /// in-process or remotely.
 pub trait CodeModeSession: Send + Sync {
-    /// Returns whether the session can still accept requests.
-    ///
-    /// Remote implementations should return `false` after their underlying
-    /// connection fails so callers can create a fresh session for later work.
-    fn is_alive(&self) -> bool;
-
    fn execute<'a>(
        &'a self,
        request: ExecuteRequest,
@@ -216,10 +216,6 @@ impl Default for InProcessCodeModeSession {
 }

 impl CodeModeSession for InProcessCodeModeSession {
-    fn is_alive(&self) -> bool {
-        self.runtime.is_alive()
-    }
-
    fn execute<'a>(
        &'a self,
        request: ExecuteRequest,
@@ -63,10 +63,6 @@ impl<D: SessionRuntimeDelegate> SessionRuntime<D> {
        }
    }

-    pub(crate) fn is_alive(&self) -> bool {
-        !self.inner.shutdown_token.is_cancelled()
-    }
-
    pub(crate) async fn execute(
        &self,
        request: CreateCellRequest,
@@ -152,7 +152,6 @@ async fn shutdown_rejects_cell_admission_queued_before_the_registry_lock() {
    })
    .await;

-    assert!(!runtime.is_alive());
    drop(cells);
    assert!(matches!(execution.await, Err(Error::ShuttingDown)));
    assert_eq!(shutdown.await, Ok(()));
@@ -1,3 +1,9 @@
+mod environment;
+
+pub use environment::EnvironmentSkillLoadOutcome;
+pub use environment::EnvironmentSkillMetadata;
+pub use environment::load_environment_skills_from_root;
+
 use crate::model::SkillDependencies;
 use crate::model::SkillError;
 use crate::model::SkillInterface;
@@ -104,6 +110,13 @@ struct DependencyTool {
    url: Option<String>,
 }

+#[derive(Debug, Clone, PartialEq, Eq)]
+struct ParsedSkillFrontmatter {
+    name: String,
+    description: String,
+    short_description: Option<String>,
+}
+
 const SKILLS_FILENAME: &str = "SKILL.md";
 const AGENTS_DIR_NAME: &str = ".agents";
 const SKILLS_METADATA_DIR: &str = "agents";
@@ -124,6 +137,17 @@ const MAX_DEPENDENCY_URL_LEN: usize = MAX_DESCRIPTION_LEN;
 const MAX_SCAN_DEPTH: usize = 6;
 const MAX_SKILLS_DIRS_PER_ROOT: usize = 2000;

+#[derive(Clone, Copy)]
+enum SymlinkPolicy {
+    FollowDirectories,
+    Ignore,
+}
+
+struct SkillFileDiscovery {
+    skill_files: Vec<PathUri>,
+    warnings: Vec<String>,
+}
+
 #[derive(Debug)]
 enum SkillParseError {
    Read(std::io::Error),
@@ -189,7 +213,7 @@ pub(crate) async fn load_skill_root(root: SkillRoot) -> SkillRootSnapshot {
    } = root;
    let root = canonicalize_for_skill_identity(file_system.as_ref(), &path).await;
    let mut outcome = SkillLoadOutcome::default();
-    discover_skills_under_root(
+    load_skills_under_root(
        file_system.as_ref(),
        &root,
        scope,
@@ -470,35 +494,31 @@ async fn canonicalize_for_skill_identity(

 async fn discover_skills_under_root(
    fs: &dyn ExecutorFileSystem,
-    root: &AbsolutePathBuf,
-    scope: SkillScope,
-    plugin_id: Option<&str>,
-    plugin_namespace: Option<&str>,
-    plugin_root: Option<&AbsolutePathBuf>,
-    outcome: &mut SkillLoadOutcome,
-) {
+    root: &PathUri,
+    symlink_policy: SymlinkPolicy,
+) -> SkillFileDiscovery {
    let root = root.clone();
-    let plugin_root = match plugin_root {
-        Some(plugin_root) => Some(canonicalize_for_skill_identity(fs, plugin_root).await),
-        None => None,
+    let mut discovery = SkillFileDiscovery {
+        skill_files: Vec::new(),
+        warnings: Vec::new(),
    };
-
-    let root_uri = PathUri::from_abs_path(&root);
-    match fs.get_metadata(&root_uri, /*sandbox*/ None).await {
+    match fs.get_metadata(&root, /*sandbox*/ None).await {
        Ok(metadata) if metadata.is_directory => {}
-        Ok(_) => return,
-        Err(err) if err.kind() == io::ErrorKind::NotFound => return,
+        Ok(_) => return discovery,
+        Err(err) if err.kind() == io::ErrorKind::NotFound => return discovery,
        Err(err) => {
-            error!("failed to stat skills root {}: {err:#}", root.display());
-            return;
+            discovery
+                .warnings
+                .push(format!("failed to stat skills root {root}: {err:#}"));
+            return discovery;
        }
    }

    fn enqueue_dir(
-        queue: &mut VecDeque<(AbsolutePathBuf, usize)>,
-        visited_dirs: &mut HashSet<AbsolutePathBuf>,
+        queue: &mut VecDeque<(PathUri, usize)>,
+        visited_dirs: &mut HashSet<PathUri>,
        truncated_by_dir_limit: &mut bool,
-        path: AbsolutePathBuf,
+        path: PathUri,
        depth: usize,
    ) {
        if depth > MAX_SCAN_DEPTH {
@@ -513,24 +533,18 @@ async fn discover_skills_under_root(
        }
    }

-    // Follow symlinked directories for user, admin, and repo skills. System skills are written by Codex itself.
-    let follow_symlinks = matches!(
-        scope,
-        SkillScope::Repo | SkillScope::User | SkillScope::Admin
-    );
-
-    let mut visited_dirs: HashSet<AbsolutePathBuf> = HashSet::new();
-    visited_dirs.insert(root.clone());
-
-    let mut queue: VecDeque<(AbsolutePathBuf, usize)> = VecDeque::from([(root.clone(), 0)]);
+    let follow_symlinks = matches!(symlink_policy, SymlinkPolicy::FollowDirectories);
+    let mut visited_dirs: HashSet<PathUri> = HashSet::from([root.clone()]);
+    let mut queue: VecDeque<(PathUri, usize)> = VecDeque::from([(root.clone(), 0)]);
    let mut truncated_by_dir_limit = false;

    while let Some((dir, depth)) = queue.pop_front() {
-        let dir_uri = PathUri::from_abs_path(&dir);
-        let entries = match fs.read_directory(&dir_uri, /*sandbox*/ None).await {
+        let entries = match fs.read_directory(&dir, /*sandbox*/ None).await {
            Ok(entries) => entries,
-            Err(e) => {
-                error!("failed to read skills dir {}: {e:#}", dir.display());
+            Err(err) => {
+                discovery
+                    .warnings
+                    .push(format!("failed to read skills directory {dir}: {err:#}"));
                continue;
            }
        };
@@ -542,25 +556,31 @@ async fn discover_skills_under_root(
                if file_name.starts_with('.') {
                    return None;
                }
-                let path = dir.join(&file_name);
-                let path_uri = PathUri::from_abs_path(&path);
-                Some((file_name, path, path_uri))
+                match dir.join(&file_name) {
+                    Ok(path) => Some((file_name, path)),
+                    Err(err) => {
+                        discovery.warnings.push(format!(
+                            "failed to resolve skill path {dir}/{file_name}: {err}"
+                        ));
+                        None
+                    }
+                }
            })
            .collect::<Vec<_>>();
        let metadata_results = join_all(
            paths
                .iter()
-                .map(|(_, _, path_uri)| fs.get_metadata(path_uri, /*sandbox*/ None)),
+                .map(|(_, path)| fs.get_metadata(path, /*sandbox*/ None)),
        )
        .await;

-        for ((file_name, path, path_uri), metadata_result) in
-            paths.into_iter().zip(metadata_results)
-        {
+        for ((file_name, path), metadata_result) in paths.into_iter().zip(metadata_results) {
            let metadata = match metadata_result {
                Ok(metadata) => metadata,
-                Err(e) => {
-                    error!("failed to stat skills path {}: {e:#}", path.display());
+                Err(err) => {
+                    discovery
+                        .warnings
+                        .push(format!("failed to stat skill path {path}: {err:#}"));
                    continue;
                }
            };
@@ -569,9 +589,9 @@ async fn discover_skills_under_root(
                if !follow_symlinks {
                    continue;
                }
-                match fs.read_directory(&path_uri, /*sandbox*/ None).await {
+                match fs.read_directory(&path, /*sandbox*/ None).await {
                    Ok(_) => {
-                        let resolved_dir = canonicalize_for_skill_identity(fs, &path).await;
+                        let resolved_dir = canonicalize_uri_for_skill_identity(fs, &path).await;
                        enqueue_dir(
                            &mut queue,
                            &mut visited_dirs,
@@ -585,51 +605,26 @@ async fn discover_skills_under_root(
                            err.kind(),
                            io::ErrorKind::NotADirectory | io::ErrorKind::NotFound
                        ) => {}
-                    Err(err) => {
-                        error!(
-                            "failed to read skills symlink dir {}: {err:#}",
-                            path.display()
-                        );
-                    }
+                    Err(err) => discovery.warnings.push(format!(
+                        "failed to read skills symlink directory {path}: {err:#}"
+                    )),
                }
                continue;
            }

            if metadata.is_directory {
-                let resolved_dir = canonicalize_for_skill_identity(fs, &path).await;
                enqueue_dir(
                    &mut queue,
                    &mut visited_dirs,
                    &mut truncated_by_dir_limit,
-                    resolved_dir,
+                    path,
                    depth + 1,
                );
                continue;
            }

            if metadata.is_file && file_name == SKILLS_FILENAME {
-                match parse_skill_file(
-                    fs,
-                    &path,
-                    scope,
-                    plugin_id,
-                    plugin_namespace,
-                    plugin_root.as_ref(),
-                )
-                .await
-                {
-                    Ok(skill) => {
-                        outcome.skills.push(skill);
-                    }
-                    Err(err) => {
-                        if scope != SkillScope::System {
-                            outcome.errors.push(SkillError {
-                                path: path.clone(),
-                                message: err.to_string(),
-                            });
-                        }
-                    }
-                }
+                discovery.skill_files.push(path);
            }
        }
    }
@@ -638,9 +633,72 @@ async fn discover_skills_under_root(
        tracing::warn!(
            "skills scan truncated after {} directories (root: {})",
            MAX_SKILLS_DIRS_PER_ROOT,
-            root.display()
+            root
        );
    }
+    discovery
+}
+
+async fn canonicalize_uri_for_skill_identity(
+    file_system: &dyn ExecutorFileSystem,
+    path: &PathUri,
+) -> PathUri {
+    file_system
+        .canonicalize(path, /*sandbox*/ None)
+        .await
+        .unwrap_or_else(|_| path.clone())
+}
+
+async fn load_skills_under_root(
+    fs: &dyn ExecutorFileSystem,
+    root: &AbsolutePathBuf,
+    scope: SkillScope,
+    plugin_id: Option<&str>,
+    plugin_namespace: Option<&str>,
+    plugin_root: Option<&AbsolutePathBuf>,
+    outcome: &mut SkillLoadOutcome,
+) {
+    let plugin_root = match plugin_root {
+        Some(plugin_root) => Some(canonicalize_for_skill_identity(fs, plugin_root).await),
+        None => None,
+    };
+    let symlink_policy = match scope {
+        SkillScope::User | SkillScope::Repo | SkillScope::Admin => SymlinkPolicy::FollowDirectories,
+        SkillScope::System => SymlinkPolicy::Ignore,
+    };
+    let SkillFileDiscovery {
+        skill_files,
+        warnings,
+    } = discover_skills_under_root(fs, &PathUri::from_abs_path(root), symlink_policy).await;
+    for warning in warnings {
+        error!("{warning}");
+    }
+    for path_uri in skill_files {
+        let path = match path_uri.to_abs_path() {
+            Ok(path) => path,
+            Err(err) => {
+                error!("failed to convert discovered skill path {path_uri}: {err}");
+                continue;
+            }
+        };
+        match parse_skill_file(
+            fs,
+            &path,
+            scope,
+            plugin_id,
+            plugin_namespace,
+            plugin_root.as_ref(),
+        )
+        .await
+        {
+            Ok(skill) => outcome.skills.push(skill),
+            Err(err) if scope != SkillScope::System => outcome.errors.push(SkillError {
+                path,
+                message: err.to_string(),
+            }),
+            Err(_) => {}
+        }
+    }
 }

 async fn parse_skill_file(
@@ -656,8 +714,40 @@ async fn parse_skill_file(
        .read_file_text(&path_uri, /*sandbox*/ None)
        .await
        .map_err(SkillParseError::Read)?;
+    let ParsedSkillFrontmatter {
+        name: base_name,
+        description,
+        short_description,
+    } = parse_skill_frontmatter_metadata_inner(&contents, || default_skill_name(path))?;
+    let name = namespaced_skill_name(fs, path, &base_name, plugin_namespace).await;
+    let LoadedSkillMetadata {
+        interface,
+        dependencies,
+        policy,
+    } = load_skill_metadata(fs, path, plugin_root).await;

-    let frontmatter = extract_frontmatter(&contents).ok_or(SkillParseError::MissingFrontmatter)?;
+    validate_len(&name, MAX_QUALIFIED_NAME_LEN, "qualified name")?;
+
+    let resolved_path = canonicalize_for_skill_identity(fs, path).await;
+
+    Ok(SkillMetadata {
+        name,
+        description,
+        short_description,
+        interface,
+        dependencies,
+        policy,
+        path_to_skills_md: resolved_path,
+        scope,
+        plugin_id: plugin_id.map(str::to_string),
+    })
+}
+
+fn parse_skill_frontmatter_metadata_inner(
+    contents: &str,
+    default_name: impl FnOnce() -> String,
+) -> Result<ParsedSkillFrontmatter, SkillParseError> {
+    let frontmatter = extract_frontmatter(contents).ok_or(SkillParseError::MissingFrontmatter)?;

    let parsed: SkillFrontmatter = match serde_yaml::from_str(&frontmatter) {
        Ok(parsed) => Ok(parsed),
@@ -673,13 +763,12 @@ async fn parse_skill_file(
    }
    .map_err(SkillParseError::InvalidYaml)?;

-    let base_name = parsed
+    let name = parsed
        .name
        .as_deref()
        .map(sanitize_single_line)
        .filter(|value| !value.is_empty())
-        .unwrap_or_else(|| default_skill_name(path));
-    let name = namespaced_skill_name(fs, path, &base_name, plugin_namespace).await;
+        .unwrap_or_else(default_name);
    let description = parsed
        .description
        .as_deref()
@@ -691,30 +780,16 @@ async fn parse_skill_file(
        .as_deref()
        .map(sanitize_single_line)
        .filter(|value| !value.is_empty());
-    let LoadedSkillMetadata {
-        interface,
-        dependencies,
-        policy,
-    } = load_skill_metadata(fs, path, plugin_root).await;

-    validate_len(&base_name, MAX_NAME_LEN, "name")?;
-    validate_len(&name, MAX_QUALIFIED_NAME_LEN, "qualified name")?;
+    validate_len(&name, MAX_NAME_LEN, "name")?;
    if description.is_empty() {
        return Err(SkillParseError::MissingField("description"));
    }

-    let resolved_path = canonicalize_for_skill_identity(fs, path).await;
-
-    Ok(SkillMetadata {
+    Ok(ParsedSkillFrontmatter {
        name,
        description,
        short_description,
-        interface,
-        dependencies,
-        policy,
-        path_to_skills_md: resolved_path,
-        scope,
-        plugin_id: plugin_id.map(str::to_string),
    })
 }

@@ -0,0 +1,180 @@
+use std::io;
+
+use codex_exec_server::ExecutorFileSystem;
+use codex_protocol::protocol::Product;
+use codex_utils_path_uri::PathUri;
+use codex_utils_plugins::plugin_namespace_for_skill_uri;
+
+use crate::model::SkillDependencies;
+use crate::model::SkillPolicy;
+
+use super::MAX_QUALIFIED_NAME_LEN;
+use super::ParsedSkillFrontmatter;
+use super::SKILLS_METADATA_DIR;
+use super::SKILLS_METADATA_FILENAME;
+use super::SkillMetadataFile;
+use super::SymlinkPolicy;
+use super::discover_skills_under_root;
+use super::parse_skill_frontmatter_metadata_inner;
+use super::resolve_dependencies;
+use super::resolve_policy;
+use super::sanitize_single_line;
+use super::validate_len;
+
+/// URI-native metadata for one skill owned by an execution environment.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct EnvironmentSkillMetadata {
+    pub path_to_skills_md: PathUri,
+    pub name: String,
+    pub description: String,
+    pub short_description: Option<String>,
+    pub dependencies: Option<SkillDependencies>,
+    pub policy: Option<SkillPolicy>,
+}
+
+impl EnvironmentSkillMetadata {
+    pub fn allows_implicit_invocation(&self) -> bool {
+        self.policy
+            .as_ref()
+            .and_then(|policy| policy.allow_implicit_invocation)
+            .unwrap_or(true)
+    }
+
+    fn matches_product_restriction(&self, restriction_product: Option<Product>) -> bool {
+        match &self.policy {
+            Some(policy) => {
+                policy.products.is_empty()
+                    || restriction_product.is_some_and(|product| {
+                        product.matches_product_restriction(&policy.products)
+                    })
+            }
+            None => true,
+        }
+    }
+
+    async fn parse(file_system: &dyn ExecutorFileSystem, path: &PathUri) -> Result<Self, String> {
+        let contents = file_system
+            .read_file_text(path, /*sandbox*/ None)
+            .await
+            .map_err(|err| format!("failed to read file: {err}"))?;
+        let ParsedSkillFrontmatter {
+            name: base_name,
+            description,
+            short_description,
+        } = parse_skill_frontmatter_metadata_inner(&contents, || default_skill_name(path))
+            .map_err(|err| err.to_string())?;
+        let name = plugin_namespace_for_skill_uri(file_system, path)
+            .await
+            .map(|namespace| format!("{namespace}:{base_name}"))
+            .unwrap_or(base_name);
+        validate_len(&name, MAX_QUALIFIED_NAME_LEN, "qualified name")
+            .map_err(|err| err.to_string())?;
+        let (dependencies, policy) = load_skill_metadata(file_system, path).await;
+
+        Ok(Self {
+            path_to_skills_md: path.clone(),
+            name,
+            description,
+            short_description,
+            dependencies,
+            policy,
+        })
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct EnvironmentSkillLoadOutcome {
+    pub skills: Vec<EnvironmentSkillMetadata>,
+    pub warnings: Vec<String>,
+}
+
+/// Discovers skills without converting environment-owned paths to host paths.
+pub async fn load_environment_skills_from_root(
+    file_system: &dyn ExecutorFileSystem,
+    root: &PathUri,
+    restriction_product: Option<Product>,
+) -> EnvironmentSkillLoadOutcome {
+    let mut outcome = EnvironmentSkillLoadOutcome::default();
+    let discovery =
+        discover_skills_under_root(file_system, root, SymlinkPolicy::FollowDirectories).await;
+    outcome.warnings.extend(discovery.warnings);
+    for path in discovery.skill_files {
+        match EnvironmentSkillMetadata::parse(file_system, &path).await {
+            Ok(skill) if skill.matches_product_restriction(restriction_product) => {
+                outcome.skills.push(skill);
+            }
+            Ok(_) => {}
+            Err(message) => outcome.warnings.push(format!(
+                "Failed to load environment skill at {path}: {message}"
+            )),
+        }
+    }
+    outcome.skills.sort_by(|left, right| {
+        left.name.cmp(&right.name).then_with(|| {
+            left.path_to_skills_md
+                .to_string()
+                .cmp(&right.path_to_skills_md.to_string())
+        })
+    });
+    outcome
+}
+
+async fn load_skill_metadata(
+    file_system: &dyn ExecutorFileSystem,
+    skill_path: &PathUri,
+) -> (Option<SkillDependencies>, Option<SkillPolicy>) {
+    let Some(skill_dir) = skill_path.parent() else {
+        return (None, None);
+    };
+    let Ok(metadata_path) =
+        skill_dir.join(&format!("{SKILLS_METADATA_DIR}/{SKILLS_METADATA_FILENAME}"))
+    else {
+        return (None, None);
+    };
+    match file_system
+        .get_metadata(&metadata_path, /*sandbox*/ None)
+        .await
+    {
+        Ok(metadata) if metadata.is_file => {}
+        Ok(_) => return (None, None),
+        Err(error) if error.kind() == io::ErrorKind::NotFound => return (None, None),
+        Err(error) => {
+            tracing::warn!("ignoring {metadata_path}: failed to stat metadata: {error}");
+            return (None, None);
+        }
+    }
+    let contents = match file_system
+        .read_file_text(&metadata_path, /*sandbox*/ None)
+        .await
+    {
+        Ok(contents) => contents,
+        Err(error) => {
+            tracing::warn!("ignoring {metadata_path}: failed to read metadata: {error}");
+            return (None, None);
+        }
+    };
+    let parsed: SkillMetadataFile = match serde_yaml::from_str(&contents) {
+        Ok(parsed) => parsed,
+        Err(error) => {
+            tracing::warn!("ignoring {metadata_path}: invalid metadata: {error}");
+            return (None, None);
+        }
+    };
+
+    (
+        resolve_dependencies(parsed.dependencies),
+        resolve_policy(parsed.policy),
+    )
+}
+
+fn default_skill_name(path: &PathUri) -> String {
+    path.parent()
+        .and_then(|parent| parent.basename())
+        .map(|name| sanitize_single_line(&name))
+        .filter(|name| !name.is_empty())
+        .unwrap_or_else(|| "skill".to_string())
+}
+
+#[cfg(test)]
+#[path = "environment_tests.rs"]
+mod tests;
@@ -0,0 +1,78 @@
+use std::fs;
+
+use codex_exec_server::LOCAL_FS;
+use codex_protocol::protocol::Product;
+use codex_utils_path_uri::PathUri;
+use pretty_assertions::assert_eq;
+use tempfile::tempdir;
+
+use crate::model::SkillDependencies;
+use crate::model::SkillPolicy;
+use crate::model::SkillToolDependency;
+
+use super::EnvironmentSkillMetadata;
+use super::load_environment_skills_from_root;
+
+#[tokio::test]
+async fn loads_plugin_namespace_dependencies_and_policy() {
+    let root = tempdir().expect("tempdir");
+    let skill_dir = root.path().join("skills/deploy");
+    fs::create_dir_all(root.path().join(".codex-plugin")).expect("manifest dir");
+    fs::create_dir_all(skill_dir.join("agents")).expect("metadata dir");
+    fs::write(
+        root.path().join(".codex-plugin/plugin.json"),
+        r#"{"name":"demo-plugin"}"#,
+    )
+    .expect("manifest");
+    fs::write(
+        skill_dir.join("SKILL.md"),
+        "---\nname: deploy\ndescription: Deploy the service.\n---\n",
+    )
+    .expect("skill");
+    fs::write(
+        skill_dir.join("agents/openai.yaml"),
+        r#"
+dependencies:
+  tools:
+    - type: mcp
+      value: deploy-server
+      description: Deploy MCP
+policy:
+  allow_implicit_invocation: false
+  products: [codex]
+"#,
+    )
+    .expect("metadata");
+
+    let root_uri = PathUri::from_host_native_path(root.path()).expect("root URI");
+    let outcome =
+        load_environment_skills_from_root(LOCAL_FS.as_ref(), &root_uri, Some(Product::Codex)).await;
+
+    assert_eq!(
+        outcome.skills,
+        vec![EnvironmentSkillMetadata {
+            path_to_skills_md: PathUri::from_host_native_path(skill_dir.join("SKILL.md"),).unwrap(),
+            name: "demo-plugin:deploy".to_string(),
+            description: "Deploy the service.".to_string(),
+            short_description: None,
+            dependencies: Some(SkillDependencies {
+                tools: vec![SkillToolDependency {
+                    r#type: "mcp".to_string(),
+                    value: "deploy-server".to_string(),
+                    description: Some("Deploy MCP".to_string()),
+                    transport: None,
+                    command: None,
+                    url: None,
+                }],
+            }),
+            policy: Some(SkillPolicy {
+                allow_implicit_invocation: Some(false),
+                products: vec![Product::Codex],
+            }),
+        }]
+    );
+    let filtered =
+        load_environment_skills_from_root(LOCAL_FS.as_ref(), &root_uri, Some(Product::Chatgpt))
+            .await;
+    assert!(filtered.skills.is_empty());
+}
@@ -275,6 +275,12 @@ async fn run_compact_task_inner_impl(
            Err(err @ (CodexErr::Interrupted | CodexErr::TurnAborted)) => {
                return Err(err);
            }
+            Err(e @ CodexErr::RolloutBudgetExceeded) => {
+                sess.track_turn_codex_error(turn_context.as_ref(), &e);
+                let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));
+                sess.send_event(&turn_context, event).await;
+                return Err(e);
+            }
            Err(e @ CodexErr::ContextWindowExceeded) => {
                if turn_input_len > 1 {
                    // Trim from the beginning to preserve cache (prefix-based) and keep recent messages intact.
@@ -30,7 +30,7 @@ impl Session {
            .rollout_budget()
            .record_usage(usage)
        {
-            return Err(CodexErr::TurnAborted);
+            return Err(CodexErr::RolloutBudgetExceeded);
        }
        Ok(())
    }
@@ -2,9 +2,9 @@ use anyhow::Result;
 use codex_core::config::RolloutBudgetConfig;
 use codex_features::Feature;
 use codex_model_provider_info::built_in_model_providers;
+use codex_protocol::protocol::CodexErrorInfo;
 use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::Op;
-use codex_protocol::protocol::TurnAbortReason;
 use codex_protocol::user_input::UserInput;
 use core_test_support::responses::ResponsesRequest;
 use core_test_support::responses::ev_assistant_message;
@@ -205,7 +205,7 @@ async fn subagent_usage_draws_from_the_shared_budget() -> Result<()> {
 }

 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn exhausted_budget_aborts_current_and_later_turns() -> Result<()> {
+async fn exhausted_budget_fails_current_and_later_turns() -> Result<()> {
    skip_if_no_network!(Ok(()));

    let server = start_mock_server().await;
@@ -248,18 +248,18 @@ async fn exhausted_budget_aborts_current_and_later_turns() -> Result<()> {
            })
            .await?;

-        let event = wait_for_event(&test.codex, |event| match event {
-            EventMsg::TurnAborted(_) => true,
-            EventMsg::TurnComplete(_) => {
-                panic!("exhausted budget completed the turn instead of aborting")
-            }
-            _ => false,
+        wait_for_event(&test.codex, |event| {
+            matches!(
+                event,
+                EventMsg::Error(error)
+                    if error.codex_error_info == Some(CodexErrorInfo::RolloutBudgetExceeded)
+            )
+        })
+        .await;
+        wait_for_event(&test.codex, |event| {
+            matches!(event, EventMsg::TurnComplete(_))
        })
        .await;
-        let EventMsg::TurnAborted(abort) = event else {
-            unreachable!("event filter only accepts TurnAborted")
-        };
-        assert_eq!(abort.reason, TurnAbortReason::Interrupted);
    }

    Ok(())
@@ -268,7 +268,7 @@ async fn exhausted_budget_aborts_current_and_later_turns() -> Result<()> {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 #[test_case(false ; "local")]
 #[test_case(true ; "remote_v2")]
-async fn compaction_budget_exhaustion_aborts_without_error_or_retry(remote_v2: bool) -> Result<()> {
+async fn compaction_budget_exhaustion_fails_without_retry(remote_v2: bool) -> Result<()> {
    skip_if_no_network!(Ok(()));

    let server = start_mock_server().await;
@@ -311,19 +311,18 @@ async fn compaction_budget_exhaustion_aborts_without_error_or_retry(remote_v2: b
        .await?;

    test.codex.submit(Op::Compact).await?;
-    let event = wait_for_event(&test.codex, |event| match event {
-        EventMsg::TurnAborted(_) => true,
-        EventMsg::Error(error) => panic!("budget exhaustion emitted an error: {}", error.message),
-        EventMsg::TurnComplete(_) => {
-            panic!("budget-exhausting compaction completed instead of aborting")
-        }
-        _ => false,
+    wait_for_event(&test.codex, |event| {
+        matches!(
+            event,
+            EventMsg::Error(error)
+                if error.codex_error_info == Some(CodexErrorInfo::RolloutBudgetExceeded)
+        )
+    })
+    .await;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
    })
    .await;
-    let EventMsg::TurnAborted(abort) = event else {
-        unreachable!("event filter only accepts TurnAborted")
-    };
-    assert_eq!(abort.reason, TurnAbortReason::Interrupted);
    assert_eq!(responses.requests().len(), 1, "compaction should not retry");

    Ok(())
@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "exec-server-protocol",
+    crate_name = "codex_exec_server_protocol",
+)
@@ -0,0 +1,26 @@
+[package]
+name = "codex-exec-server-protocol"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[lib]
+name = "codex_exec_server_protocol"
+path = "src/lib.rs"
+doctest = false
+
+[lints]
+workspace = true
+
+[dependencies]
+base64 = { workspace = true }
+codex-file-system = { workspace = true }
+codex-network-proxy = { workspace = true }
+codex-protocol = { workspace = true }
+codex-shell-command = { workspace = true }
+codex-utils-path-uri = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+
+[dev-dependencies]
+pretty_assertions = { workspace = true }
@@ -0,0 +1,7 @@
+mod process_id;
+mod protocol;
+pub mod rpc;
+
+pub use process_id::ProcessId;
+pub use protocol::*;
+pub use rpc::*;
@@ -4,6 +4,7 @@ use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
 use codex_file_system::FileSystemSandboxContext;
 use codex_network_proxy::ManagedNetworkSandboxContext;
 use codex_protocol::config_types::ShellEnvironmentPolicyInherit;
+use codex_shell_command::shell_detect::DetectedShell;
 use codex_utils_path_uri::PathUri;
 use serde::Deserialize;
 use serde::Serialize;
@@ -22,9 +23,9 @@ pub const EXEC_EXITED_METHOD: &str = "process/exited";
 pub const EXEC_CLOSED_METHOD: &str = "process/closed";
 pub const ENVIRONMENT_INFO_METHOD: &str = "environment/info";
 pub const FS_READ_FILE_METHOD: &str = "fs/readFile";
-pub(crate) const FS_OPEN_METHOD: &str = "fs/open";
-pub(crate) const FS_READ_BLOCK_METHOD: &str = "fs/readBlock";
-pub(crate) const FS_CLOSE_METHOD: &str = "fs/close";
+pub const FS_OPEN_METHOD: &str = "fs/open";
+pub const FS_READ_BLOCK_METHOD: &str = "fs/readBlock";
+pub const FS_CLOSE_METHOD: &str = "fs/close";
 pub const FS_WRITE_FILE_METHOD: &str = "fs/writeFile";
 pub const FS_CREATE_DIRECTORY_METHOD: &str = "fs/createDirectory";
 pub const FS_GET_METADATA_METHOD: &str = "fs/getMetadata";
@@ -77,6 +78,18 @@ pub struct EnvironmentInfo {
    pub cwd: Option<PathUri>,
 }

+impl EnvironmentInfo {
+    /// Returns information about the current local exec-server process.
+    pub fn local() -> Self {
+        Self {
+            shell: codex_shell_command::shell_detect::default_user_shell().into(),
+            cwd: std::env::current_dir()
+                .ok()
+                .and_then(|cwd| PathUri::from_host_native_path(cwd).ok()),
+        }
+    }
+}
+
 /// Shell detected for an execution/filesystem environment.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -88,6 +101,15 @@ pub struct ShellInfo {
    pub path: String,
 }

+impl From<DetectedShell> for ShellInfo {
+    fn from(shell: DetectedShell) -> Self {
+        Self {
+            name: shell.name().to_string(),
+            path: shell.shell_path.to_string_lossy().into_owned(),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ExecParams {
@@ -0,0 +1,78 @@
+//! JSON-RPC wire envelopes used by exec-server.
+//!
+//! Exec-server uses the Codex JSON-RPC dialect, which omits the
+//! `"jsonrpc": "2.0"` field on the wire.
+
+use std::fmt;
+
+use serde::Deserialize;
+use serde::Serialize;
+
+pub const JSONRPC_VERSION: &str = "2.0";
+
+#[derive(Debug, Clone, PartialEq, PartialOrd, Ord, Deserialize, Serialize, Hash, Eq)]
+#[serde(untagged)]
+pub enum RequestId {
+    String(String),
+    Integer(i64),
+}
+
+impl fmt::Display for RequestId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::String(value) => f.write_str(value),
+            Self::Integer(value) => write!(f, "{value}"),
+        }
+    }
+}
+
+pub type Result = serde_json::Value;
+
+/// Any valid exec-server JSON-RPC object that can be decoded from or encoded onto the wire.
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+#[serde(untagged)]
+pub enum JSONRPCMessage {
+    Request(JSONRPCRequest),
+    Notification(JSONRPCNotification),
+    Response(JSONRPCResponse),
+    Error(JSONRPCError),
+}
+
+/// A request that expects a response.
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct JSONRPCRequest {
+    pub id: RequestId,
+    pub method: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub params: Option<serde_json::Value>,
+}
+
+/// A notification that does not expect a response.
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct JSONRPCNotification {
+    pub method: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub params: Option<serde_json::Value>,
+}
+
+/// A successful response to a request.
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct JSONRPCResponse {
+    pub id: RequestId,
+    pub result: Result,
+}
+
+/// A response indicating that a request failed.
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct JSONRPCError {
+    pub error: JSONRPCErrorError,
+    pub id: RequestId,
+}
+
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct JSONRPCErrorError {
+    pub code: i64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub data: Option<serde_json::Value>,
+    pub message: String,
+}
@@ -16,14 +16,13 @@ axum = { workspace = true, features = ["http1", "tokio", "ws"] }
 base64 = { workspace = true }
 bytes = { workspace = true }
 clatter = { workspace = true }
-codex-app-server-protocol = { workspace = true }
 codex-api = { workspace = true }
 codex-client = { workspace = true }
+codex-exec-server-protocol = { workspace = true }
 codex-file-system = { workspace = true }
 codex-network-proxy = { workspace = true }
 codex-protocol = { workspace = true }
 codex-sandboxing = { workspace = true }
-codex-shell-command = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-path-uri = { workspace = true }
 codex-utils-pty = { workspace = true }
@@ -16,8 +16,8 @@ filesystem operations and `codex-linux-sandbox`.

 ## Transport

-The server speaks the shared `codex-app-server-protocol` message envelope on
-the wire.
+The server speaks the exec-specific `codex-exec-server-protocol` message
+envelope on the wire.

 The CLI entrypoint supports:

@@ -9,7 +9,7 @@ use std::sync::atomic::Ordering;
 use std::time::Duration;

 use arc_swap::ArcSwap;
-use codex_app_server_protocol::JSONRPCNotification;
+use codex_exec_server_protocol::JSONRPCNotification;
 use futures::FutureExt;
 use futures::future::BoxFuture;
 use serde_json::Value;
@@ -1187,9 +1187,9 @@ async fn handle_server_notification(

 #[cfg(test)]
 mod tests {
-    use codex_app_server_protocol::JSONRPCMessage;
-    use codex_app_server_protocol::JSONRPCNotification;
-    use codex_app_server_protocol::JSONRPCResponse;
+    use codex_exec_server_protocol::JSONRPCMessage;
+    use codex_exec_server_protocol::JSONRPCNotification;
+    use codex_exec_server_protocol::JSONRPCResponse;
    use futures::SinkExt;
    use futures::StreamExt;
    use pretty_assertions::assert_eq;
@@ -8,8 +8,8 @@
 use std::error::Error as StdError;
 use std::time::Duration;

-use codex_app_server_protocol::JSONRPCErrorError;
 use codex_client::build_reqwest_client_with_custom_ca;
+use codex_exec_server_protocol::JSONRPCErrorError;
 use futures::FutureExt;
 use futures::StreamExt;
 use futures::future::BoxFuture;
@@ -7,7 +7,7 @@ use std::time::Duration;

 use axum::extract::ws::Message as AxumWebSocketMessage;
 use axum::extract::ws::WebSocket as AxumWebSocket;
-use codex_app_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCMessage;
 use futures::Sink;
 use futures::SinkExt;
 use futures::Stream;
@@ -597,8 +597,8 @@ mod tests {
    use std::task::Context;
    use std::task::Poll;

-    use codex_app_server_protocol::JSONRPCRequest;
-    use codex_app_server_protocol::RequestId;
+    use codex_exec_server_protocol::JSONRPCRequest;
+    use codex_exec_server_protocol::RequestId;
    use futures::channel::mpsc as futures_mpsc;
    use futures::task::AtomicWaker;
    use tokio::net::TcpListener;
@@ -667,7 +667,6 @@ mod tests {
            id: RequestId::Integer(1),
            method: "test".to_string(),
            params: None,
-            trace: None,
        });

        server_websocket
@@ -732,7 +731,6 @@ mod tests {
            id: RequestId::Integer(1),
            method: "test".to_string(),
            params: None,
-            trace: None,
        })
    }

@@ -23,12 +23,9 @@ use crate::local_file_system::LocalFileSystem;
 use crate::local_process::LocalProcess;
 use crate::process::ExecBackend;
 use crate::protocol::EnvironmentInfo;
-use crate::protocol::ShellInfo;
 use crate::remote::NoiseRendezvousEnvironmentConfig;
 use crate::remote_file_system::RemoteFileSystem;
 use crate::remote_process::RemoteProcess;
-use codex_shell_command::shell_detect::DetectedShell;
-use codex_utils_path_uri::PathUri;
 use tokio_util::task::AbortOnDropHandle;

 pub const CODEX_EXEC_SERVER_URL_ENV_VAR: &str = "CODEX_EXEC_SERVER_URL";
@@ -605,26 +602,6 @@ impl Environment {
    }
 }

-impl EnvironmentInfo {
-    pub(crate) fn local() -> Self {
-        Self {
-            shell: codex_shell_command::shell_detect::default_user_shell().into(),
-            cwd: std::env::current_dir()
-                .ok()
-                .and_then(|cwd| PathUri::from_host_native_path(cwd).ok()),
-        }
-    }
-}
-
-impl From<DetectedShell> for ShellInfo {
-    fn from(shell: DetectedShell) -> Self {
-        Self {
-            name: shell.name().to_string(),
-            path: shell.shell_path.to_string_lossy().into_owned(),
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use std::collections::HashMap;
@@ -1,6 +1,6 @@
 use base64::Engine as _;
 use base64::engine::general_purpose::STANDARD;
-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;
 use serde::Deserialize;
 use serde::Serialize;
 use tokio::io;
@@ -1,6 +1,6 @@
 use std::collections::HashMap;

-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;
 use codex_protocol::models::PermissionProfile;
 use codex_protocol::permissions::FileSystemAccessMode;
 use codex_protocol::permissions::FileSystemPath;
@@ -15,9 +15,7 @@ mod local_process;
 mod noise_channel;
 mod noise_relay;
 mod process;
-mod process_id;
 mod process_sandbox;
-mod protocol;
 mod regular_file;
 mod relay;
 mod relay_proto;
@@ -29,6 +27,8 @@ mod runtime_paths;
 mod sandboxed_file_system;
 mod server;

+use codex_exec_server_protocol as protocol;
+
 pub use client::ExecServerClient;
 pub use client::ExecServerError;
 pub use client::http_client::HttpResponseBodyStream;
@@ -39,6 +39,7 @@ pub use client_api::NoiseRendezvousConnectArgs;
 pub use client_api::NoiseRendezvousConnectBundle;
 pub use client_api::NoiseRendezvousConnectProvider;
 pub use client_api::RemoteExecServerConnectArgs;
+pub use codex_exec_server_protocol::ProcessId;
 pub use codex_file_system::CopyOptions;
 pub use codex_file_system::CreateDirectoryOptions;
 pub use codex_file_system::ExecutorFileSystem;
@@ -78,7 +79,6 @@ pub use process::ExecProcessEvent;
 pub use process::ExecProcessEventReceiver;
 pub use process::ExecProcessFuture;
 pub use process::StartedExecProcess;
-pub use process_id::ProcessId;
 pub use protocol::ByteChunk;
 pub use protocol::EnvironmentInfo;
 pub use protocol::ExecClosedNotification;
@@ -7,7 +7,7 @@ use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use std::time::Duration;

-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;
 use codex_protocol::config_types::EnvironmentVariablePattern;
 use codex_protocol::config_types::ShellEnvironmentPolicy;
 use codex_protocol::exec_output::ExecToolCallOutput;
@@ -1,9 +1,9 @@
 use std::time::Duration;

 use anyhow::Result;
-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCResponse;
-use codex_app_server_protocol::RequestId;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCResponse;
+use codex_exec_server_protocol::RequestId;
 use tokio::sync::mpsc;
 use tokio::time::timeout;

@@ -1,4 +1,4 @@
-use codex_app_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCMessage;

 use crate::ExecServerError;

@@ -1,5 +1,5 @@
-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCNotification;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCNotification;
 use pretty_assertions::assert_eq;

 use super::JsonRpcMessageDecoder;
@@ -1,6 +1,6 @@
 use std::collections::HashMap;

-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;
 use codex_network_proxy::CUSTOM_CA_ENV_KEYS;
 use codex_network_proxy::is_managed_mitm_ca_trust_bundle_path;
 use codex_protocol::models::PermissionProfile;
@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::time::Duration;

-use codex_app_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCMessage;
 use futures::Sink;
 use futures::SinkExt;
 use futures::Stream;
@@ -847,8 +847,8 @@ mod tests {
    use std::task::Poll;
    use std::time::Duration;

-    use codex_app_server_protocol::JSONRPCRequest;
-    use codex_app_server_protocol::RequestId;
+    use codex_exec_server_protocol::JSONRPCRequest;
+    use codex_exec_server_protocol::RequestId;
    use futures::Sink;
    use futures::Stream;
    use futures::channel::mpsc as futures_mpsc;
@@ -1101,7 +1101,6 @@ mod tests {
            id: RequestId::Integer(1),
            method: "test".to_string(),
            params: None,
-            trace: None,
        })
    }

@@ -1,7 +1,7 @@
 #![allow(clippy::expect_used)]

-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCResponse;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCResponse;
 use codex_protocol::models::PermissionProfile;
 use codex_protocol::permissions::FileSystemAccessMode;
 use codex_protocol::permissions::FileSystemPath;
@@ -6,13 +6,13 @@ use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicI64;
 use std::sync::atomic::Ordering;

-use codex_app_server_protocol::JSONRPCError;
-use codex_app_server_protocol::JSONRPCErrorError;
-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCNotification;
-use codex_app_server_protocol::JSONRPCRequest;
-use codex_app_server_protocol::JSONRPCResponse;
-use codex_app_server_protocol::RequestId;
+use codex_exec_server_protocol::JSONRPCError;
+use codex_exec_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCNotification;
+use codex_exec_server_protocol::JSONRPCRequest;
+use codex_exec_server_protocol::JSONRPCResponse;
+use codex_exec_server_protocol::RequestId;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use serde_json::Value;
@@ -362,7 +362,6 @@ impl RpcClient {
                id: request_id.clone(),
                method: method.to_string(),
                params: Some(params),
-                trace: None,
            }))
            .await
            .is_err()
@@ -552,8 +551,8 @@ async fn drain_pending(pending: &Mutex<HashMap<RequestId, PendingRequest>>) {
 mod tests {
    use std::time::Duration;

-    use codex_app_server_protocol::JSONRPCMessage;
-    use codex_app_server_protocol::JSONRPCResponse;
+    use codex_exec_server_protocol::JSONRPCMessage;
+    use codex_exec_server_protocol::JSONRPCResponse;
    use pretty_assertions::assert_eq;
    use tokio::io::AsyncBufReadExt;
    use tokio::io::AsyncWriteExt;
@@ -1,6 +1,6 @@
 use base64::Engine as _;
 use base64::engine::general_purpose::STANDARD;
-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;
 use codex_utils_path_uri::PathUri;
 use tokio::io;

@@ -2,7 +2,7 @@ use std::io;

 use base64::Engine as _;
 use base64::engine::general_purpose::STANDARD;
-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;

 use crate::CopyOptions;
 use crate::CreateDirectoryOptions;
@@ -3,8 +3,8 @@ use std::sync::Mutex as StdMutex;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;

-use codex_app_server_protocol::JSONRPCErrorError;
-use codex_app_server_protocol::RequestId;
+use codex_exec_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::RequestId;
 use serde_json::to_value;
 use std::collections::HashSet;
 use tokio::sync::Mutex;
@@ -1,4 +1,4 @@
-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;

 use crate::ExecServerRuntimePaths;
 use crate::local_process::LocalProcess;
@@ -89,7 +89,7 @@ async fn run_connection(
                warn!("ignoring malformed exec-server message: {reason}");
                if outgoing_tx
                    .send(RpcServerOutboundMessage::Error {
-                        request_id: codex_app_server_protocol::RequestId::Integer(-1),
+                        request_id: codex_exec_server_protocol::RequestId::Integer(-1),
                        error: invalid_request(reason),
                    })
                    .await
@@ -99,7 +99,7 @@ async fn run_connection(
                }
            }
            JsonRpcConnectionEvent::Message(message) => match message {
-                codex_app_server_protocol::JSONRPCMessage::Request(request) => {
+                codex_exec_server_protocol::JSONRPCMessage::Request(request) => {
                    if let Some(route) = router.request_route(request.method.as_str()) {
                        let message = tokio::select! {
                            message = route(Arc::clone(&handler), request) => message,
@@ -127,7 +127,7 @@ async fn run_connection(
                        break;
                    }
                }
-                codex_app_server_protocol::JSONRPCMessage::Notification(notification) => {
+                codex_exec_server_protocol::JSONRPCMessage::Notification(notification) => {
                    let Some(route) = router.notification_route(notification.method.as_str())
                    else {
                        warn!(
@@ -150,14 +150,14 @@ async fn run_connection(
                        break;
                    }
                }
-                codex_app_server_protocol::JSONRPCMessage::Response(response) => {
+                codex_exec_server_protocol::JSONRPCMessage::Response(response) => {
                    warn!(
                        "closing exec-server connection after unexpected client response: {:?}",
                        response.id
                    );
                    break;
                }
-                codex_app_server_protocol::JSONRPCMessage::Error(error) => {
+                codex_exec_server_protocol::JSONRPCMessage::Error(error) => {
                    warn!(
                        "closing exec-server connection after unexpected client error: {:?}",
                        error.id
@@ -190,11 +190,11 @@ mod tests {
    use std::sync::Arc;
    use std::time::Duration;

-    use codex_app_server_protocol::JSONRPCMessage;
-    use codex_app_server_protocol::JSONRPCNotification;
-    use codex_app_server_protocol::JSONRPCRequest;
-    use codex_app_server_protocol::JSONRPCResponse;
-    use codex_app_server_protocol::RequestId;
+    use codex_exec_server_protocol::JSONRPCMessage;
+    use codex_exec_server_protocol::JSONRPCNotification;
+    use codex_exec_server_protocol::JSONRPCRequest;
+    use codex_exec_server_protocol::JSONRPCResponse;
+    use codex_exec_server_protocol::RequestId;
    use codex_utils_path_uri::PathUri;
    use pretty_assertions::assert_eq;
    use serde::Serialize;
@@ -382,7 +382,6 @@ mod tests {
                id: RequestId::Integer(id),
                method: method.to_string(),
                params: Some(serde_json::to_value(params).expect("serialize params")),
-                trace: None,
            }),
        )
        .await;
@@ -3,7 +3,7 @@ use std::sync::Arc;
 use std::sync::Mutex as StdMutex;
 use std::time::Duration;

-use codex_app_server_protocol::JSONRPCErrorError;
+use codex_exec_server_protocol::JSONRPCErrorError;
 use tokio::sync::Mutex;
 use uuid::Uuid;

@@ -1,11 +1,11 @@
 use std::net::SocketAddr;
 use std::time::Duration;

-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCNotification;
-use codex_app_server_protocol::JSONRPCRequest;
-use codex_app_server_protocol::JSONRPCResponse;
-use codex_app_server_protocol::RequestId;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCNotification;
+use codex_exec_server_protocol::JSONRPCRequest;
+use codex_exec_server_protocol::JSONRPCResponse;
+use codex_exec_server_protocol::RequestId;
 use pretty_assertions::assert_eq;
 use tokio::io::AsyncBufReadExt;
 use tokio::io::AsyncWriteExt;
@@ -74,7 +74,6 @@ async fn stdio_listen_transport_serves_initialize() {
            })
            .expect("initialize params should serialize"),
        ),
-        trace: None,
    });
    write_jsonrpc_line(&mut client_writer, &initialize).await;

@@ -5,10 +5,10 @@ use std::process::Stdio;
 use std::time::Duration;

 use anyhow::anyhow;
-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCNotification;
-use codex_app_server_protocol::JSONRPCRequest;
-use codex_app_server_protocol::RequestId;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCNotification;
+use codex_exec_server_protocol::JSONRPCRequest;
+use codex_exec_server_protocol::RequestId;
 use futures::SinkExt;
 use futures::StreamExt;
 use tempfile::TempDir;
@@ -165,7 +165,6 @@ impl ExecServerHarness {
            id: id.clone(),
            method: method.to_string(),
            params: Some(params),
-            trace: None,
        }))
        .await?;
        Ok(id)
@@ -4,11 +4,6 @@ use std::time::Duration;
 use anyhow::Context;
 use anyhow::Result;
 use anyhow::bail;
-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCNotification;
-use codex_app_server_protocol::JSONRPCRequest;
-use codex_app_server_protocol::JSONRPCResponse;
-use codex_app_server_protocol::RequestId;
 use codex_exec_server::ExecServerClient;
 use codex_exec_server::HttpHeader;
 use codex_exec_server::HttpRequestBodyDeltaNotification;
@@ -17,6 +12,11 @@ use codex_exec_server::HttpRequestResponse;
 use codex_exec_server::InitializeParams;
 use codex_exec_server::InitializeResponse;
 use codex_exec_server::RemoteExecServerConnectArgs;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCNotification;
+use codex_exec_server_protocol::JSONRPCRequest;
+use codex_exec_server_protocol::JSONRPCResponse;
+use codex_exec_server_protocol::RequestId;
 use futures::SinkExt;
 use futures::StreamExt;
 use pretty_assertions::assert_eq;
@@ -6,16 +6,16 @@ use std::collections::BTreeMap;
 use std::io::ErrorKind;
 use std::time::Duration;

-use codex_app_server_protocol::JSONRPCError;
-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCNotification;
-use codex_app_server_protocol::JSONRPCResponse;
-use codex_app_server_protocol::RequestId;
 use codex_exec_server::HttpHeader;
 use codex_exec_server::HttpRequestBodyDeltaNotification;
 use codex_exec_server::HttpRequestParams;
 use codex_exec_server::HttpRequestResponse;
 use codex_exec_server::InitializeParams;
+use codex_exec_server_protocol::JSONRPCError;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCNotification;
+use codex_exec_server_protocol::JSONRPCResponse;
+use codex_exec_server_protocol::RequestId;
 use common::exec_server::ExecServerHarness;
 use common::exec_server::exec_server;
 use pretty_assertions::assert_eq;
@@ -392,7 +392,7 @@ where
 async fn wait_for_error_response(
    server: &mut ExecServerHarness,
    request_id: RequestId,
-) -> anyhow::Result<codex_app_server_protocol::JSONRPCErrorError> {
+) -> anyhow::Result<codex_exec_server_protocol::JSONRPCErrorError> {
    let response = server
        .wait_for_event(|event| {
            matches!(
@@ -2,10 +2,10 @@

 mod common;

-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCResponse;
 use codex_exec_server::InitializeParams;
 use codex_exec_server::InitializeResponse;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCResponse;
 use common::exec_server::exec_server;
 use pretty_assertions::assert_eq;
 use uuid::Uuid;
@@ -2,8 +2,6 @@

 mod common;

-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCResponse;
 use codex_exec_server::ExecResponse;
 use codex_exec_server::InitializeParams;
 use codex_exec_server::InitializeResponse;
@@ -12,6 +10,8 @@ use codex_exec_server::ReadResponse;
 use codex_exec_server::TerminateResponse;
 use codex_exec_server::WriteResponse;
 use codex_exec_server::WriteStatus;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCResponse;
 use codex_utils_path_uri::PathUri;
 use common::exec_server::exec_server;
 use pretty_assertions::assert_eq;
@@ -2,11 +2,11 @@

 mod common;

-use codex_app_server_protocol::JSONRPCError;
-use codex_app_server_protocol::JSONRPCMessage;
-use codex_app_server_protocol::JSONRPCResponse;
 use codex_exec_server::InitializeParams;
 use codex_exec_server::InitializeResponse;
+use codex_exec_server_protocol::JSONRPCError;
+use codex_exec_server_protocol::JSONRPCMessage;
+use codex_exec_server_protocol::JSONRPCResponse;
 use common::exec_server::exec_server;
 use pretty_assertions::assert_eq;
 use tokio_tungstenite::connect_async;
@@ -28,7 +28,7 @@ async fn exec_server_reports_malformed_websocket_json_and_keeps_running() -> any
    let JSONRPCMessage::Error(JSONRPCError { id, error }) = response else {
        panic!("expected malformed-message error response");
    };
-    assert_eq!(id, codex_app_server_protocol::RequestId::Integer(-1));
+    assert_eq!(id, codex_exec_server_protocol::RequestId::Integer(-1));
    assert_eq!(error.code, -32600);
    assert!(
        error
@@ -70,15 +70,14 @@ async fn exec_server_reports_malformed_websocket_json_and_keeps_running() -> any
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn exec_server_accepts_binary_websocket_json() -> anyhow::Result<()> {
    let mut server = exec_server().await?;
-    let initialize_id = codex_app_server_protocol::RequestId::Integer(1);
-    let initialize = JSONRPCMessage::Request(codex_app_server_protocol::JSONRPCRequest {
+    let initialize_id = codex_exec_server_protocol::RequestId::Integer(1);
+    let initialize = JSONRPCMessage::Request(codex_exec_server_protocol::JSONRPCRequest {
        id: initialize_id.clone(),
        method: "initialize".to_string(),
        params: Some(serde_json::to_value(InitializeParams {
            client_name: "exec-server-binary-test".to_string(),
            resume_session_id: None,
        })?),
-        trace: None,
    });
    server
        .send_raw_binary(serde_json::to_vec(&initialize)?)
@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 #![allow(clippy::expect_used)]

 use std::sync::Arc;
@@ -20,7 +20,6 @@ codex-extension-api = { workspace = true }
 codex-mcp = { workspace = true }
 codex-protocol = { workspace = true }
 codex-tools = { workspace = true }
-codex-utils-absolute-path = { workspace = true }
 codex-utils-path-uri = { workspace = true }
 codex-utils-string = { workspace = true }
 schemars = { workspace = true }
@@ -31,5 +30,6 @@ tracing = { workspace = true }
 url = { workspace = true }

 [dev-dependencies]
+codex-utils-absolute-path = { workspace = true }
 pretty_assertions = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
@@ -1,5 +1,5 @@
 use codex_core_skills::model::SkillDependencies;
-use codex_utils_absolute_path::AbsolutePathBuf;
+use codex_utils_path_uri::PathUri;

 /// Source authority that owns a skill package and must be used to read it.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -76,7 +76,7 @@ impl SkillResourceId {
    pub fn environment(
        id: impl Into<String>,
        environment_id: impl Into<String>,
-        path: AbsolutePathBuf,
+        path: PathUri,
    ) -> Self {
        Self {
            id: id.into(),
@@ -91,7 +91,7 @@ impl SkillResourceId {
        &self.id
    }

-    pub(crate) fn environment_path(&self) -> Option<(&str, &AbsolutePathBuf)> {
+    pub(crate) fn environment_path(&self) -> Option<(&str, &PathUri)> {
        self.environment_path
            .as_ref()
            .map(|resource| (resource.environment_id.as_str(), &resource.path))
@@ -101,7 +101,7 @@ impl SkillResourceId {
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 struct EnvironmentSkillResource {
    environment_id: String,
-    path: AbsolutePathBuf,
+    path: PathUri,
 }

 /// Metadata shown in the always-visible skills catalog.
@@ -1,15 +1,11 @@
 use std::sync::Arc;

-use codex_core_skills::SkillMetadata;
-use codex_core_skills::filter_skill_load_outcome_for_product;
-use codex_core_skills::loader::SkillRoot;
-use codex_core_skills::loader::load_skills_from_roots;
+use codex_core_skills::loader::EnvironmentSkillMetadata;
+use codex_core_skills::loader::load_environment_skills_from_root;
 use codex_exec_server::EnvironmentManager;
 use codex_protocol::capabilities::CapabilityRootLocation;
 use codex_protocol::protocol::Product;
-use codex_protocol::protocol::SkillScope;
-use codex_utils_absolute_path::AbsolutePathBuf;
-use codex_utils_path_uri::PathUri;
+use codex_utils_path_uri::PathConvention;

 use crate::catalog::SkillAuthority;
 use crate::catalog::SkillCatalog;
@@ -64,42 +60,17 @@ impl SkillProvider for ExecutorSkillProvider {
                    ));
                    continue;
                };
-                let root_path = match executor_absolute_path(&path) {
-                    Ok(root_path) => root_path,
-                    Err(err) => {
-                        catalog.warnings.push(format!(
-                            "Selected capability root `{selected_root_id}` has invalid path `{path}`: {err}"
-                        ));
-                        continue;
-                    }
-                };
                let file_system = environment.get_filesystem();
-                let outcome = filter_skill_load_outcome_for_product(
-                    load_skills_from_roots(
-                        [SkillRoot {
-                            path: root_path.clone(),
-                            scope: SkillScope::User,
-                            file_system: Arc::clone(&file_system),
-                            plugin_id: None,
-                            plugin_namespace: None,
-                            plugin_root: None,
-                        }],
-                        /*plugin_skill_snapshots*/ None,
-                    )
-                    .await,
+                let outcome = load_environment_skills_from_root(
+                    file_system.as_ref(),
+                    &path,
                    self.restriction_product,
-                );
-                catalog.warnings.extend(outcome.errors.iter().map(|err| {
-                    format!(
-                        "Failed to load executor skill at {}: {}",
-                        err.path.display(),
-                        err.message
-                    )
-                }));
-                for (skill, enabled) in outcome.skills_with_enabled() {
+                )
+                .await;
+                catalog.warnings.extend(outcome.warnings);
+                for skill in outcome.skills {
                    catalog.push_entry(catalog_entry_from_skill(
-                        skill,
-                        enabled,
+                        &skill,
                        authority.clone(),
                        &selected_root_id,
                        &environment_id,
@@ -134,10 +105,9 @@ impl SkillProvider for ExecutorSkillProvider {
                    "executor skill resource references unavailable environment `{environment_id}`"
                )));
            };
-            let resource_path = PathUri::from_abs_path(resource_path);
            let contents = environment
                .get_filesystem()
-                .read_file_text(&resource_path, /*sandbox*/ None)
+                .read_file_text(resource_path, /*sandbox*/ None)
                .await
                .map_err(|err| {
                    SkillProviderError::new(format!(
@@ -159,19 +129,21 @@ impl SkillProvider for ExecutorSkillProvider {
 }

 fn catalog_entry_from_skill(
-    skill: &SkillMetadata,
-    enabled: bool,
+    skill: &EnvironmentSkillMetadata,
    authority: SkillAuthority,
    selected_root_id: &str,
    environment_id: &str,
 ) -> SkillCatalogEntry {
-    let skill_path = skill.path_to_skills_md.to_string_lossy().into_owned();
-    let normalized_path = skill_path.replace('\\', "/");
+    let skill_path = skill.path_to_skills_md.inferred_native_path_string();
+    let normalized_path = match skill.path_to_skills_md.infer_path_convention() {
+        Some(PathConvention::Windows) => skill_path.replace('\\', "/"),
+        Some(PathConvention::Posix) | None => skill_path,
+    };
    let display_path = format!(
        "skill://{selected_root_id}/{}",
        normalized_path.trim_start_matches('/')
    );
-    let mut entry = SkillCatalogEntry::new(
+    let entry = SkillCatalogEntry::new(
        SkillPackageId(display_path.clone()),
        authority,
        skill.name.clone(),
@@ -186,16 +158,9 @@ fn catalog_entry_from_skill(
    .with_display_path(display_path)
    .with_dependencies(skill.dependencies.clone());

-    if !enabled {
-        entry = entry.disabled();
+    if skill.allows_implicit_invocation() {
+        entry
+    } else {
+        entry.hidden_from_prompt()
    }
-    if !skill.allows_implicit_invocation() {
-        entry = entry.hidden_from_prompt();
-    }
-
-    entry
-}
-
-fn executor_absolute_path(path: &PathUri) -> std::io::Result<AbsolutePathBuf> {
-    path.to_abs_path()
 }
@@ -28,40 +28,48 @@ use pretty_assertions::assert_eq;

 const SKILL_CONTENTS: &str =
    "---\nname: synthetic\ndescription: Synthetic executor skill.\n---\n\nEXECUTOR_ONLY_BODY\n";
+const PLUGIN_MANIFEST: &str = r#"{"name":"synthetic-plugin"}"#;
 static NEXT_TEST_ROOT_ID: AtomicUsize = AtomicUsize::new(0);

 struct SyntheticFileSystem {
-    alias_root: AbsolutePathBuf,
-    canonical_root: AbsolutePathBuf,
+    alias_root: PathUri,
+    canonical_root: PathUri,
+    has_plugin_manifest: bool,
 }

 impl SyntheticFileSystem {
+    fn path(&self, relative_path: &str) -> io::Result<PathUri> {
+        self.canonical_root
+            .join(relative_path)
+            .map_err(io::Error::other)
+    }
+
    async fn canonicalize(&self, path: &PathUri) -> io::Result<PathUri> {
-        let path = path.to_abs_path()?;
-        if path == self.alias_root {
-            return Ok(PathUri::from_abs_path(&self.canonical_root));
+        if path == &self.alias_root {
+            return Ok(self.canonical_root.clone());
        }
-        self.metadata(&path)?;
-        Ok(PathUri::from_abs_path(&path))
+        self.metadata(path)?;
+        Ok(path.clone())
    }

    async fn read_file(&self, path: &PathUri) -> io::Result<Vec<u8>> {
-        if path.to_abs_path()? == self.canonical_root.join("skill/SKILL.md") {
+        if path == &self.path("skill/SKILL.md")? {
            Ok(SKILL_CONTENTS.as_bytes().to_vec())
+        } else if self.has_plugin_manifest && path == &self.path(".claude-plugin/plugin.json")? {
+            Ok(PLUGIN_MANIFEST.as_bytes().to_vec())
        } else {
            Err(io::Error::new(io::ErrorKind::NotFound, "not found"))
        }
    }

    async fn read_directory(&self, path: &PathUri) -> io::Result<Vec<ReadDirectoryEntry>> {
-        let path = path.to_abs_path()?;
-        if path == self.canonical_root {
+        if path == &self.canonical_root {
            Ok(vec![ReadDirectoryEntry {
                file_name: "skill".to_string(),
                is_directory: true,
                is_file: false,
            }])
-        } else if path == self.canonical_root.join("skill") {
+        } else if path == &self.path("skill")? {
            Ok(vec![ReadDirectoryEntry {
                file_name: "SKILL.md".to_string(),
                is_directory: false,
@@ -72,12 +80,13 @@ impl SyntheticFileSystem {
        }
    }

-    fn metadata(&self, path: &AbsolutePathBuf) -> io::Result<FileMetadata> {
-        let skill_dir = self.canonical_root.join("skill");
-        let skill_path = skill_dir.join("SKILL.md");
+    fn metadata(&self, path: &PathUri) -> io::Result<FileMetadata> {
+        let skill_dir = self.path("skill")?;
+        let skill_path = self.path("skill/SKILL.md")?;
+        let manifest_path = self.path(".claude-plugin/plugin.json")?;
        let (is_directory, is_file) = if path == &self.canonical_root || path == &skill_dir {
            (true, false)
-        } else if path == &skill_path {
+        } else if path == &skill_path || self.has_plugin_manifest && path == &manifest_path {
            (false, true)
        } else {
            return Err(io::Error::new(io::ErrorKind::NotFound, "not found"));
@@ -146,7 +155,7 @@ impl ExecutorFileSystem for SyntheticFileSystem {
        path: &'a PathUri,
        _sandbox: Option<&'a FileSystemSandboxContext>,
    ) -> ExecutorFileSystemFuture<'a, FileMetadata> {
-        Box::pin(async move { self.metadata(&path.to_abs_path()?) })
+        Box::pin(async move { self.metadata(path) })
    }

    fn read_directory<'a>(
@@ -193,8 +202,9 @@ async fn skill_loading_and_reads_use_the_supplied_executor_file_system() {
            path: alias_root.clone(),
            scope: SkillScope::User,
            file_system: Arc::new(SyntheticFileSystem {
-                alias_root,
-                canonical_root: canonical_root.clone(),
+                alias_root: PathUri::from_abs_path(&alias_root),
+                canonical_root: PathUri::from_abs_path(&canonical_root),
+                has_plugin_manifest: false,
            }),
            plugin_id: None,
            plugin_namespace: None,
@@ -221,13 +231,18 @@ async fn skill_loading_and_reads_use_the_supplied_executor_file_system() {

 #[tokio::test]
 async fn selected_root_id_distinguishes_identical_executor_paths() {
-    let test_root = create_local_skill_root("root-identity").expect("create local skill root");
-    let canonical_root = AbsolutePathBuf::from_absolute_path_checked(&test_root)
-        .expect("absolute skill root")
-        .canonicalize()
-        .expect("canonicalize skill root")
-        .to_string_lossy()
-        .replace('\\', "/");
+    let root_label = if cfg!(unix) {
+        r"root\identity"
+    } else {
+        "root-identity"
+    };
+    let test_root = create_local_skill_root(root_label).expect("create local skill root");
+    let selected_root = test_root.to_string_lossy().into_owned();
+    let selected_root = if cfg!(windows) {
+        selected_root.replace('\\', "/")
+    } else {
+        selected_root
+    };
    let provider = ExecutorSkillProvider::new_with_restriction_product(
        Arc::new(EnvironmentManager::default_for_tests()),
        /*restriction_product*/ None,
@@ -268,14 +283,14 @@ async fn selected_root_id_distinguishes_identical_executor_paths() {
                "root-a".to_string(),
                format!(
                    "skill://root-a/{}/skill/SKILL.md",
-                    canonical_root.trim_start_matches('/')
+                    selected_root.trim_start_matches('/')
                ),
            ),
            (
                "root-b".to_string(),
                format!(
                    "skill://root-b/{}/skill/SKILL.md",
-                    canonical_root.trim_start_matches('/')
+                    selected_root.trim_start_matches('/')
                ),
            ),
        ]
@@ -69,6 +69,9 @@ pub enum CodexErr {
    #[error("turn aborted. Something went wrong? Hit `/feedback` to report the issue.")]
    TurnAborted,

+    #[error("shared rollout token budget exhausted")]
+    RolloutBudgetExceeded,
+
    /// Returned by ResponsesClient when the SSE stream disconnects or errors out **after** the HTTP
    /// handshake has succeeded but **before** it finished emitting `response.completed`.
    ///
@@ -173,6 +176,7 @@ impl CodexErr {
    pub fn is_retryable(&self) -> bool {
        match self {
            CodexErr::TurnAborted
+            | CodexErr::RolloutBudgetExceeded
            | CodexErr::Interrupted
            | CodexErr::EnvVar(_)
            | CodexErr::Fatal(_)
@@ -220,6 +224,7 @@ impl CodexErr {
    pub fn to_codex_protocol_error(&self) -> CodexErrorInfo {
        match self {
            CodexErr::ContextWindowExceeded => CodexErrorInfo::ContextWindowExceeded,
+            CodexErr::RolloutBudgetExceeded => CodexErrorInfo::RolloutBudgetExceeded,
            CodexErr::UsageLimitReached(_)
            | CodexErr::QuotaExceeded
            | CodexErr::UsageNotIncluded => CodexErrorInfo::UsageLimitExceeded,
@@ -1690,6 +1690,7 @@ pub enum NonSteerableTurnKind {
 #[ts(rename_all = "snake_case")]
 pub enum CodexErrorInfo {
    ContextWindowExceeded,
+    RolloutBudgetExceeded,
    UsageLimitExceeded,
    ServerOverloaded,
    CyberPolicy,
@@ -1727,6 +1728,7 @@ impl CodexErrorInfo {
        match self {
            Self::ThreadRollbackFailed | Self::ActiveTurnNotSteerable { .. } => false,
            Self::ContextWindowExceeded
+            | Self::RolloutBudgetExceeded
            | Self::UsageLimitExceeded
            | Self::ServerOverloaded
            | Self::CyberPolicy
@@ -10,6 +10,7 @@ pub mod plugin_namespace;
 pub use plugin_namespace::DISCOVERABLE_PLUGIN_MANIFEST_PATHS;
 pub use plugin_namespace::find_plugin_manifest_path;
 pub use plugin_namespace::plugin_namespace_for_skill_path;
+pub use plugin_namespace::plugin_namespace_for_skill_uri;

 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct PluginSkillRoot {
@@ -26,13 +26,12 @@ struct RawPluginManifestName {

 async fn plugin_manifest_name(
    fs: &dyn ExecutorFileSystem,
-    plugin_root: &AbsolutePathBuf,
+    plugin_root: &PathUri,
 ) -> Option<String> {
    let mut manifest_path = None;
    for relative_path in DISCOVERABLE_PLUGIN_MANIFEST_PATHS {
-        let candidate = plugin_root.join(relative_path);
-        let candidate_uri = PathUri::from_abs_path(&candidate);
-        match fs.get_metadata(&candidate_uri, /*sandbox*/ None).await {
+        let candidate = plugin_root.join(relative_path).ok()?;
+        match fs.get_metadata(&candidate, /*sandbox*/ None).await {
            Ok(metadata) if metadata.is_file => {
                manifest_path = Some(candidate);
                break;
@@ -40,20 +39,16 @@ async fn plugin_manifest_name(
            Ok(_) | Err(_) => {}
        }
    }
-    let manifest_path = manifest_path?;
-    let manifest_path_uri = PathUri::from_abs_path(&manifest_path);
    let contents = fs
-        .read_file_text(&manifest_path_uri, /*sandbox*/ None)
+        .read_file_text(&manifest_path?, /*sandbox*/ None)
        .await
        .ok()?;
    let RawPluginManifestName { name: raw_name } = serde_json::from_str(&contents).ok()?;
    Some(
        plugin_root
-            .file_name()
-            .and_then(|entry| entry.to_str())
+            .basename()
            .filter(|_| raw_name.trim().is_empty())
-            .unwrap_or(raw_name.as_str())
-            .to_string(),
+            .unwrap_or(raw_name),
    )
 }

@@ -63,10 +58,20 @@ pub async fn plugin_namespace_for_skill_path(
    fs: &dyn ExecutorFileSystem,
    path: &AbsolutePathBuf,
 ) -> Option<String> {
-    for ancestor in path.ancestors() {
-        if let Some(name) = plugin_manifest_name(fs, &ancestor).await {
+    plugin_namespace_for_skill_uri(fs, &PathUri::from_abs_path(path)).await
+}
+
+/// Returns the plugin manifest `name` for the nearest URI ancestor of `path`.
+pub async fn plugin_namespace_for_skill_uri(
+    fs: &dyn ExecutorFileSystem,
+    path: &PathUri,
+) -> Option<String> {
+    let mut ancestor = Some(path.clone());
+    while let Some(path) = ancestor {
+        if let Some(name) = plugin_manifest_name(fs, &path).await {
            return Some(name);
        }
+        ancestor = path.parent();
    }
    None
 }