diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 48674d3968..41a1e69229 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -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 diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index a2953be530..6b5b5afdf6 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 17c2d1561b..2bc8ac3400 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -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" } diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index a8d1d6d6b0..df619a96e9 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -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, diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index dcdbfa53f1..214f136d7e 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -6834,4 +6834,4 @@ } ], "title": "ClientRequest" -} +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index ea6076fa89..fc22192cc5 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -573,6 +573,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 3cecc0e116..e61969c47c 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -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" -} +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index b94593a009..3809c0b44a 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -3310,6 +3310,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", @@ -18468,4 +18469,4 @@ }, "title": "CodexAppServerProtocolV2", "type": "object" -} +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json index fd55d08764..9f54bb0efa 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json @@ -7,6 +7,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json index d87dd74401..c87350ff90 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json @@ -30,6 +30,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json index e735aaa89b..ccbec39499 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json @@ -111,6 +111,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json index a428f109db..63972a7c78 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json @@ -33,6 +33,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json index 2731ab409a..38996de6a2 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json @@ -33,6 +33,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json index c591beea3c..b3fe534d2b 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json @@ -33,6 +33,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json index 120f972e51..4fcc9d5e88 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json @@ -111,6 +111,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json index 409a1ca5f5..5877ee0a70 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json @@ -33,6 +33,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json index e447c644d7..50c0a9cf7f 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json @@ -388,4 +388,4 @@ }, "title": "ThreadStartParams", "type": "object" -} +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json index 63efc4e3d6..6f72c42df3 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json @@ -111,6 +111,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json index 152d83a150..b7b1941d14 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json @@ -33,6 +33,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json index 291b88a923..c2db095ee7 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json @@ -33,6 +33,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json index 566922e80c..9c401d18a1 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json @@ -30,6 +30,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json index 627aea2126..1dc27a7c35 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json @@ -30,6 +30,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json index cacfb01564..980260a49e 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json @@ -30,6 +30,7 @@ { "enum": [ "contextWindowExceeded", + "rolloutBudgetExceeded", "usageLimitExceeded", "serverOverloaded", "cyberPolicy", diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/CodexErrorInfo.ts b/codex-rs/app-server-protocol/schema/typescript/v2/CodexErrorInfo.ts index 6e975abf41..0eea30deed 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/CodexErrorInfo.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/CodexErrorInfo.ts @@ -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"; diff --git a/codex-rs/app-server-protocol/src/lib.rs b/codex-rs/app-server-protocol/src/lib.rs index f6d7670e10..0797fcdf7e 100644 --- a/codex-rs/app-server-protocol/src/lib.rs +++ b/codex-rs/app-server-protocol/src/lib.rs @@ -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; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/shared.rs b/codex-rs/app-server-protocol/src/protocol/v2/shared.rs index 94e650d923..00e7c85a05 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/shared.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/shared.rs @@ -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 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, diff --git a/codex-rs/app-server-protocol/src/jsonrpc_lite.rs b/codex-rs/app-server-protocol/src/rpc.rs similarity index 100% rename from codex-rs/app-server-protocol/src/jsonrpc_lite.rs rename to codex-rs/app-server-protocol/src/rpc.rs diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index d9357313b2..4317b6a820 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -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 diff --git a/codex-rs/code-mode-protocol/src/host/error.rs b/codex-rs/code-mode-protocol/src/host/error.rs new file mode 100644 index 0000000000..423202e44b --- /dev/null +++ b/codex-rs/code-mode-protocol/src/host/error.rs @@ -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 }, +} diff --git a/codex-rs/code-mode-protocol/src/host/host_tests.rs b/codex-rs/code-mode-protocol/src/host/host_tests.rs new file mode 100644 index 0000000000..2d64993440 --- /dev/null +++ b/codex-rs/code-mode-protocol/src/host/host_tests.rs @@ -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::(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::(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::(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::(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::(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![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::(invalid).is_err()); + } +} + +#[test] +fn unknown_fields_are_rejected() { + assert!( + serde_json::from_value::(json!({ + "type": "session/open", + "sessionId": "session-1", + "unexpected": true, + })) + .is_err() + ); + assert!( + serde_json::from_value::(json!({ + "type": "session/ready", + "sessionId": "session-1", + "unexpected": true, + })) + .is_err() + ); +} diff --git a/codex-rs/code-mode-protocol/src/host/message.rs b/codex-rs/code-mode-protocol/src/host/message.rs new file mode 100644 index 0000000000..ed30ac8f17 --- /dev/null +++ b/codex-rs/code-mode-protocol/src/host/message.rs @@ -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 { + 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(deserializer: D) -> Result + 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 }, +} diff --git a/codex-rs/code-mode-protocol/src/host/mod.rs b/codex-rs/code-mode-protocol/src/host/mod.rs new file mode 100644 index 0000000000..f705f80e36 --- /dev/null +++ b/codex-rs/code-mode-protocol/src/host/mod.rs @@ -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; diff --git a/codex-rs/code-mode-protocol/src/host/types.rs b/codex-rs/code-mode-protocol/src/host/types.rs new file mode 100644 index 0000000000..69c8fd0881 --- /dev/null +++ b/codex-rs/code-mode-protocol/src/host/types.rs @@ -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 { + 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) -> Result { + 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(&self, serializer: S) -> Result + where + S: Serializer, + { + self.0.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for NonEmptyString { + fn deserialize(deserializer: D) -> Result + 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) -> Result { + 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) -> Result { + 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); + +impl CapabilitySet { + pub fn empty() -> Self { + Self::default() + } + + pub fn try_new( + capabilities: impl IntoIterator, + ) -> Result { + 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 { + self.0.iter() + } +} + +impl<'de> Deserialize<'de> for CapabilitySet { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + Self::try_new(Vec::::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); + +impl SupportedProtocolVersions { + pub fn try_new( + versions: impl IntoIterator, + ) -> Result { + 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 + '_ { + self.0.iter().copied() + } +} + +impl<'de> Deserialize<'de> for SupportedProtocolVersions { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + Self::try_new(Vec::::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 {} diff --git a/codex-rs/code-mode-protocol/src/lib.rs b/codex-rs/code-mode-protocol/src/lib.rs index b18fff4b6c..bb47b1a4c4 100644 --- a/codex-rs/code-mode-protocol/src/lib.rs +++ b/codex-rs/code-mode-protocol/src/lib.rs @@ -1,4 +1,5 @@ mod description; +pub mod host; mod response; mod runtime; mod session; diff --git a/codex-rs/code-mode-protocol/src/session.rs b/codex-rs/code-mode-protocol/src/session.rs index 2ebbe6e8f1..57669c4314 100644 --- a/codex-rs/code-mode-protocol/src/session.rs +++ b/codex-rs/code-mode-protocol/src/session.rs @@ -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, diff --git a/codex-rs/code-mode/src/service.rs b/codex-rs/code-mode/src/service.rs index 5c459081a6..57400001f0 100644 --- a/codex-rs/code-mode/src/service.rs +++ b/codex-rs/code-mode/src/service.rs @@ -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, diff --git a/codex-rs/code-mode/src/session_runtime/mod.rs b/codex-rs/code-mode/src/session_runtime/mod.rs index 0f7557a750..5688c04303 100644 --- a/codex-rs/code-mode/src/session_runtime/mod.rs +++ b/codex-rs/code-mode/src/session_runtime/mod.rs @@ -63,10 +63,6 @@ impl SessionRuntime { } } - pub(crate) fn is_alive(&self) -> bool { - !self.inner.shutdown_token.is_cancelled() - } - pub(crate) async fn execute( &self, request: CreateCellRequest, diff --git a/codex-rs/code-mode/src/session_runtime/tests.rs b/codex-rs/code-mode/src/session_runtime/tests.rs index e0838d89d8..969f06a22d 100644 --- a/codex-rs/code-mode/src/session_runtime/tests.rs +++ b/codex-rs/code-mode/src/session_runtime/tests.rs @@ -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(())); diff --git a/codex-rs/core-skills/src/loader.rs b/codex-rs/core-skills/src/loader.rs index b377e909ea..deab6f395b 100644 --- a/codex-rs/core-skills/src/loader.rs +++ b/codex-rs/core-skills/src/loader.rs @@ -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, } +#[derive(Debug, Clone, PartialEq, Eq)] +struct ParsedSkillFrontmatter { + name: String, + description: String, + short_description: Option, +} + 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, + warnings: Vec, +} + #[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, + queue: &mut VecDeque<(PathUri, usize)>, + visited_dirs: &mut HashSet, 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 = 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 = 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::>(); 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 { + 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), }) } diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs new file mode 100644 index 0000000000..55fe6e29f0 --- /dev/null +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -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, + pub dependencies: Option, + pub policy: Option, +} + +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) -> 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 { + 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, + pub warnings: Vec, +} + +/// 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, +) -> 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, Option) { + 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; diff --git a/codex-rs/core-skills/src/loader/environment_tests.rs b/codex-rs/core-skills/src/loader/environment_tests.rs new file mode 100644 index 0000000000..a817830b37 --- /dev/null +++ b/codex-rs/core-skills/src/loader/environment_tests.rs @@ -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()); +} diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index d049fd49f8..aa871f72ac 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -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. diff --git a/codex-rs/core/src/session/rollout_budget.rs b/codex-rs/core/src/session/rollout_budget.rs index fa90cf7b97..1b76047b7d 100644 --- a/codex-rs/core/src/session/rollout_budget.rs +++ b/codex-rs/core/src/session/rollout_budget.rs @@ -30,7 +30,7 @@ impl Session { .rollout_budget() .record_usage(usage) { - return Err(CodexErr::TurnAborted); + return Err(CodexErr::RolloutBudgetExceeded); } Ok(()) } diff --git a/codex-rs/core/tests/suite/rollout_budget.rs b/codex-rs/core/tests/suite/rollout_budget.rs index 6aa211bcc2..1e7a405f12 100644 --- a/codex-rs/core/tests/suite/rollout_budget.rs +++ b/codex-rs/core/tests/suite/rollout_budget.rs @@ -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(()) diff --git a/codex-rs/exec-server-protocol/BUILD.bazel b/codex-rs/exec-server-protocol/BUILD.bazel new file mode 100644 index 0000000000..6f79947562 --- /dev/null +++ b/codex-rs/exec-server-protocol/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "exec-server-protocol", + crate_name = "codex_exec_server_protocol", +) diff --git a/codex-rs/exec-server-protocol/Cargo.toml b/codex-rs/exec-server-protocol/Cargo.toml new file mode 100644 index 0000000000..5a8628705f --- /dev/null +++ b/codex-rs/exec-server-protocol/Cargo.toml @@ -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 } diff --git a/codex-rs/exec-server-protocol/src/lib.rs b/codex-rs/exec-server-protocol/src/lib.rs new file mode 100644 index 0000000000..b5a6ebc74f --- /dev/null +++ b/codex-rs/exec-server-protocol/src/lib.rs @@ -0,0 +1,7 @@ +mod process_id; +mod protocol; +pub mod rpc; + +pub use process_id::ProcessId; +pub use protocol::*; +pub use rpc::*; diff --git a/codex-rs/exec-server/src/process_id.rs b/codex-rs/exec-server-protocol/src/process_id.rs similarity index 100% rename from codex-rs/exec-server/src/process_id.rs rename to codex-rs/exec-server-protocol/src/process_id.rs diff --git a/codex-rs/exec-server/src/protocol.rs b/codex-rs/exec-server-protocol/src/protocol.rs similarity index 96% rename from codex-rs/exec-server/src/protocol.rs rename to codex-rs/exec-server-protocol/src/protocol.rs index d53dae471a..5ae80c37e4 100644 --- a/codex-rs/exec-server/src/protocol.rs +++ b/codex-rs/exec-server-protocol/src/protocol.rs @@ -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, } +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 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 { diff --git a/codex-rs/exec-server-protocol/src/rpc.rs b/codex-rs/exec-server-protocol/src/rpc.rs new file mode 100644 index 0000000000..09bedf0fe0 --- /dev/null +++ b/codex-rs/exec-server-protocol/src/rpc.rs @@ -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, +} + +/// 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, +} + +/// 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, + pub message: String, +} diff --git a/codex-rs/exec-server/Cargo.toml b/codex-rs/exec-server/Cargo.toml index 02f23dfc49..9f5383aa06 100644 --- a/codex-rs/exec-server/Cargo.toml +++ b/codex-rs/exec-server/Cargo.toml @@ -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 } diff --git a/codex-rs/exec-server/README.md b/codex-rs/exec-server/README.md index 6d2e663e2f..b1f9279186 100644 --- a/codex-rs/exec-server/README.md +++ b/codex-rs/exec-server/README.md @@ -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: diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 25d1b54e8b..f913892e18 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -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; diff --git a/codex-rs/exec-server/src/client/reqwest_http_client.rs b/codex-rs/exec-server/src/client/reqwest_http_client.rs index 9e22fc01d3..9de35315e1 100644 --- a/codex-rs/exec-server/src/client/reqwest_http_client.rs +++ b/codex-rs/exec-server/src/client/reqwest_http_client.rs @@ -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; diff --git a/codex-rs/exec-server/src/connection.rs b/codex-rs/exec-server/src/connection.rs index 2d2e5afe06..ebc12b2f39 100644 --- a/codex-rs/exec-server/src/connection.rs +++ b/codex-rs/exec-server/src/connection.rs @@ -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, }) } diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index b1e4ae5c55..664ef730d0 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -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 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; diff --git a/codex-rs/exec-server/src/fs_helper.rs b/codex-rs/exec-server/src/fs_helper.rs index 9fa74bf3d1..1eba3d0be3 100644 --- a/codex-rs/exec-server/src/fs_helper.rs +++ b/codex-rs/exec-server/src/fs_helper.rs @@ -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; diff --git a/codex-rs/exec-server/src/fs_sandbox.rs b/codex-rs/exec-server/src/fs_sandbox.rs index 41a3e12215..812d76751c 100644 --- a/codex-rs/exec-server/src/fs_sandbox.rs +++ b/codex-rs/exec-server/src/fs_sandbox.rs @@ -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; diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index d5c71674ac..c861e47bab 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -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; diff --git a/codex-rs/exec-server/src/local_process.rs b/codex-rs/exec-server/src/local_process.rs index d77209e7cf..973019808a 100644 --- a/codex-rs/exec-server/src/local_process.rs +++ b/codex-rs/exec-server/src/local_process.rs @@ -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; diff --git a/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs b/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs index 9119236cfd..232dfa479e 100644 --- a/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs +++ b/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs @@ -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; diff --git a/codex-rs/exec-server/src/noise_relay/message_framing.rs b/codex-rs/exec-server/src/noise_relay/message_framing.rs index 24ee48173e..402b68691a 100644 --- a/codex-rs/exec-server/src/noise_relay/message_framing.rs +++ b/codex-rs/exec-server/src/noise_relay/message_framing.rs @@ -1,4 +1,4 @@ -use codex_app_server_protocol::JSONRPCMessage; +use codex_exec_server_protocol::JSONRPCMessage; use crate::ExecServerError; diff --git a/codex-rs/exec-server/src/noise_relay/message_framing_tests.rs b/codex-rs/exec-server/src/noise_relay/message_framing_tests.rs index c3df14bf56..3b9ac7939e 100644 --- a/codex-rs/exec-server/src/noise_relay/message_framing_tests.rs +++ b/codex-rs/exec-server/src/noise_relay/message_framing_tests.rs @@ -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; diff --git a/codex-rs/exec-server/src/process_sandbox.rs b/codex-rs/exec-server/src/process_sandbox.rs index ef97d0ce5f..b5dd54e0b2 100644 --- a/codex-rs/exec-server/src/process_sandbox.rs +++ b/codex-rs/exec-server/src/process_sandbox.rs @@ -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; diff --git a/codex-rs/exec-server/src/relay.rs b/codex-rs/exec-server/src/relay.rs index 033da4b2a7..950fd60e7c 100644 --- a/codex-rs/exec-server/src/relay.rs +++ b/codex-rs/exec-server/src/relay.rs @@ -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, }) } diff --git a/codex-rs/exec-server/src/remote_file_system_path_uri_tests.rs b/codex-rs/exec-server/src/remote_file_system_path_uri_tests.rs index 1205c6d7e4..562f5d5643 100644 --- a/codex-rs/exec-server/src/remote_file_system_path_uri_tests.rs +++ b/codex-rs/exec-server/src/remote_file_system_path_uri_tests.rs @@ -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; diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index 82cf2b200a..8279f795ae 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -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>) { 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; diff --git a/codex-rs/exec-server/src/sandboxed_file_system.rs b/codex-rs/exec-server/src/sandboxed_file_system.rs index d0beb57a43..81007e65e3 100644 --- a/codex-rs/exec-server/src/sandboxed_file_system.rs +++ b/codex-rs/exec-server/src/sandboxed_file_system.rs @@ -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; diff --git a/codex-rs/exec-server/src/server/file_system_handler.rs b/codex-rs/exec-server/src/server/file_system_handler.rs index 1a6e073936..a77b7ac71f 100644 --- a/codex-rs/exec-server/src/server/file_system_handler.rs +++ b/codex-rs/exec-server/src/server/file_system_handler.rs @@ -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; diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index 07575a1c42..28d78a9006 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -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; diff --git a/codex-rs/exec-server/src/server/process_handler.rs b/codex-rs/exec-server/src/server/process_handler.rs index f628b242b0..9ff6690411 100644 --- a/codex-rs/exec-server/src/server/process_handler.rs +++ b/codex-rs/exec-server/src/server/process_handler.rs @@ -1,4 +1,4 @@ -use codex_app_server_protocol::JSONRPCErrorError; +use codex_exec_server_protocol::JSONRPCErrorError; use crate::ExecServerRuntimePaths; use crate::local_process::LocalProcess; diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index d960e53ca0..e163592130 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -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; diff --git a/codex-rs/exec-server/src/server/session_registry.rs b/codex-rs/exec-server/src/server/session_registry.rs index 73b14ff7cd..4d2ef7f893 100644 --- a/codex-rs/exec-server/src/server/session_registry.rs +++ b/codex-rs/exec-server/src/server/session_registry.rs @@ -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; diff --git a/codex-rs/exec-server/src/server/transport_tests.rs b/codex-rs/exec-server/src/server/transport_tests.rs index b9787d8a37..49d769e412 100644 --- a/codex-rs/exec-server/src/server/transport_tests.rs +++ b/codex-rs/exec-server/src/server/transport_tests.rs @@ -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; diff --git a/codex-rs/exec-server/tests/common/exec_server.rs b/codex-rs/exec-server/tests/common/exec_server.rs index ad089d5cda..bbd2d81168 100644 --- a/codex-rs/exec-server/tests/common/exec_server.rs +++ b/codex-rs/exec-server/tests/common/exec_server.rs @@ -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) diff --git a/codex-rs/exec-server/tests/http_client.rs b/codex-rs/exec-server/tests/http_client.rs index 6a087dd115..248564b493 100644 --- a/codex-rs/exec-server/tests/http_client.rs +++ b/codex-rs/exec-server/tests/http_client.rs @@ -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; diff --git a/codex-rs/exec-server/tests/http_request.rs b/codex-rs/exec-server/tests/http_request.rs index f45261dad7..a4734bf92d 100644 --- a/codex-rs/exec-server/tests/http_request.rs +++ b/codex-rs/exec-server/tests/http_request.rs @@ -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 { +) -> anyhow::Result { let response = server .wait_for_event(|event| { matches!( diff --git a/codex-rs/exec-server/tests/initialize.rs b/codex-rs/exec-server/tests/initialize.rs index 9c67395663..daf4306db6 100644 --- a/codex-rs/exec-server/tests/initialize.rs +++ b/codex-rs/exec-server/tests/initialize.rs @@ -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; diff --git a/codex-rs/exec-server/tests/process.rs b/codex-rs/exec-server/tests/process.rs index 7a1da19ac2..2255d4f8c2 100644 --- a/codex-rs/exec-server/tests/process.rs +++ b/codex-rs/exec-server/tests/process.rs @@ -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; diff --git a/codex-rs/exec-server/tests/websocket.rs b/codex-rs/exec-server/tests/websocket.rs index 3d0bebf106..4df1639139 100644 --- a/codex-rs/exec-server/tests/websocket.rs +++ b/codex-rs/exec-server/tests/websocket.rs @@ -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)?) diff --git a/codex-rs/ext/goal/tests/goal_extension_backend.rs b/codex-rs/ext/goal/tests/goal_extension_backend.rs index 32043c9808..b38bd63938 100644 --- a/codex-rs/ext/goal/tests/goal_extension_backend.rs +++ b/codex-rs/ext/goal/tests/goal_extension_backend.rs @@ -1,3 +1,4 @@ +#![recursion_limit = "256"] #![allow(clippy::expect_used)] use std::sync::Arc; diff --git a/codex-rs/ext/skills/Cargo.toml b/codex-rs/ext/skills/Cargo.toml index afd5111e95..35827ac578 100644 --- a/codex-rs/ext/skills/Cargo.toml +++ b/codex-rs/ext/skills/Cargo.toml @@ -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"] } diff --git a/codex-rs/ext/skills/src/catalog.rs b/codex-rs/ext/skills/src/catalog.rs index 96b8fec10f..59977a47ca 100644 --- a/codex-rs/ext/skills/src/catalog.rs +++ b/codex-rs/ext/skills/src/catalog.rs @@ -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, environment_id: impl Into, - 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. diff --git a/codex-rs/ext/skills/src/provider/executor.rs b/codex-rs/ext/skills/src/provider/executor.rs index 273a1614ba..9938035e92 100644 --- a/codex-rs/ext/skills/src/provider/executor.rs +++ b/codex-rs/ext/skills/src/provider/executor.rs @@ -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 { - path.to_abs_path() } diff --git a/codex-rs/ext/skills/tests/executor_file_system_authority.rs b/codex-rs/ext/skills/tests/executor_file_system_authority.rs index 8c5160e2c1..06bf062eac 100644 --- a/codex-rs/ext/skills/tests/executor_file_system_authority.rs +++ b/codex-rs/ext/skills/tests/executor_file_system_authority.rs @@ -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 { + self.canonical_root + .join(relative_path) + .map_err(io::Error::other) + } + async fn canonicalize(&self, path: &PathUri) -> io::Result { - 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> { - 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> { - 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 { - let skill_dir = self.canonical_root.join("skill"); - let skill_path = skill_dir.join("SKILL.md"); + fn metadata(&self, path: &PathUri) -> io::Result { + 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('/') ), ), ] diff --git a/codex-rs/protocol/src/error.rs b/codex-rs/protocol/src/error.rs index a2ea0eb691..4523c517da 100644 --- a/codex-rs/protocol/src/error.rs +++ b/codex-rs/protocol/src/error.rs @@ -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, diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index cc0d258075..971c37fcef 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -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 diff --git a/codex-rs/utils/plugins/src/lib.rs b/codex-rs/utils/plugins/src/lib.rs index 203b1f22e4..70e4dcdb86 100644 --- a/codex-rs/utils/plugins/src/lib.rs +++ b/codex-rs/utils/plugins/src/lib.rs @@ -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 { diff --git a/codex-rs/utils/plugins/src/plugin_namespace.rs b/codex-rs/utils/plugins/src/plugin_namespace.rs index b29ff5c5d6..085c4e1dbd 100644 --- a/codex-rs/utils/plugins/src/plugin_namespace.rs +++ b/codex-rs/utils/plugins/src/plugin_namespace.rs @@ -26,13 +26,12 @@ struct RawPluginManifestName { async fn plugin_manifest_name( fs: &dyn ExecutorFileSystem, - plugin_root: &AbsolutePathBuf, + plugin_root: &PathUri, ) -> Option { 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 { - 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 { + 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 }