1#![doc = include_str!("../README.md")]
2#![warn(missing_docs)]
3#![deny(rustdoc::broken_intra_doc_links)]
4#![cfg_attr(test, allow(clippy::unwrap_used))]
5
6#[cfg(not(feature = "bundled-cli"))]
7mod cache_paths;
8pub mod canvas;
10mod canvas_dispatch;
11#[cfg(feature = "bundled-cli")]
13pub(crate) mod embeddedcli;
14mod errors;
15pub mod extension_launch_provider;
17#[cfg(feature = "bundled-in-process")]
19pub(crate) mod ffi;
20pub use errors::*;
21pub mod copilot_request_handler;
25#[doc(hidden)]
28pub mod github_telemetry;
29pub mod github_token;
31pub mod handler;
33pub mod hooks;
35mod jsonrpc;
36pub mod permission;
38mod process_tree;
39pub mod provider_token;
41mod provider_token_dispatch;
42pub(crate) mod resolve;
44mod router;
45pub mod session;
47pub mod session_fs;
49mod session_fs_dispatch;
50pub mod startup_timings;
52pub mod subscription;
54pub mod tool;
56pub mod trace_context;
58pub mod transforms;
60pub mod types;
62mod wire;
63
64pub mod session_events;
66
67pub mod rpc;
70
71#[derive(serde::Deserialize)]
72struct SessionDetachResponse {
73 success: bool,
74 error: Option<String>,
75}
76
77pub(crate) mod generated;
82
83pub mod mode;
86
87use std::ffi::OsString;
88use std::path::{Path, PathBuf};
89use std::process::Stdio;
90use std::sync::{Arc, OnceLock};
91use std::time::{Duration, Instant};
92
93use async_trait::async_trait;
94pub use github_token::{
95 GitHubToken, GitHubTokenProvider, GitHubTokenProviderArgs, GitHubTokenProviderResult,
96 GitHubTokenRequestReason,
97};
98pub use indexmap::IndexMap;
102pub(crate) use jsonrpc::{
105 JsonRpcClient, JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, error_codes,
106};
107pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet};
108pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs};
109
110#[cfg(feature = "test-support")]
112pub mod test_support {
113 pub use crate::jsonrpc::{
114 JsonRpcClient, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
115 error_codes,
116 };
117}
118use serde::{Deserialize, Serialize};
119use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader};
120use tokio::net::TcpStream;
121use tokio::process::{Child, Command};
122use tokio::sync::{broadcast, mpsc, oneshot};
123use tracing::{Instrument, debug, error, info, warn};
124pub use types::*;
125
126mod sdk_protocol_version;
127pub use sdk_protocol_version::{SDK_PROTOCOL_VERSION, get_sdk_protocol_version};
128pub use startup_timings::StartupTimings;
129pub use subscription::{EventSubscription, LifecycleSubscription};
130
131const MIN_PROTOCOL_VERSION: u32 = 3;
133const RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
134
135fn record_optional_millis(span: &tracing::Span, field: &'static str, value: Option<u64>) {
136 match value {
137 Some(value) => {
138 span.record(field, value);
139 }
140 None => {
141 span.record(field, "None");
142 }
143 }
144}
145
146#[derive(Debug, Default)]
148#[non_exhaustive]
149pub enum Transport {
150 #[default]
153 Default,
154 Stdio,
156 InProcess,
169 Tcp {
171 port: u16,
173 connection_token: Option<String>,
177 },
178 External {
180 host: String,
182 port: u16,
184 connection_token: Option<String>,
187 },
188}
189
190#[derive(Debug, Clone, Default)]
192pub enum CliProgram {
193 #[default]
198 Resolve,
199 Path(PathBuf),
201}
202
203impl From<PathBuf> for CliProgram {
204 fn from(path: PathBuf) -> Self {
205 Self::Path(path)
206 }
207}
208
209pub const HAS_BUNDLED_CLI: bool = cfg!(has_bundled_cli);
216
217pub fn install_bundled_cli() -> Option<PathBuf> {
238 #[cfg(feature = "bundled-cli")]
239 {
240 embeddedcli::path()
241 }
242 #[cfg(not(feature = "bundled-cli"))]
243 {
244 None
245 }
246}
247
248pub fn install_bundled_runtime() -> Option<PathBuf> {
259 #[cfg(feature = "bundled-cli")]
260 {
261 embeddedcli::runtime_path()
262 }
263 #[cfg(not(feature = "bundled-cli"))]
264 {
265 None
266 }
267}
268
269#[non_exhaustive]
281pub struct ClientOptions {
282 pub program: CliProgram,
284 pub prefix_args: Vec<OsString>,
286 pub working_directory: PathBuf,
290 pub env: Vec<(OsString, OsString)>,
292 pub env_remove: Vec<OsString>,
294 pub extra_args: Vec<String>,
296 pub builtin_plugin_directories: Vec<PathBuf>,
301 pub transport: Transport,
303 pub github_token: Option<String>,
308 pub use_logged_in_user: Option<bool>,
312 pub log_level: Option<LogLevel>,
316 pub session_idle_timeout_seconds: Option<u64>,
322 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
330 pub session_fs: Option<SessionFsConfig>,
338 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
347 pub extension_launch_provider:
354 Option<Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>>,
355 #[doc(hidden)]
363 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
364 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
374 pub telemetry: Option<TelemetryConfig>,
378 pub base_directory: Option<PathBuf>,
383 pub enable_remote_sessions: bool,
389 pub bundled_cli_extract_dir: Option<PathBuf>,
408 pub mode: ClientMode,
412 pub client_info: Option<ClientInfo>,
419}
420
421#[derive(Debug, Clone, Default, PartialEq, Eq)]
433#[non_exhaustive]
434pub struct ClientInfo {
435 pub application_name: Option<String>,
437 pub application_version: Option<String>,
439 pub integration_name: Option<String>,
442 pub integration_version: Option<String>,
444}
445
446impl ClientInfo {
447 pub fn new() -> Self {
450 Self::default()
451 }
452
453 pub fn with_application_name(mut self, application_name: impl Into<String>) -> Self {
455 self.application_name = Some(application_name.into());
456 self
457 }
458
459 pub fn with_application_version(mut self, application_version: impl Into<String>) -> Self {
461 self.application_version = Some(application_version.into());
462 self
463 }
464
465 pub fn with_integration_name(mut self, integration_name: impl Into<String>) -> Self {
468 self.integration_name = Some(integration_name.into());
469 self
470 }
471
472 pub fn with_integration_version(mut self, integration_version: impl Into<String>) -> Self {
475 self.integration_version = Some(integration_version.into());
476 self
477 }
478
479 fn is_empty(&self) -> bool {
483 Self::non_empty(&self.application_name).is_none()
484 && Self::non_empty(&self.application_version).is_none()
485 && Self::non_empty(&self.integration_name).is_none()
486 && Self::non_empty(&self.integration_version).is_none()
487 }
488
489 fn non_empty(value: &Option<String>) -> Option<String> {
492 value.as_ref().filter(|s| !s.is_empty()).cloned()
493 }
494
495 fn to_wire(&self) -> Option<crate::generated::api_types::ConnectClientInfo> {
498 if self.is_empty() {
499 return None;
500 }
501 Some(crate::generated::api_types::ConnectClientInfo {
502 editor_name: Self::non_empty(&self.application_name),
503 editor_version: Self::non_empty(&self.application_version),
504 extension_name: Self::non_empty(&self.integration_name),
505 extension_version: Self::non_empty(&self.integration_version),
506 })
507 }
508}
509
510impl std::fmt::Debug for ClientOptions {
511 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
512 f.debug_struct("ClientOptions")
513 .field("program", &self.program)
514 .field("prefix_args", &self.prefix_args)
515 .field("working_directory", &self.working_directory)
516 .field("env", &self.env)
517 .field("env_remove", &self.env_remove)
518 .field("extra_args", &self.extra_args)
519 .field(
520 "builtin_plugin_directories",
521 &self.builtin_plugin_directories,
522 )
523 .field("transport", &self.transport)
524 .field(
525 "github_token",
526 &self.github_token.as_ref().map(|_| "<redacted>"),
527 )
528 .field("use_logged_in_user", &self.use_logged_in_user)
529 .field("log_level", &self.log_level)
530 .field(
531 "session_idle_timeout_seconds",
532 &self.session_idle_timeout_seconds,
533 )
534 .field(
535 "on_list_models",
536 &self.on_list_models.as_ref().map(|_| "<set>"),
537 )
538 .field("session_fs", &self.session_fs)
539 .field(
540 "request_handler",
541 &self.request_handler.as_ref().map(|_| "<set>"),
542 )
543 .field(
544 "extension_launch_provider",
545 &self.extension_launch_provider.as_ref().map(|_| "<set>"),
546 )
547 .field(
548 "on_github_telemetry",
549 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
550 )
551 .field(
552 "on_get_trace_context",
553 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
554 )
555 .field("telemetry", &self.telemetry)
556 .field("base_directory", &self.base_directory)
557 .field("enable_remote_sessions", &self.enable_remote_sessions)
558 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
559 .field("client_info", &self.client_info)
560 .finish()
561 }
562}
563
564#[async_trait]
573pub trait ListModelsHandler: Send + Sync + 'static {
574 async fn list_models(&self) -> Result<Vec<Model>>;
576}
577
578#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
580#[serde(rename_all = "lowercase")]
581pub enum LogLevel {
582 None,
584 Error,
586 Warning,
588 Info,
590 Debug,
592 All,
594}
595
596impl LogLevel {
597 pub fn as_str(self) -> &'static str {
599 match self {
600 Self::None => "none",
601 Self::Error => "error",
602 Self::Warning => "warning",
603 Self::Info => "info",
604 Self::Debug => "debug",
605 Self::All => "all",
606 }
607 }
608}
609
610impl std::fmt::Display for LogLevel {
611 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
612 f.write_str(self.as_str())
613 }
614}
615
616#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
621#[serde(rename_all = "kebab-case")]
622#[non_exhaustive]
623pub enum OtelExporterType {
624 OtlpHttp,
627 File,
630}
631
632impl OtelExporterType {
633 pub fn as_str(self) -> &'static str {
635 match self {
636 Self::OtlpHttp => "otlp-http",
637 Self::File => "file",
638 }
639 }
640}
641
642#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
648#[non_exhaustive]
649pub enum OtlpHttpProtocol {
650 #[serde(rename = "http/json")]
652 HttpJson,
653 #[serde(rename = "http/protobuf")]
655 HttpProtobuf,
656}
657
658impl OtlpHttpProtocol {
659 pub fn as_str(self) -> &'static str {
661 match self {
662 Self::HttpJson => "http/json",
663 Self::HttpProtobuf => "http/protobuf",
664 }
665 }
666}
667
668#[derive(Debug, Clone, Default)]
703#[non_exhaustive]
704pub struct TelemetryConfig {
705 pub otlp_endpoint: Option<String>,
707 pub otlp_protocol: Option<OtlpHttpProtocol>,
709 pub file_path: Option<PathBuf>,
711 pub exporter_type: Option<OtelExporterType>,
714 pub source_name: Option<String>,
718 pub capture_content: Option<bool>,
722}
723
724impl TelemetryConfig {
725 pub fn new() -> Self {
728 Self::default()
729 }
730
731 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
733 self.otlp_endpoint = Some(endpoint.into());
734 self
735 }
736
737 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
739 self.otlp_protocol = Some(protocol);
740 self
741 }
742
743 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
745 self.file_path = Some(path.into());
746 self
747 }
748
749 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
751 self.exporter_type = Some(exporter_type);
752 self
753 }
754
755 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
759 self.source_name = Some(source_name.into());
760 self
761 }
762
763 pub fn with_capture_content(mut self, capture: bool) -> Self {
767 self.capture_content = Some(capture);
768 self
769 }
770
771 pub fn is_empty(&self) -> bool {
774 self.otlp_endpoint.is_none()
775 && self.otlp_protocol.is_none()
776 && self.file_path.is_none()
777 && self.exporter_type.is_none()
778 && self.source_name.is_none()
779 && self.capture_content.is_none()
780 }
781}
782
783impl Default for ClientOptions {
784 fn default() -> Self {
785 Self {
786 program: CliProgram::Resolve,
787 prefix_args: Vec::new(),
788 working_directory: PathBuf::new(),
789 env: Vec::new(),
790 env_remove: Vec::new(),
791 extra_args: Vec::new(),
792 builtin_plugin_directories: Vec::new(),
793 transport: Transport::default(),
794 github_token: None,
795 use_logged_in_user: None,
796 log_level: None,
797 session_idle_timeout_seconds: None,
798 on_list_models: None,
799 session_fs: None,
800 request_handler: None,
801 extension_launch_provider: None,
802 on_github_telemetry: None,
803 on_get_trace_context: None,
804 telemetry: None,
805 base_directory: None,
806 enable_remote_sessions: false,
807 bundled_cli_extract_dir: None,
808 mode: ClientMode::default(),
809 client_info: None,
810 }
811 }
812}
813
814impl ClientOptions {
815 pub fn new() -> Self {
831 Self::default()
832 }
833
834 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
836 self.program = program.into();
837 self
838 }
839
840 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
842 where
843 I: IntoIterator<Item = S>,
844 S: Into<OsString>,
845 {
846 self.prefix_args = args.into_iter().map(Into::into).collect();
847 self
848 }
849
850 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
852 self.working_directory = cwd.into();
853 self
854 }
855
856 pub fn with_env<I, K, V>(mut self, env: I) -> Self
858 where
859 I: IntoIterator<Item = (K, V)>,
860 K: Into<OsString>,
861 V: Into<OsString>,
862 {
863 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
864 self
865 }
866
867 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
869 where
870 I: IntoIterator<Item = S>,
871 S: Into<OsString>,
872 {
873 self.env_remove = names.into_iter().map(Into::into).collect();
874 self
875 }
876
877 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
879 where
880 I: IntoIterator<Item = S>,
881 S: Into<String>,
882 {
883 self.extra_args = args.into_iter().map(Into::into).collect();
884 self
885 }
886
887 pub fn with_builtin_plugin_directories<I, P>(mut self, paths: I) -> Self
892 where
893 I: IntoIterator<Item = P>,
894 P: Into<PathBuf>,
895 {
896 self.builtin_plugin_directories = paths.into_iter().map(Into::into).collect();
897 self
898 }
899
900 pub fn with_transport(mut self, transport: Transport) -> Self {
902 self.transport = transport;
903 self
904 }
905
906 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
909 self.github_token = Some(token.into());
910 self
911 }
912
913 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
916 self.use_logged_in_user = Some(use_logged_in);
917 self
918 }
919
920 pub fn with_log_level(mut self, level: LogLevel) -> Self {
922 self.log_level = Some(level);
923 self
924 }
925
926 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
929 self.session_idle_timeout_seconds = Some(seconds);
930 self
931 }
932
933 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
936 where
937 H: ListModelsHandler + 'static,
938 {
939 self.on_list_models = Some(Arc::new(handler));
940 self
941 }
942
943 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
945 self.session_fs = Some(config);
946 self
947 }
948
949 pub fn with_request_handler<H>(mut self, handler: H) -> Self
954 where
955 H: crate::copilot_request_handler::CopilotRequestHandler,
956 {
957 self.request_handler = Some(Arc::new(handler));
958 self
959 }
960
961 pub fn with_extension_launch_provider<P>(mut self, provider: P) -> Self
966 where
967 P: crate::extension_launch_provider::ExtensionLaunchProvider,
968 {
969 self.extension_launch_provider = Some(Arc::new(provider));
970 self
971 }
972
973 #[doc(hidden)]
979 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
980 where
981 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
982 {
983 self.on_github_telemetry = Some(Arc::new(callback));
984 self
985 }
986
987 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
991 where
992 P: TraceContextProvider + 'static,
993 {
994 self.on_get_trace_context = Some(Arc::new(provider));
995 self
996 }
997
998 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
1000 self.telemetry = Some(config);
1001 self
1002 }
1003
1004 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
1007 self.base_directory = Some(dir.into());
1008 self
1009 }
1010
1011 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
1014 self.enable_remote_sessions = enabled;
1015 self
1016 }
1017
1018 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
1028 self.bundled_cli_extract_dir = Some(dir.into());
1029 self
1030 }
1031
1032 pub fn with_mode(mut self, mode: ClientMode) -> Self {
1037 self.mode = mode;
1038 self
1039 }
1040
1041 pub fn with_client_info(mut self, client_info: ClientInfo) -> Self {
1045 self.client_info = Some(client_info);
1046 self
1047 }
1048}
1049
1050fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
1052 if cfg.initial_cwd.trim().is_empty() {
1053 return Err(Error::with_message(
1054 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
1055 "invalid SessionFsConfig: initial_cwd must not be empty",
1056 ));
1057 }
1058 if cfg.session_state_path.trim().is_empty() {
1059 return Err(Error::with_message(
1060 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
1061 "invalid SessionFsConfig: session_state_path must not be empty",
1062 ));
1063 }
1064 Ok(())
1065}
1066
1067fn generate_connection_token() -> String {
1074 let mut bytes = [0u8; 16];
1075 getrandom::getrandom(&mut bytes)
1076 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
1077 let mut hex = String::with_capacity(32);
1078 for byte in bytes {
1079 use std::fmt::Write;
1080 let _ = write!(hex, "{byte:02x}");
1081 }
1082 hex
1083}
1084
1085const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
1090
1091fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
1093 let configured = options
1094 .env
1095 .iter()
1096 .find(|(key, _)| {
1097 key.to_string_lossy()
1098 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
1099 })
1100 .map(|(_, value)| value.to_string_lossy().into_owned());
1101 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
1102 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
1103}
1104
1105fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
1106 match value {
1107 None => Ok(Transport::Stdio),
1108 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
1109 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
1110 Some(v) => Err(Error::with_message(
1111 ErrorKind::InvalidConfig,
1112 format!(
1113 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
1114 Expected 'inprocess', 'stdio', or unset."
1115 ),
1116 )),
1117 }
1118}
1119
1120#[cfg(any(feature = "bundled-in-process", test))]
1121fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
1122 if !matches!(&options.program, CliProgram::Resolve) {
1123 return Err(Error::with_message(
1124 ErrorKind::InvalidConfig,
1125 "ClientOptions::program is not supported with Transport::InProcess; \
1126 set COPILOT_CLI_PATH only when using an externally provisioned runtime package",
1127 ));
1128 }
1129 if !options.extra_args.is_empty() {
1130 return Err(Error::with_message(
1131 ErrorKind::InvalidConfig,
1132 "ClientOptions::extra_args is not supported with Transport::InProcess; \
1133 use typed client options instead",
1134 ));
1135 }
1136
1137 let unsupported = if !options.working_directory.as_os_str().is_empty() {
1138 Some("working_directory")
1139 } else if !options.env.is_empty() {
1140 Some("env")
1141 } else if !options.env_remove.is_empty() {
1142 Some("env_remove")
1143 } else if options.telemetry.is_some() {
1144 Some("telemetry")
1145 } else if !options.prefix_args.is_empty() {
1146 Some("prefix_args")
1147 } else {
1148 None
1149 };
1150
1151 if let Some(option) = unsupported {
1152 return Err(Error::with_message(
1153 ErrorKind::InvalidConfig,
1154 format!(
1155 "ClientOptions::{option} is not supported with Transport::InProcess; \
1156 configure process-global settings on the host process instead"
1157 ),
1158 ));
1159 }
1160
1161 Ok(())
1162}
1163
1164#[derive(Clone)]
1169pub struct Client {
1170 inner: Arc<ClientInner>,
1171}
1172
1173impl std::fmt::Debug for Client {
1174 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1175 f.debug_struct("Client")
1176 .field("working_directory", &self.inner.cwd)
1177 .field("pid", &self.pid())
1178 .finish()
1179 }
1180}
1181
1182struct ClientInner {
1183 child: parking_lot::Mutex<Option<Child>>,
1184 process_tree: parking_lot::Mutex<Option<process_tree::ProcessTree>>,
1185 #[cfg(feature = "bundled-in-process")]
1186 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
1189 rpc: JsonRpcClient,
1190 cwd: PathBuf,
1191 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
1192 notification_tx: broadcast::Sender<JsonRpcNotification>,
1193 router: router::SessionRouter,
1194 github_token_registry: Arc<github_token::GitHubTokenRegistry>,
1195 negotiated_protocol_version: OnceLock<u32>,
1196 state: parking_lot::Mutex<ConnectionState>,
1197 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
1198 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1199 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
1200 session_fs_configured: bool,
1201 session_fs_sqlite_declared: bool,
1202 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
1205 extension_launch_provider: Arc<extension_launch_provider::ExtensionLaunchProviderDispatcher>,
1206 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1211 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1212 effective_connection_token: Option<String>,
1217 client_info: Option<ClientInfo>,
1221 pub(crate) mode: ClientMode,
1224 startup_timings: OnceLock<StartupTimings>,
1228}
1229
1230impl Client {
1231 pub async fn start(options: ClientOptions) -> Result<Self> {
1244 let start_time = Instant::now();
1245 let mut timings = StartupTimings::default();
1246 let mut options = options;
1247 if matches!(options.transport, Transport::Default) {
1248 options.transport = resolve_default_transport(&options)?;
1249 }
1250 if matches!(options.transport, Transport::InProcess) {
1251 #[cfg(not(feature = "bundled-in-process"))]
1252 {
1253 return Err(Error::with_message(
1254 ErrorKind::InvalidConfig,
1255 "Transport::InProcess requires the `bundled-in-process` Cargo feature",
1256 ));
1257 }
1258 #[cfg(feature = "bundled-in-process")]
1259 validate_inprocess_options(&options)?;
1260 }
1261 if options.mode == ClientMode::Empty
1262 && options.base_directory.is_none()
1263 && options.session_fs.is_none()
1264 {
1265 return Err(Error::with_message(
1266 ErrorKind::InvalidConfig,
1267 "ClientMode::Empty requires either `base_directory` or \
1268 `session_fs` to be set (no implicit ~/.copilot fallback).",
1269 ));
1270 }
1271 if let Some(cfg) = &options.session_fs {
1272 validate_session_fs_config(cfg)?;
1273 }
1274 let builtin_plugin_directories = options
1275 .builtin_plugin_directories
1276 .iter()
1277 .map(|path| {
1278 if !path.is_absolute() {
1279 return Err(Error::with_message(
1280 ErrorKind::InvalidConfig,
1281 format!(
1282 "builtin_plugin_directories must contain only absolute paths: {}",
1283 path.display()
1284 ),
1285 ));
1286 }
1287 path.to_str().map(str::to_owned).ok_or_else(|| {
1288 Error::with_message(
1289 ErrorKind::InvalidConfig,
1290 format!(
1291 "builtin_plugin_directories must contain valid UTF-8 paths: {}",
1292 path.display()
1293 ),
1294 )
1295 })
1296 })
1297 .collect::<Result<Vec<_>>>()?;
1298 if matches!(options.transport, Transport::External { .. }) {
1301 if options.github_token.is_some() {
1302 return Err(Error::with_message(
1303 ErrorKind::InvalidConfig,
1304 "invalid client configuration: github_token cannot be used with \
1305 Transport::External (external server manages its own auth)",
1306 ));
1307 }
1308 if options.use_logged_in_user == Some(true) {
1309 return Err(Error::with_message(
1310 ErrorKind::InvalidConfig,
1311 "invalid client configuration: use_logged_in_user cannot be used with \
1312 Transport::External (external server manages its own auth)",
1313 ));
1314 }
1315 }
1316 match &options.transport {
1320 Transport::Tcp {
1321 connection_token: Some(t),
1322 ..
1323 }
1324 | Transport::External {
1325 connection_token: Some(t),
1326 ..
1327 } if t.is_empty() => {
1328 return Err(Error::with_message(
1329 ErrorKind::InvalidConfig,
1330 "invalid client configuration: connection_token must be a non-empty string",
1331 ));
1332 }
1333 _ => {}
1334 }
1335 let effective_connection_token: Option<String> = match &mut options.transport {
1340 Transport::Default => unreachable!("default transport resolved above"),
1341 Transport::Stdio | Transport::InProcess => None,
1342 Transport::Tcp {
1343 connection_token, ..
1344 } => Some(
1345 connection_token
1346 .get_or_insert_with(generate_connection_token)
1347 .clone(),
1348 ),
1349 Transport::External {
1350 connection_token, ..
1351 } => connection_token.clone(),
1352 };
1353 let session_fs_config = options.session_fs.clone();
1354 let request_handler = options.request_handler.clone();
1355 let extension_launch_provider = options.extension_launch_provider.clone();
1356 let session_fs_sqlite_declared = session_fs_config
1357 .as_ref()
1358 .and_then(|c| c.capabilities.as_ref())
1359 .is_some_and(|caps| caps.sqlite);
1360 let program = match &options.program {
1361 CliProgram::Path(path) => {
1362 info!(path = %path.display(), "using explicit copilot CLI path");
1363 path.clone()
1364 }
1365 CliProgram::Resolve => {
1366 let resolve_start = Instant::now();
1367 let resolved = resolve::copilot_binary_with_extract_dir(
1368 options.bundled_cli_extract_dir.as_deref(),
1369 true,
1370 )?;
1371 let resolve_elapsed = resolve_start.elapsed();
1372 timings.program_resolve_ms = Some(StartupTimings::millis(resolve_elapsed));
1373 debug!(
1374 elapsed_ms = resolve_elapsed.as_millis(),
1375 "Client::start CLI program resolution complete"
1376 );
1377 info!(path = %resolved.display(), "resolved copilot runtime");
1378 #[cfg(windows)]
1379 {
1380 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1381 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1382 }) {
1383 warn!(
1384 path = %resolved.display(),
1385 ext = %ext,
1386 "resolved copilot CLI is a .cmd/.bat wrapper; \
1387 this may cause console window flashes on Windows"
1388 );
1389 }
1390 }
1391 resolved
1392 }
1393 };
1394 let working_directory = {
1395 let cwd = options.working_directory.clone();
1396 if cwd.as_os_str().is_empty() {
1397 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1398 } else {
1399 cwd
1400 }
1401 };
1402
1403 let transport_setup_start = Instant::now();
1404 let client = match options.transport {
1405 Transport::Default => unreachable!("default transport resolved above"),
1406 Transport::External {
1407 ref host,
1408 port,
1409 connection_token: _,
1410 } => {
1411 info!(host = %host, port = %port, "connecting to external CLI server");
1412 let connect_start = Instant::now();
1413 let stream = TcpStream::connect((host.as_str(), port)).await?;
1414 debug!(
1415 elapsed_ms = connect_start.elapsed().as_millis(),
1416 host = %host,
1417 port,
1418 "Client::start TCP connect complete"
1419 );
1420 let (reader, writer) = tokio::io::split(stream);
1421 Self::from_transport(
1422 reader,
1423 writer,
1424 None,
1425 None,
1426 working_directory,
1427 options.on_list_models,
1428 extension_launch_provider.clone(),
1429 session_fs_config.is_some(),
1430 session_fs_sqlite_declared,
1431 options.on_get_trace_context,
1432 options.on_github_telemetry,
1433 effective_connection_token.clone(),
1434 options.mode,
1435 options.client_info,
1436 )?
1437 }
1438 Transport::Tcp {
1439 port,
1440 connection_token: _,
1441 } => {
1442 let (mut child, tree, actual_port, spawn_elapsed, port_wait_elapsed) =
1443 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1444 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1445 timings.port_wait_ms = Some(StartupTimings::millis(port_wait_elapsed));
1446 let connect_start = Instant::now();
1447 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1448 debug!(
1449 elapsed_ms = connect_start.elapsed().as_millis(),
1450 port = actual_port,
1451 "Client::start TCP connect complete"
1452 );
1453 let (reader, writer) = tokio::io::split(stream);
1454 Self::drain_stderr(&mut child);
1455 Self::from_transport(
1456 reader,
1457 writer,
1458 Some(child),
1459 tree,
1460 working_directory,
1461 options.on_list_models,
1462 extension_launch_provider.clone(),
1463 session_fs_config.is_some(),
1464 session_fs_sqlite_declared,
1465 options.on_get_trace_context,
1466 options.on_github_telemetry,
1467 effective_connection_token.clone(),
1468 options.mode,
1469 options.client_info,
1470 )?
1471 }
1472 Transport::Stdio => {
1473 let (mut child, tree, spawn_elapsed) =
1474 Self::spawn_stdio(&program, &options, &working_directory)?;
1475 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1476 let stdin = child.stdin.take().expect("stdin is piped");
1477 let stdout = child.stdout.take().expect("stdout is piped");
1478 Self::drain_stderr(&mut child);
1479 Self::from_transport(
1480 stdout,
1481 stdin,
1482 Some(child),
1483 tree,
1484 working_directory,
1485 options.on_list_models,
1486 extension_launch_provider.clone(),
1487 session_fs_config.is_some(),
1488 session_fs_sqlite_declared,
1489 options.on_get_trace_context,
1490 options.on_github_telemetry,
1491 effective_connection_token.clone(),
1492 options.mode,
1493 options.client_info,
1494 )?
1495 }
1496 Transport::InProcess => {
1497 #[cfg(feature = "bundled-in-process")]
1498 {
1499 info!(runtime_path = %program.display(), "hosting copilot runtime in-process (FFI)");
1500 let mut environment = Vec::new();
1501 if let Some(base_directory) = &options.base_directory {
1502 let value = base_directory.to_str().ok_or_else(|| {
1503 Error::with_message(
1504 ErrorKind::InvalidConfig,
1505 "base_directory must be valid UTF-8 for Transport::InProcess",
1506 )
1507 })?;
1508 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1509 }
1510 if options.mode == ClientMode::Empty {
1511 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1512 }
1513 if let Some(github_token) = &options.github_token {
1514 environment
1515 .push(("COPILOT_SDK_AUTH_TOKEN".to_string(), github_token.clone()));
1516 }
1517 let mut args = Vec::new();
1518 args.extend(
1519 Self::log_level_args(&options)
1520 .into_iter()
1521 .map(str::to_string),
1522 );
1523 args.extend(Self::session_idle_timeout_args(&options));
1524 args.extend(Self::remote_args(&options));
1525 if options.github_token.is_some() {
1526 args.extend([
1527 "--auth-token-env".to_string(),
1528 "COPILOT_SDK_AUTH_TOKEN".to_string(),
1529 ]);
1530 }
1531 let use_logged_in_user = options
1532 .use_logged_in_user
1533 .unwrap_or(options.github_token.is_none());
1534 if !use_logged_in_user {
1535 args.push("--no-auto-login".to_string());
1536 }
1537 let explicit_cli = std::env::var_os("COPILOT_CLI_PATH")
1538 .map(PathBuf::from)
1539 .filter(|path| path.is_file());
1540 let host = crate::ffi::FfiHost::create(
1541 &program,
1542 explicit_cli.as_deref(),
1543 environment,
1544 args,
1545 )?;
1546 let (reader, writer, shared) = host.start().await?;
1547 let client = Self::from_transport(
1548 reader,
1549 writer,
1550 None,
1551 None,
1552 working_directory,
1553 options.on_list_models,
1554 extension_launch_provider.clone(),
1555 session_fs_config.is_some(),
1556 session_fs_sqlite_declared,
1557 options.on_get_trace_context,
1558 options.on_github_telemetry,
1559 effective_connection_token.clone(),
1560 options.mode,
1561 options.client_info,
1562 )?;
1563 *client.inner.ffi_host.lock() = Some(shared);
1564 client
1565 }
1566 #[cfg(not(feature = "bundled-in-process"))]
1567 unreachable!("in-process feature validation returned above")
1568 }
1569 };
1570 timings.transport_setup_ms = StartupTimings::millis(transport_setup_start.elapsed());
1571 debug!(
1572 elapsed_ms = start_time.elapsed().as_millis(),
1573 "Client::start transport setup complete"
1574 );
1575 let handshake_start = Instant::now();
1576 client.verify_protocol_version().await?;
1577 timings.handshake_ms = StartupTimings::millis(handshake_start.elapsed());
1578 debug!(
1579 elapsed_ms = start_time.elapsed().as_millis(),
1580 "Client::start protocol verification complete"
1581 );
1582 let request_dispatcher = request_handler.map(|handler| {
1583 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1584 handler,
1585 ));
1586 dispatcher.set_client(Arc::downgrade(&client.inner));
1587 let _ = client.inner.llm_inference.set(dispatcher.clone());
1588 dispatcher
1589 });
1590 if client.inner.extension_launch_provider.is_configured() {
1591 client.inner.router.ensure_started(
1592 &client.inner.notification_tx,
1593 &client.inner.request_rx,
1594 client.inner.extension_launch_provider.clone(),
1595 request_dispatcher.clone(),
1596 client.inner.on_github_telemetry.clone(),
1597 client.inner.github_token_registry.clone(),
1598 );
1599 client.rpc().register_extension_launch_provider().await?;
1600 }
1601 if !builtin_plugin_directories.is_empty() {
1602 client
1603 .call(
1604 "plugins.builtin.set",
1605 Some(serde_json::json!({ "paths": builtin_plugin_directories })),
1606 )
1607 .await?;
1608 }
1609 if let Some(cfg) = session_fs_config {
1610 let session_fs_start = Instant::now();
1611 let capabilities = cfg.capabilities.as_ref().map(|c| {
1612 crate::generated::api_types::SessionFsSetProviderCapabilities {
1613 sqlite: Some(c.sqlite),
1614 }
1615 });
1616 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1617 capabilities,
1618 conventions: cfg.conventions.into_wire(),
1619 initial_cwd: cfg.initial_cwd,
1620 session_state_path: cfg.session_state_path,
1621 };
1622 client.rpc().session_fs().set_provider(request).await?;
1623 let session_fs_elapsed = session_fs_start.elapsed();
1624 timings.session_fs_ms = Some(StartupTimings::millis(session_fs_elapsed));
1625 debug!(
1626 elapsed_ms = session_fs_elapsed.as_millis(),
1627 "Client::start session filesystem setup complete"
1628 );
1629 }
1630 if let Some(dispatcher) = request_dispatcher {
1631 let llm_inference_start = Instant::now();
1632 client.inner.router.ensure_started(
1635 &client.inner.notification_tx,
1636 &client.inner.request_rx,
1637 client.inner.extension_launch_provider.clone(),
1638 Some(dispatcher.clone()),
1639 client.inner.on_github_telemetry.clone(),
1640 client.inner.github_token_registry.clone(),
1641 );
1642 client.rpc().llm_inference().set_provider().await?;
1643 let llm_inference_elapsed = llm_inference_start.elapsed();
1644 timings.llm_handler_ms = Some(StartupTimings::millis(llm_inference_elapsed));
1645 debug!(
1646 elapsed_ms = llm_inference_elapsed.as_millis(),
1647 "Client::start Copilot request handler registration complete"
1648 );
1649 }
1650 timings.total_ms = StartupTimings::millis(start_time.elapsed());
1651 let timings_span = tracing::debug_span!(
1654 "Client::start timings",
1655 program_resolve_ms = tracing::field::Empty,
1656 process_spawn_ms = tracing::field::Empty,
1657 port_wait_ms = tracing::field::Empty,
1658 transport_setup_ms = timings.transport_setup_ms,
1659 handshake_ms = timings.handshake_ms,
1660 session_fs_ms = tracing::field::Empty,
1661 llm_handler_ms = tracing::field::Empty,
1662 total_ms = timings.total_ms,
1663 );
1664 record_optional_millis(
1665 &timings_span,
1666 "program_resolve_ms",
1667 timings.program_resolve_ms,
1668 );
1669 record_optional_millis(&timings_span, "process_spawn_ms", timings.process_spawn_ms);
1670 record_optional_millis(&timings_span, "port_wait_ms", timings.port_wait_ms);
1671 record_optional_millis(&timings_span, "session_fs_ms", timings.session_fs_ms);
1672 record_optional_millis(&timings_span, "llm_handler_ms", timings.llm_handler_ms);
1673 timings_span.in_scope(|| debug!("Client::start timings"));
1674 let _ = client.inner.startup_timings.set(timings);
1675 debug!(
1676 elapsed_ms = start_time.elapsed().as_millis(),
1677 "Client::start complete"
1678 );
1679 Ok(client)
1680 }
1681
1682 pub fn from_streams(
1686 reader: impl AsyncRead + Unpin + Send + 'static,
1687 writer: impl AsyncWrite + Unpin + Send + 'static,
1688 cwd: PathBuf,
1689 ) -> Result<Self> {
1690 Self::from_transport(
1691 reader,
1692 writer,
1693 None,
1694 None,
1695 cwd,
1696 None,
1697 None,
1698 false,
1699 false,
1700 None,
1701 None,
1702 None,
1703 ClientMode::default(),
1704 None,
1705 )
1706 }
1707
1708 #[doc(hidden)]
1711 #[cfg(any(test, feature = "test-support"))]
1712 pub fn from_streams_with_extension_launch_provider(
1713 reader: impl AsyncRead + Unpin + Send + 'static,
1714 writer: impl AsyncWrite + Unpin + Send + 'static,
1715 cwd: PathBuf,
1716 provider: Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1717 ) -> Result<Self> {
1718 Self::from_transport(
1719 reader,
1720 writer,
1721 None,
1722 None,
1723 cwd,
1724 None,
1725 Some(provider),
1726 false,
1727 false,
1728 None,
1729 None,
1730 None,
1731 ClientMode::default(),
1732 None,
1733 )
1734 }
1735
1736 #[cfg(any(test, feature = "test-support"))]
1744 pub fn from_streams_with_trace_provider(
1745 reader: impl AsyncRead + Unpin + Send + 'static,
1746 writer: impl AsyncWrite + Unpin + Send + 'static,
1747 cwd: PathBuf,
1748 provider: Arc<dyn TraceContextProvider>,
1749 ) -> Result<Self> {
1750 Self::from_transport(
1751 reader,
1752 writer,
1753 None,
1754 None,
1755 cwd,
1756 None,
1757 None,
1758 false,
1759 false,
1760 Some(provider),
1761 None,
1762 None,
1763 ClientMode::default(),
1764 None,
1765 )
1766 }
1767
1768 #[cfg(any(test, feature = "test-support"))]
1772 pub fn from_streams_with_connection_token(
1773 reader: impl AsyncRead + Unpin + Send + 'static,
1774 writer: impl AsyncWrite + Unpin + Send + 'static,
1775 cwd: PathBuf,
1776 token: Option<String>,
1777 ) -> Result<Self> {
1778 Self::from_transport(
1779 reader,
1780 writer,
1781 None,
1782 None,
1783 cwd,
1784 None,
1785 None,
1786 false,
1787 false,
1788 None,
1789 None,
1790 token,
1791 ClientMode::default(),
1792 None,
1793 )
1794 }
1795
1796 #[doc(hidden)]
1799 #[cfg(any(test, feature = "test-support"))]
1800 pub fn from_streams_with_github_telemetry(
1801 reader: impl AsyncRead + Unpin + Send + 'static,
1802 writer: impl AsyncWrite + Unpin + Send + 'static,
1803 cwd: PathBuf,
1804 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1805 ) -> Result<Self> {
1806 Self::from_transport(
1807 reader,
1808 writer,
1809 None,
1810 None,
1811 cwd,
1812 None,
1813 None,
1814 false,
1815 false,
1816 None,
1817 Some(on_github_telemetry),
1818 None,
1819 ClientMode::default(),
1820 None,
1821 )
1822 }
1823
1824 #[cfg(any(test, feature = "test-support"))]
1830 pub fn generate_connection_token_for_test() -> String {
1831 generate_connection_token()
1832 }
1833
1834 #[doc(hidden)]
1838 #[cfg(any(test, feature = "test-support"))]
1839 pub fn from_streams_with_client_info(
1840 reader: impl AsyncRead + Unpin + Send + 'static,
1841 writer: impl AsyncWrite + Unpin + Send + 'static,
1842 cwd: PathBuf,
1843 client_info: Option<ClientInfo>,
1844 ) -> Result<Self> {
1845 Self::from_transport(
1846 reader,
1847 writer,
1848 None,
1849 None,
1850 cwd,
1851 None,
1852 None,
1853 false,
1854 false,
1855 None,
1856 None,
1857 None,
1858 ClientMode::default(),
1859 client_info,
1860 )
1861 }
1862
1863 #[allow(clippy::too_many_arguments)]
1864 fn from_transport(
1865 reader: impl AsyncRead + Unpin + Send + 'static,
1866 writer: impl AsyncWrite + Unpin + Send + 'static,
1867 child: Option<Child>,
1868 process_tree: Option<process_tree::ProcessTree>,
1869 cwd: PathBuf,
1870 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1871 extension_launch_provider: Option<
1872 Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1873 >,
1874 session_fs_configured: bool,
1875 session_fs_sqlite_declared: bool,
1876 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1877 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1878 effective_connection_token: Option<String>,
1879 mode: ClientMode,
1880 client_info: Option<ClientInfo>,
1881 ) -> Result<Self> {
1882 let setup_start = Instant::now();
1883 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1884 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1885 let rpc = JsonRpcClient::new(
1886 writer,
1887 reader,
1888 notification_broadcast_tx.clone(),
1889 request_tx,
1890 );
1891
1892 let pid = child.as_ref().and_then(|c| c.id());
1893 info!(pid = ?pid, "copilot CLI client ready");
1894
1895 let github_token_registry = Arc::new(github_token::GitHubTokenRegistry::new());
1896 let extension_launch_provider = Arc::new(
1897 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(
1898 extension_launch_provider,
1899 ),
1900 );
1901 let client = Self {
1902 inner: Arc::new(ClientInner {
1903 child: parking_lot::Mutex::new(child),
1904 process_tree: parking_lot::Mutex::new(process_tree),
1905 #[cfg(feature = "bundled-in-process")]
1906 ffi_host: parking_lot::Mutex::new(None),
1907 rpc,
1908 cwd,
1909 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1910 notification_tx: notification_broadcast_tx,
1911 router: router::SessionRouter::new(),
1912 github_token_registry: github_token_registry.clone(),
1913 negotiated_protocol_version: OnceLock::new(),
1914 state: parking_lot::Mutex::new(ConnectionState::Connected),
1915 lifecycle_tx: broadcast::channel(256).0,
1916 on_list_models,
1917 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1918 session_fs_configured,
1919 session_fs_sqlite_declared,
1920 llm_inference: OnceLock::new(),
1921 extension_launch_provider: extension_launch_provider.clone(),
1922 on_github_telemetry,
1923 on_get_trace_context,
1924 effective_connection_token,
1925 mode,
1926 client_info,
1927 startup_timings: OnceLock::new(),
1928 }),
1929 };
1930 github_token_registry.set_client(Arc::downgrade(&client.inner));
1931 extension_launch_provider.set_client(Arc::downgrade(&client.inner));
1932 client.spawn_lifecycle_dispatcher();
1933 debug!(
1934 elapsed_ms = setup_start.elapsed().as_millis(),
1935 pid = ?pid,
1936 "Client::from_transport setup complete"
1937 );
1938 Ok(client)
1939 }
1940
1941 fn spawn_lifecycle_dispatcher(&self) {
1945 let mut notif_rx = self.inner.notification_tx.subscribe();
1946 let lifecycle_tx = self.inner.lifecycle_tx.clone();
1947 tokio::spawn(async move {
1948 loop {
1949 match notif_rx.recv().await {
1950 Ok(notification) => {
1951 if notification.method != "session.lifecycle" {
1952 continue;
1953 }
1954 let Some(params) = notification.params.as_ref() else {
1955 continue;
1956 };
1957 let event: SessionLifecycleEvent =
1958 match serde_json::from_value(params.clone()) {
1959 Ok(e) => e,
1960 Err(e) => {
1961 warn!(
1962 error = %e,
1963 "failed to deserialize session.lifecycle notification"
1964 );
1965 continue;
1966 }
1967 };
1968 let _ = lifecycle_tx.send(event);
1971 }
1972 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
1973 warn!(missed = n, "lifecycle dispatcher lagged");
1974 }
1975 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
1976 }
1977 }
1978 });
1979 }
1980
1981 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
1982 let mut command = Command::new(program);
1983 command.kill_on_drop(true);
1984 for arg in &options.prefix_args {
1985 command.arg(arg);
1986 }
1987 if let Some(token) = &options.github_token {
1990 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1991 }
1992 if let Some(telemetry) = &options.telemetry {
1995 command.env("COPILOT_OTEL_ENABLED", "true");
1996 if let Some(endpoint) = &telemetry.otlp_endpoint {
1997 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1998 }
1999 if let Some(protocol) = telemetry.otlp_protocol {
2000 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
2001 }
2002 if let Some(path) = &telemetry.file_path {
2003 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
2004 }
2005 if let Some(exporter) = telemetry.exporter_type {
2006 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
2007 }
2008 if let Some(source) = &telemetry.source_name {
2009 command.env("COPILOT_OTEL_SOURCE_NAME", source);
2010 }
2011 if let Some(capture) = telemetry.capture_content {
2012 command.env(
2013 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2014 if capture { "true" } else { "false" },
2015 );
2016 }
2017 }
2018 if let Some(dir) = &options.base_directory {
2019 command.env("COPILOT_HOME", dir);
2020 }
2021 if options.mode == ClientMode::Empty {
2024 command.env("COPILOT_DISABLE_KEYTAR", "1");
2025 }
2026 if let Transport::Tcp {
2027 connection_token: Some(token),
2028 ..
2029 } = &options.transport
2030 {
2031 command.env("COPILOT_CONNECTION_TOKEN", token);
2032 }
2033 for (key, value) in &options.env {
2034 command.env(key, value);
2035 }
2036 for key in &options.env_remove {
2037 command.env_remove(key);
2038 }
2039 command
2040 .current_dir(working_directory)
2041 .stdout(Stdio::piped())
2042 .stderr(Stdio::piped());
2043
2044 command
2045 }
2046
2047 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
2055 let mut args: Vec<&'static str> = Vec::new();
2056 if options.github_token.is_some() {
2057 args.push("--auth-token-env");
2058 args.push("COPILOT_SDK_AUTH_TOKEN");
2059 }
2060 let use_logged_in = options
2061 .use_logged_in_user
2062 .unwrap_or(options.github_token.is_none());
2063 if !use_logged_in {
2064 args.push("--no-auto-login");
2065 }
2066 args
2067 }
2068
2069 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
2073 match options.session_idle_timeout_seconds {
2074 Some(secs) if secs > 0 => {
2075 vec!["--session-idle-timeout".to_string(), secs.to_string()]
2076 }
2077 _ => Vec::new(),
2078 }
2079 }
2080
2081 fn remote_args(options: &ClientOptions) -> Vec<String> {
2082 if options.enable_remote_sessions {
2083 vec!["--remote".to_string()]
2084 } else {
2085 Vec::new()
2086 }
2087 }
2088
2089 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
2090 match options.log_level {
2091 Some(level) => vec!["--log-level", level.as_str()],
2092 None => Vec::new(),
2093 }
2094 }
2095
2096 fn spawn_stdio(
2097 program: &Path,
2098 options: &ClientOptions,
2099 working_directory: &Path,
2100 ) -> Result<(Child, Option<process_tree::ProcessTree>, Duration)> {
2101 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
2102 let mut command = Self::build_command(program, options, working_directory);
2103 command
2104 .args(["--server", "--stdio", "--no-auto-update"])
2105 .args(Self::log_level_args(options))
2106 .args(Self::auth_args(options))
2107 .args(Self::session_idle_timeout_args(options))
2108 .args(Self::remote_args(options))
2109 .args(&options.extra_args)
2110 .stdin(Stdio::piped());
2111 let spawn_start = Instant::now();
2112 let (child, tree) = process_tree::spawn(&mut command)?;
2113 let spawn_elapsed = spawn_start.elapsed();
2114 debug!(
2115 elapsed_ms = spawn_elapsed.as_millis(),
2116 "Client::spawn_stdio subprocess spawned"
2117 );
2118 Ok((child, tree, spawn_elapsed))
2119 }
2120
2121 async fn spawn_tcp(
2122 program: &Path,
2123 options: &ClientOptions,
2124 working_directory: &Path,
2125 port: u16,
2126 ) -> Result<(
2127 Child,
2128 Option<process_tree::ProcessTree>,
2129 u16,
2130 Duration,
2131 Duration,
2132 )> {
2133 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
2134 let mut command = Self::build_command(program, options, working_directory);
2135 command
2136 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
2137 .args(Self::log_level_args(options))
2138 .args(Self::auth_args(options))
2139 .args(Self::session_idle_timeout_args(options))
2140 .args(Self::remote_args(options))
2141 .args(&options.extra_args)
2142 .stdin(Stdio::null());
2143 let spawn_start = Instant::now();
2144 let (mut child, tree) = process_tree::spawn(&mut command)?;
2145 let spawn_elapsed = spawn_start.elapsed();
2146 debug!(
2147 elapsed_ms = spawn_elapsed.as_millis(),
2148 "Client::spawn_tcp subprocess spawned"
2149 );
2150 let stdout = child.stdout.take().expect("stdout is piped");
2151
2152 let (port_tx, port_rx) = oneshot::channel::<u16>();
2153 let span = tracing::error_span!("copilot_cli_port_scan");
2154 tokio::spawn(
2155 async move {
2156 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
2158 let mut lines = BufReader::new(stdout).lines();
2159 let mut port_tx = Some(port_tx);
2160 while let Ok(Some(line)) = lines.next_line().await {
2161 debug!(line = %line, "CLI stdout");
2162 if let Some(tx) = port_tx.take() {
2163 if let Some(caps) = port_re.captures(&line)
2164 && let Some(p) =
2165 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
2166 {
2167 let _ = tx.send(p);
2168 continue;
2169 }
2170 port_tx = Some(tx);
2172 }
2173 }
2174 }
2175 .instrument(span),
2176 );
2177
2178 let port_wait_start = Instant::now();
2179 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
2180 .await
2181 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
2182 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
2183
2184 let port_wait_elapsed = port_wait_start.elapsed();
2185 debug!(
2186 elapsed_ms = port_wait_elapsed.as_millis(),
2187 port = actual_port,
2188 "Client::spawn_tcp TCP port wait complete"
2189 );
2190 info!(port = %actual_port, "CLI server listening");
2191 Ok((child, tree, actual_port, spawn_elapsed, port_wait_elapsed))
2192 }
2193
2194 fn drain_stderr(child: &mut Child) {
2195 if let Some(stderr) = child.stderr.take() {
2196 let span = tracing::error_span!("copilot_cli");
2197 tokio::spawn(
2198 async move {
2199 let mut reader = BufReader::new(stderr).lines();
2200 while let Ok(Some(line)) = reader.next_line().await {
2201 warn!(line = %line, "CLI stderr");
2202 }
2203 }
2204 .instrument(span),
2205 );
2206 }
2207 }
2208
2209 pub fn cwd(&self) -> &PathBuf {
2211 &self.inner.cwd
2212 }
2213
2214 pub fn mode(&self) -> ClientMode {
2216 self.inner.mode
2217 }
2218
2219 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
2230 crate::generated::rpc::ClientRpc { client: self }
2231 }
2232
2233 #[allow(dead_code, reason = "convenience for future internal use")]
2235 pub(crate) async fn send_request(
2236 &self,
2237 method: &str,
2238 params: Option<serde_json::Value>,
2239 ) -> Result<JsonRpcResponse> {
2240 self.inner.rpc.send_request(method, params).await
2241 }
2242
2243 pub async fn call(
2263 &self,
2264 method: &str,
2265 params: Option<serde_json::Value>,
2266 ) -> Result<serde_json::Value> {
2267 self.call_with_inline_callback(method, params, None).await
2268 }
2269
2270 pub(crate) async fn detach_session(&self, session_id: &str) -> Result<()> {
2271 let value = self
2272 .call(
2273 "session.detach",
2274 Some(serde_json::json!({ "sessionId": session_id })),
2275 )
2276 .await?;
2277 let response: SessionDetachResponse = serde_json::from_value(value)?;
2278 if response.success {
2279 return Ok(());
2280 }
2281 Err(Error::with_message(
2282 ErrorKind::Session(SessionErrorKind::DetachFailed),
2283 response
2284 .error
2285 .unwrap_or_else(|| "unknown error".to_string()),
2286 ))
2287 }
2288
2289 pub(crate) async fn call_with_inline_callback(
2304 &self,
2305 method: &str,
2306 params: Option<serde_json::Value>,
2307 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
2308 ) -> Result<serde_json::Value> {
2309 let session_id: Option<SessionId> = params
2310 .as_ref()
2311 .and_then(|p| p.get("sessionId"))
2312 .and_then(|v| v.as_str())
2313 .map(SessionId::from);
2314 let response = self
2315 .inner
2316 .rpc
2317 .send_request_with_inline_callback(method, params, inline_callback)
2318 .await?;
2319 if let Some(err) = response.error {
2320 if err.message.contains("Session not found") {
2321 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
2322 session_id.unwrap_or_else(|| "unknown".into()),
2323 ))
2324 .into());
2325 }
2326 return Err(Error::from_rpc(err.code, err.message, err.data));
2327 }
2328 Ok(response.result.unwrap_or(serde_json::Value::Null))
2329 }
2330
2331 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
2333 self.inner.rpc.write(response).await
2334 }
2335
2336 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
2338 Self { inner }
2339 }
2340
2341 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
2345 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
2346 self.inner.request_rx.lock().take()
2347 }
2348
2349 pub(crate) fn register_session(
2360 &self,
2361 session_id: &SessionId,
2362 ) -> crate::router::SessionRegistration {
2363 self.inner.router.ensure_started(
2364 &self.inner.notification_tx,
2365 &self.inner.request_rx,
2366 self.inner.extension_launch_provider.clone(),
2367 self.inner.llm_inference.get().cloned(),
2368 self.inner.on_github_telemetry.clone(),
2369 self.inner.github_token_registry.clone(),
2370 );
2371 self.inner.router.register(session_id)
2372 }
2373
2374 pub(crate) fn unregister_session_owned(
2382 &self,
2383 session_id: &SessionId,
2384 token: crate::router::RegistrationToken,
2385 ) {
2386 self.inner.router.unregister_owned(session_id, token);
2387 }
2388
2389 #[cfg(any(test, feature = "test-support"))]
2396 pub(crate) fn registered_session_ids(&self) -> Vec<SessionId> {
2397 self.inner.router.session_ids()
2398 }
2399
2400 pub(crate) fn register_github_token_provider(
2401 &self,
2402 provider: Arc<dyn GitHubTokenProvider>,
2403 ) -> github_token::GitHubTokenRegistration {
2404 self.inner.router.ensure_started(
2405 &self.inner.notification_tx,
2406 &self.inner.request_rx,
2407 self.inner.extension_launch_provider.clone(),
2408 self.inner.llm_inference.get().cloned(),
2409 self.inner.on_github_telemetry.clone(),
2410 self.inner.github_token_registry.clone(),
2411 );
2412 let id = self.inner.github_token_registry.register(provider);
2413 github_token::GitHubTokenRegistration::new(self.inner.github_token_registry.clone(), id)
2414 }
2415
2416 pub(crate) fn retire_github_token_provider(&self, session_id: &SessionId) {
2417 self.inner.github_token_registry.retire_session(session_id);
2418 }
2419
2420 pub fn protocol_version(&self) -> Option<u32> {
2427 self.inner.negotiated_protocol_version.get().copied()
2428 }
2429
2430 pub fn startup_timings(&self) -> Option<StartupTimings> {
2437 self.inner.startup_timings.get().cloned()
2438 }
2439
2440 pub async fn verify_protocol_version(&self) -> Result<()> {
2464 let handshake_start = Instant::now();
2465 let mut used_fallback_ping = false;
2466 let server_version = match self.connect_handshake().await {
2470 Ok(v) => v,
2471 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2472 used_fallback_ping = true;
2473 self.ping(None).await?.protocol_version
2474 }
2475 Err(e) => return Err(e),
2476 };
2477
2478 match server_version {
2479 None => {
2480 warn!("CLI server did not report protocolVersion; skipping version check");
2481 }
2482 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2483 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2484 server: v,
2485 min: MIN_PROTOCOL_VERSION,
2486 max: SDK_PROTOCOL_VERSION,
2487 })
2488 .into());
2489 }
2490 Some(v) => {
2491 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2492 if existing != v {
2493 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2494 previous: existing,
2495 current: v,
2496 })
2497 .into());
2498 }
2499 } else {
2500 let _ = self.inner.negotiated_protocol_version.set(v);
2501 }
2502 }
2503 }
2504
2505 debug!(
2506 elapsed_ms = handshake_start.elapsed().as_millis(),
2507 protocol_version = ?server_version,
2508 used_fallback_ping,
2509 "Client::verify_protocol_version protocol handshake complete"
2510 );
2511 Ok(())
2512 }
2513
2514 async fn connect_handshake(&self) -> Result<Option<u32>> {
2521 let params = crate::generated::api_types::ConnectRequest {
2522 token: self.inner.effective_connection_token.clone(),
2523 enable_git_hub_telemetry_forwarding: self
2524 .inner
2525 .on_github_telemetry
2526 .is_some()
2527 .then_some(true),
2528 supported_task_kinds: Some(vec![
2529 crate::generated::api_types::TaskKind::Agent,
2530 crate::generated::api_types::TaskKind::Client,
2531 crate::generated::api_types::TaskKind::Shell,
2532 ]),
2533 client_info: self
2538 .inner
2539 .client_info
2540 .as_ref()
2541 .and_then(ClientInfo::to_wire),
2542 };
2543 let value = self
2544 .call(
2545 crate::generated::api_types::rpc_methods::CONNECT,
2546 Some(serde_json::to_value(params)?),
2547 )
2548 .await?;
2549 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2550 Ok(Some(u32::try_from(result.protocol_version).map_err(
2551 |_| ProtocolErrorKind::InvalidProtocolVersion {
2552 server: result.protocol_version,
2553 },
2554 )?))
2555 }
2556
2557 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2565 let params = match message {
2566 Some(m) => serde_json::json!({ "message": m }),
2567 None => serde_json::json!({}),
2568 };
2569 let value = self
2570 .call(generated::api_types::rpc_methods::PING, Some(params))
2571 .await?;
2572 Ok(serde_json::from_value(value)?)
2573 }
2574
2575 pub async fn list_sessions(
2578 &self,
2579 filter: Option<SessionListFilter>,
2580 ) -> Result<Vec<SessionMetadata>> {
2581 let params = match filter {
2582 Some(f) => serde_json::json!({ "filter": f }),
2583 None => serde_json::json!({}),
2584 };
2585 let result = self.call("session.list", Some(params)).await?;
2586 let response: ListSessionsResponse = serde_json::from_value(result)?;
2587 Ok(response.sessions)
2588 }
2589
2590 pub async fn get_session_metadata(
2608 &self,
2609 session_id: &SessionId,
2610 ) -> Result<Option<SessionMetadata>> {
2611 let result = self
2612 .call(
2613 "session.getMetadata",
2614 Some(serde_json::json!({ "sessionId": session_id })),
2615 )
2616 .await?;
2617 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2618 Ok(response.session)
2619 }
2620
2621 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2623 self.call(
2624 "session.delete",
2625 Some(serde_json::json!({ "sessionId": session_id })),
2626 )
2627 .await?;
2628 self.retire_github_token_provider(session_id);
2629 Ok(())
2630 }
2631
2632 #[cfg(feature = "test-support")]
2635 #[doc(hidden)]
2636 pub fn start_router_for_test(&self) {
2637 self.inner.router.ensure_started(
2638 &self.inner.notification_tx,
2639 &self.inner.request_rx,
2640 self.inner.extension_launch_provider.clone(),
2641 self.inner.llm_inference.get().cloned(),
2642 self.inner.on_github_telemetry.clone(),
2643 self.inner.github_token_registry.clone(),
2644 );
2645 }
2646
2647 #[cfg(feature = "test-support")]
2648 #[doc(hidden)]
2649 pub fn registered_session_ids_for_test(&self) -> Vec<SessionId> {
2653 self.registered_session_ids()
2654 }
2655
2656 #[cfg(feature = "test-support")]
2657 #[doc(hidden)]
2658 pub fn registered_session_count_for_test(&self) -> usize {
2662 self.inner.router.session_count()
2663 }
2664
2665 #[cfg(feature = "test-support")]
2666 #[doc(hidden)]
2667 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2670 let mut first_error = None;
2671
2672 for session_id in self.inner.router.session_ids() {
2673 if let Err(error) = self.detach_session(&session_id).await
2674 && first_error.is_none()
2675 {
2676 first_error = Some(error);
2677 }
2678 self.inner.router.unregister(&session_id);
2679 }
2680 self.inner.github_token_registry.clear();
2681
2682 match self.list_sessions(None).await {
2683 Ok(sessions) => {
2684 for session in sessions {
2685 if let Err(error) = self.delete_session(&session.session_id).await
2686 && first_error.is_none()
2687 {
2688 first_error = Some(error);
2689 }
2690 }
2691 }
2692 Err(error) if first_error.is_none() => first_error = Some(error),
2693 Err(_) => {}
2694 }
2695
2696 match first_error {
2697 Some(error) => Err(error),
2698 None => Ok(()),
2699 }
2700 }
2701
2702 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2718 let result = self
2719 .call("session.getLastId", Some(serde_json::json!({})))
2720 .await?;
2721 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2722 Ok(response.session_id)
2723 }
2724
2725 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2730 let result = self
2731 .call("session.getForeground", Some(serde_json::json!({})))
2732 .await?;
2733 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2734 Ok(response.session_id)
2735 }
2736
2737 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2742 self.call(
2743 "session.setForeground",
2744 Some(serde_json::json!({ "sessionId": session_id })),
2745 )
2746 .await?;
2747 Ok(())
2748 }
2749
2750 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2752 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2753 Ok(serde_json::from_value(result)?)
2754 }
2755
2756 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2758 let result = self
2759 .call("auth.getStatus", Some(serde_json::json!({})))
2760 .await?;
2761 Ok(serde_json::from_value(result)?)
2762 }
2763
2764 pub async fn list_models(&self) -> Result<Vec<Model>> {
2769 let cache = self.inner.models_cache.lock().clone();
2770 let models = cache
2771 .get_or_try_init(|| async {
2772 if let Some(handler) = &self.inner.on_list_models {
2773 handler.list_models().await
2774 } else {
2775 Ok(self.rpc().models().list().await?.models)
2776 }
2777 })
2778 .await?;
2779 Ok(models.clone())
2780 }
2781
2782 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2785 if let Some(provider) = &self.inner.on_get_trace_context {
2786 provider.get_trace_context().await
2787 } else {
2788 TraceContext::default()
2789 }
2790 }
2791
2792 pub fn pid(&self) -> Option<u32> {
2794 self.inner.child.lock().as_ref().and_then(|c| c.id())
2795 }
2796
2797 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2825 let pid = self.pid();
2826 info!(pid = ?pid, "stopping CLI process");
2827 let mut errors: Vec<Error> = Vec::new();
2828 self.inner.extension_launch_provider.clear();
2829
2830 for session_id in self.inner.router.session_ids() {
2833 match self.detach_session(&session_id).await {
2834 Ok(_) => {}
2835 Err(e) => {
2836 warn!(
2837 session_id = %session_id,
2838 error = %e,
2839 "session.detach failed during Client::stop",
2840 );
2841 errors.push(e);
2842 }
2843 }
2844 self.inner.router.unregister(&session_id);
2845 }
2846 self.inner.github_token_registry.clear();
2847
2848 let should_shutdown_runtime = self.inner.child.lock().is_some();
2849 #[cfg(feature = "bundled-in-process")]
2850 let should_shutdown_runtime =
2851 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2852 if should_shutdown_runtime {
2853 let runtime_shutdown_start = Instant::now();
2854 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2855 .await
2856 {
2857 Ok(Ok(())) => {
2858 debug!(
2859 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2860 "Client::stop runtime shutdown complete"
2861 );
2862 }
2863 Ok(Err(e)) => {
2864 warn!(
2865 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2866 error = %e,
2867 "runtime.shutdown failed during Client::stop",
2868 );
2869 errors.push(e);
2870 }
2871 Err(_) => {
2872 let e = std::io::Error::new(
2873 std::io::ErrorKind::TimedOut,
2874 "runtime.shutdown timed out during Client::stop",
2875 );
2876 warn!(
2877 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2878 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2879 error = %e,
2880 "runtime.shutdown timed out during Client::stop",
2881 );
2882 errors.push(e.into());
2883 }
2884 }
2885 }
2886
2887 let child = self.inner.child.lock().take();
2888 let process_tree = self.inner.process_tree.lock().take();
2889 *self.inner.state.lock() = ConnectionState::Disconnected;
2890 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2891 if let Some(process_tree) = process_tree
2892 && let Err(error) = process_tree.terminate()
2893 {
2894 errors.push(error.into());
2895 }
2896 if let Some(mut child) = child {
2897 match child.try_wait() {
2898 Ok(Some(_status)) => {}
2899 Ok(None) => {
2900 if let Err(e) = child.kill().await {
2907 errors.push(e.into());
2908 }
2909 }
2910 Err(e) => errors.push(e.into()),
2911 }
2912 }
2913
2914 self.inner.rpc.force_close();
2917 #[cfg(feature = "bundled-in-process")]
2918 {
2919 if let Some(host) = self.inner.ffi_host.lock().take() {
2920 host.close();
2921 }
2922 }
2923
2924 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2925 if errors.is_empty() {
2926 Ok(())
2927 } else {
2928 Err(StopErrors(errors))
2929 }
2930 }
2931
2932 pub fn force_stop(&self) {
2961 let pid = self.pid();
2962 info!(pid = ?pid, "force-stopping CLI process");
2963 self.inner.extension_launch_provider.clear();
2964 if let Some(process_tree) = self.inner.process_tree.lock().take()
2965 && let Err(error) = process_tree.terminate()
2966 {
2967 error!(pid = ?pid, %error, "failed to terminate CLI process tree");
2968 }
2969 if let Some(mut child) = self.inner.child.lock().take()
2970 && let Err(e) = child.start_kill()
2971 {
2972 error!(pid = ?pid, error = %e, "failed to send kill signal");
2973 }
2974 self.inner.rpc.force_close();
2975 #[cfg(feature = "bundled-in-process")]
2976 {
2977 if let Some(host) = self.inner.ffi_host.lock().take() {
2978 host.close();
2979 }
2980 }
2981 self.inner.router.clear();
2984 self.inner.github_token_registry.clear();
2985 *self.inner.state.lock() = ConnectionState::Disconnected;
2986 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2987 }
2988
2989 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
3024 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
3025 }
3026}
3027
3028impl Drop for ClientInner {
3029 fn drop(&mut self) {
3030 let pid = self.child.lock().as_ref().and_then(Child::id);
3031 if let Some(process_tree) = self.process_tree.lock().take()
3032 && let Err(error) = process_tree.terminate()
3033 {
3034 error!(pid = ?pid, %error, "failed to terminate CLI process tree on drop");
3035 }
3036 if let Some(ref mut child) = *self.child.lock() {
3037 if let Err(e) = child.start_kill() {
3038 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
3039 } else {
3040 info!(pid = ?pid, "kill signal sent for CLI process on drop");
3041 }
3042 }
3043 #[cfg(feature = "bundled-in-process")]
3044 {
3045 if let Some(host) = self.ffi_host.lock().take() {
3046 self.rpc.force_close();
3047 host.close();
3048 }
3049 }
3050 }
3051}
3052
3053#[cfg(test)]
3054mod tests {
3055 use super::*;
3056
3057 #[test]
3058 fn is_transport_failure_matches_request_cancelled() {
3059 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
3060 assert!(err.is_transport_failure());
3061 }
3062
3063 #[test]
3064 fn is_transport_failure_matches_io_error() {
3065 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
3066 assert!(err.is_transport_failure());
3067 }
3068
3069 #[test]
3070 fn is_transport_failure_rejects_rpc_error() {
3071 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
3072 assert!(!err.is_transport_failure());
3073 }
3074
3075 #[test]
3076 fn is_transport_failure_rejects_session_error() {
3077 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
3078 assert!(!err.is_transport_failure());
3079 }
3080
3081 #[test]
3082 fn client_options_builder_composes() {
3083 let opts = ClientOptions::new()
3084 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
3085 .with_prefix_args(["node"])
3086 .with_cwd(PathBuf::from("/tmp"))
3087 .with_env([("KEY", "value")])
3088 .with_env_remove(["UNWANTED"])
3089 .with_extra_args(["--quiet"])
3090 .with_github_token("ghp_test")
3091 .with_use_logged_in_user(false)
3092 .with_log_level(LogLevel::Debug)
3093 .with_session_idle_timeout_seconds(120)
3094 .with_enable_remote_sessions(true);
3095 assert!(matches!(opts.program, CliProgram::Path(_)));
3096 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
3097 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
3098 assert_eq!(
3099 opts.env,
3100 vec![(
3101 std::ffi::OsString::from("KEY"),
3102 std::ffi::OsString::from("value")
3103 )]
3104 );
3105 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
3106 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
3107 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
3108 assert_eq!(opts.use_logged_in_user, Some(false));
3109 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
3110 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
3111 assert!(opts.enable_remote_sessions);
3112 }
3113
3114 #[test]
3115 fn default_transport_values_resolve_without_process_state() {
3116 assert!(matches!(
3117 resolve_default_transport_value(None).unwrap(),
3118 Transport::Stdio
3119 ));
3120 assert!(matches!(
3121 resolve_default_transport_value(Some("stdio")).unwrap(),
3122 Transport::Stdio
3123 ));
3124 assert!(matches!(
3125 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
3126 Transport::InProcess
3127 ));
3128 assert!(resolve_default_transport_value(Some("tcp")).is_err());
3129 }
3130
3131 #[test]
3132 fn inprocess_rejects_process_scoped_options() {
3133 let invalid = [
3134 ClientOptions::new().with_cwd("."),
3135 ClientOptions::new().with_env([("KEY", "value")]),
3136 ClientOptions::new().with_env_remove(["KEY"]),
3137 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
3138 ClientOptions::new().with_prefix_args(["index.js"]),
3139 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
3140 ClientOptions::new().with_extra_args(["--verbose"]),
3141 ];
3142
3143 for options in invalid {
3144 assert!(validate_inprocess_options(&options).is_err());
3145 }
3146 }
3147
3148 #[test]
3149 fn inprocess_allows_typed_runtime_options() {
3150 let options = ClientOptions::new()
3151 .with_base_directory("state")
3152 .with_log_level(LogLevel::Debug)
3153 .with_session_idle_timeout_seconds(10)
3154 .with_github_token("token")
3155 .with_use_logged_in_user(false)
3156 .with_enable_remote_sessions(true);
3157
3158 assert!(validate_inprocess_options(&options).is_ok());
3159 }
3160
3161 #[cfg(not(feature = "bundled-in-process"))]
3162 #[tokio::test]
3163 async fn inprocess_requires_cargo_feature() {
3164 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
3165 .await
3166 .unwrap_err();
3167
3168 assert!(error.to_string().contains("bundled-in-process"));
3169 }
3170
3171 #[test]
3172 fn is_transport_failure_rejects_other_protocol_errors() {
3173 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
3174 assert!(!err.is_transport_failure());
3175 }
3176
3177 #[test]
3178 fn build_command_lets_env_remove_strip_injected_token() {
3179 let opts = ClientOptions {
3180 github_token: Some("secret".to_string()),
3181 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
3182 ..Default::default()
3183 };
3184 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3185 let action = cmd
3187 .as_std()
3188 .get_envs()
3189 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3190 .map(|(_, v)| v);
3191 assert_eq!(
3192 action,
3193 Some(None),
3194 "env_remove should win over github_token"
3195 );
3196 }
3197
3198 #[test]
3199 fn build_command_lets_env_override_injected_token() {
3200 let opts = ClientOptions {
3201 github_token: Some("from-options".to_string()),
3202 env: vec![(
3203 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
3204 std::ffi::OsString::from("from-env"),
3205 )],
3206 ..Default::default()
3207 };
3208 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3209 let value = cmd
3210 .as_std()
3211 .get_envs()
3212 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3213 .and_then(|(_, v)| v);
3214 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
3215 }
3216
3217 #[test]
3218 fn build_command_injects_github_token_by_default() {
3219 let opts = ClientOptions {
3220 github_token: Some("just-the-token".to_string()),
3221 ..Default::default()
3222 };
3223 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3224 let value = cmd
3225 .as_std()
3226 .get_envs()
3227 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3228 .and_then(|(_, v)| v);
3229 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
3230 }
3231
3232 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
3233 cmd.as_std()
3234 .get_envs()
3235 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
3236 .and_then(|(_, v)| v)
3237 }
3238
3239 #[test]
3240 fn telemetry_config_builder_composes() {
3241 let cfg = TelemetryConfig::new()
3242 .with_otlp_endpoint("http://collector:4318")
3243 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
3244 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
3245 .with_exporter_type(OtelExporterType::OtlpHttp)
3246 .with_source_name("my-app")
3247 .with_capture_content(true);
3248
3249 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
3250 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
3251 assert_eq!(
3252 cfg.file_path.as_deref(),
3253 Some(Path::new("/var/log/copilot.jsonl")),
3254 );
3255 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
3256 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
3257 assert_eq!(cfg.capture_content, Some(true));
3258 assert!(!cfg.is_empty());
3259 assert!(TelemetryConfig::new().is_empty());
3260 }
3261
3262 #[test]
3263 fn otlp_http_protocol_serde_matches_env_value() {
3264 for (protocol, wire) in [
3265 (OtlpHttpProtocol::HttpJson, "http/json"),
3266 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
3267 ] {
3268 assert_eq!(protocol.as_str(), wire);
3269
3270 let serialized = serde_json::to_string(&protocol).unwrap();
3271 assert_eq!(serialized, format!("\"{wire}\""));
3272
3273 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
3274 assert_eq!(deserialized, protocol);
3275 }
3276 }
3277
3278 #[test]
3279 fn build_command_sets_otel_env_when_telemetry_enabled() {
3280 let opts = ClientOptions {
3281 telemetry: Some(TelemetryConfig {
3282 otlp_endpoint: Some("http://collector:4318".to_string()),
3283 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
3284 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
3285 exporter_type: Some(OtelExporterType::OtlpHttp),
3286 source_name: Some("my-app".to_string()),
3287 capture_content: Some(true),
3288 }),
3289 ..Default::default()
3290 };
3291 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3292 assert_eq!(
3293 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
3294 Some(std::ffi::OsStr::new("true")),
3295 );
3296 assert_eq!(
3297 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3298 Some(std::ffi::OsStr::new("http://collector:4318")),
3299 );
3300 assert_eq!(
3301 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
3302 Some(std::ffi::OsStr::new("http/protobuf")),
3303 );
3304 assert_eq!(
3305 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
3306 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
3307 );
3308 assert_eq!(
3309 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
3310 Some(std::ffi::OsStr::new("otlp-http")),
3311 );
3312 assert_eq!(
3313 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
3314 Some(std::ffi::OsStr::new("my-app")),
3315 );
3316 assert_eq!(
3317 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
3318 Some(std::ffi::OsStr::new("true")),
3319 );
3320 }
3321
3322 #[test]
3323 fn build_command_omits_otel_env_when_telemetry_none() {
3324 let opts = ClientOptions::default();
3325 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3326 for key in [
3327 "COPILOT_OTEL_ENABLED",
3328 "OTEL_EXPORTER_OTLP_ENDPOINT",
3329 "OTEL_EXPORTER_OTLP_PROTOCOL",
3330 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3331 "COPILOT_OTEL_EXPORTER_TYPE",
3332 "COPILOT_OTEL_SOURCE_NAME",
3333 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3334 ] {
3335 assert!(
3336 env_value(&cmd, key).is_none(),
3337 "expected {key} to be unset when telemetry is None",
3338 );
3339 }
3340 }
3341
3342 #[test]
3343 fn build_command_omits_unset_telemetry_fields() {
3344 let opts = ClientOptions {
3345 telemetry: Some(TelemetryConfig {
3346 otlp_endpoint: Some("http://collector:4318".to_string()),
3347 ..Default::default()
3348 }),
3349 ..Default::default()
3350 };
3351 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3352 assert_eq!(
3354 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
3355 Some(std::ffi::OsStr::new("true")),
3356 );
3357 assert_eq!(
3358 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3359 Some(std::ffi::OsStr::new("http://collector:4318")),
3360 );
3361 for key in [
3363 "OTEL_EXPORTER_OTLP_PROTOCOL",
3364 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3365 "COPILOT_OTEL_EXPORTER_TYPE",
3366 "COPILOT_OTEL_SOURCE_NAME",
3367 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3368 ] {
3369 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
3370 }
3371 }
3372
3373 #[test]
3374 fn build_command_lets_user_env_override_telemetry() {
3375 let opts = ClientOptions {
3376 telemetry: Some(TelemetryConfig {
3377 otlp_endpoint: Some("http://from-config:4318".to_string()),
3378 ..Default::default()
3379 }),
3380 env: vec![(
3381 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
3382 std::ffi::OsString::from("http://from-user-env:4318"),
3383 )],
3384 ..Default::default()
3385 };
3386 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3387 assert_eq!(
3388 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3389 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
3390 "user-supplied options.env should override telemetry config",
3391 );
3392 }
3393
3394 #[test]
3395 fn build_command_sets_copilot_home_env_when_configured() {
3396 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
3397 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3398 assert_eq!(
3399 env_value(&cmd, "COPILOT_HOME"),
3400 Some(std::ffi::OsStr::new("/custom/copilot")),
3401 );
3402
3403 let opts = ClientOptions::default();
3404 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3405 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
3406 }
3407
3408 #[test]
3409 fn build_command_sets_connection_token_env_when_configured() {
3410 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3411 port: 0,
3412 connection_token: Some("secret-token".to_string()),
3413 });
3414 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3415 assert_eq!(
3416 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3417 Some(std::ffi::OsStr::new("secret-token")),
3418 );
3419
3420 let opts = ClientOptions::default();
3421 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3422 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3423 }
3424
3425 #[tokio::test]
3426 async fn start_rejects_empty_connection_token() {
3427 let opts = ClientOptions::new()
3428 .with_transport(Transport::Tcp {
3429 port: 0,
3430 connection_token: Some(String::new()),
3431 })
3432 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3433 let err = Client::start(opts).await.unwrap_err();
3434 assert!(
3435 matches!(err.kind(), ErrorKind::InvalidConfig),
3436 "got {err:?}"
3437 );
3438 }
3439
3440 #[tokio::test]
3441 async fn start_rejects_empty_external_connection_token() {
3442 let opts = ClientOptions::new()
3443 .with_transport(Transport::External {
3444 host: "127.0.0.1".to_string(),
3445 port: 1,
3446 connection_token: Some(String::new()),
3447 })
3448 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3449 let err = Client::start(opts).await.unwrap_err();
3450 assert!(
3451 matches!(err.kind(), ErrorKind::InvalidConfig),
3452 "got {err:?}"
3453 );
3454 }
3455
3456 #[test]
3457 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3458 let opts_true = ClientOptions {
3459 telemetry: Some(TelemetryConfig {
3460 capture_content: Some(true),
3461 ..Default::default()
3462 }),
3463 ..Default::default()
3464 };
3465 let opts_false = ClientOptions {
3466 telemetry: Some(TelemetryConfig {
3467 capture_content: Some(false),
3468 ..Default::default()
3469 }),
3470 ..Default::default()
3471 };
3472 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3473 let cmd_false =
3474 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3475 assert_eq!(
3476 env_value(
3477 &cmd_true,
3478 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3479 ),
3480 Some(std::ffi::OsStr::new("true")),
3481 );
3482 assert_eq!(
3483 env_value(
3484 &cmd_false,
3485 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3486 ),
3487 Some(std::ffi::OsStr::new("false")),
3488 );
3489 }
3490
3491 #[test]
3492 fn session_idle_timeout_args_are_omitted_by_default() {
3493 let opts = ClientOptions::default();
3494 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3495 }
3496
3497 #[test]
3498 fn session_idle_timeout_args_omitted_for_zero() {
3499 let opts = ClientOptions {
3500 session_idle_timeout_seconds: Some(0),
3501 ..Default::default()
3502 };
3503 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3504 }
3505
3506 #[test]
3507 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3508 let opts = ClientOptions {
3509 session_idle_timeout_seconds: Some(300),
3510 ..Default::default()
3511 };
3512 assert_eq!(
3513 Client::session_idle_timeout_args(&opts),
3514 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3515 );
3516 }
3517
3518 #[test]
3519 fn remote_args_omitted_by_default() {
3520 let opts = ClientOptions::default();
3521 assert!(Client::remote_args(&opts).is_empty());
3522 }
3523
3524 #[test]
3525 fn remote_args_emit_flag_when_enabled() {
3526 let opts = ClientOptions {
3527 enable_remote_sessions: true,
3528 ..Default::default()
3529 };
3530 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3531 }
3532
3533 #[test]
3534 fn log_level_args_omitted_when_unset() {
3535 let opts = ClientOptions::default();
3536 assert!(opts.log_level.is_none());
3537 assert!(
3538 Client::log_level_args(&opts).is_empty(),
3539 "with no caller-supplied log_level the SDK must not pass --log-level"
3540 );
3541 }
3542
3543 #[test]
3544 fn log_level_args_emit_flag_when_set() {
3545 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3546 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3547 }
3548
3549 #[test]
3550 fn log_level_str_round_trips() {
3551 for level in [
3552 LogLevel::None,
3553 LogLevel::Error,
3554 LogLevel::Warning,
3555 LogLevel::Info,
3556 LogLevel::Debug,
3557 LogLevel::All,
3558 ] {
3559 let s = level.as_str();
3560 let json = serde_json::to_string(&level).unwrap();
3561 assert_eq!(json, format!("\"{s}\""));
3562 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3563 assert_eq!(parsed, level);
3564 }
3565 }
3566
3567 #[test]
3568 fn client_options_debug_redacts_handler() {
3569 struct StubHandler;
3570 #[async_trait]
3571 impl ListModelsHandler for StubHandler {
3572 async fn list_models(&self) -> Result<Vec<Model>> {
3573 Ok(vec![])
3574 }
3575 }
3576 let opts = ClientOptions {
3577 on_list_models: Some(Arc::new(StubHandler)),
3578 github_token: Some("secret-token".into()),
3579 ..Default::default()
3580 };
3581 let debug = format!("{opts:?}");
3582 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3583 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3584 assert!(!debug.contains("secret-token"));
3585 }
3586
3587 #[tokio::test]
3588 async fn list_models_uses_on_list_models_handler_when_set() {
3589 use std::sync::atomic::{AtomicUsize, Ordering};
3590
3591 struct CountingHandler {
3592 calls: Arc<AtomicUsize>,
3593 models: Vec<Model>,
3594 }
3595 #[async_trait]
3596 impl ListModelsHandler for CountingHandler {
3597 async fn list_models(&self) -> Result<Vec<Model>> {
3598 self.calls.fetch_add(1, Ordering::SeqCst);
3599 Ok(self.models.clone())
3600 }
3601 }
3602
3603 let calls = Arc::new(AtomicUsize::new(0));
3604 let model = Model {
3605 id: "byok-gpt-4".into(),
3606 name: "BYOK GPT-4".into(),
3607 ..Default::default()
3608 };
3609 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3610 calls: Arc::clone(&calls),
3611 models: vec![model.clone()],
3612 });
3613
3614 let client = client_with_list_models_handler(handler);
3615
3616 let result = client.list_models().await.unwrap();
3617 assert_eq!(result.len(), 1);
3618 assert_eq!(result[0].id, "byok-gpt-4");
3619 assert_eq!(calls.load(Ordering::SeqCst), 1);
3620 }
3621
3622 #[tokio::test]
3623 async fn list_models_serializes_concurrent_cache_misses() {
3624 use std::sync::atomic::{AtomicUsize, Ordering};
3625
3626 struct SlowCountingHandler {
3627 calls: Arc<AtomicUsize>,
3628 models: Vec<Model>,
3629 }
3630 #[async_trait]
3631 impl ListModelsHandler for SlowCountingHandler {
3632 async fn list_models(&self) -> Result<Vec<Model>> {
3633 self.calls.fetch_add(1, Ordering::SeqCst);
3634 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3635 Ok(self.models.clone())
3636 }
3637 }
3638
3639 let calls = Arc::new(AtomicUsize::new(0));
3640 let model = Model {
3641 id: "single-flight-model".into(),
3642 name: "Single Flight Model".into(),
3643 ..Default::default()
3644 };
3645 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3646 calls: Arc::clone(&calls),
3647 models: vec![model],
3648 });
3649 let client = client_with_list_models_handler(handler);
3650
3651 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3652 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3653 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3654 assert_eq!(calls.load(Ordering::SeqCst), 1);
3655 }
3656
3657 #[tokio::test]
3658 async fn cancelled_resume_session_unregisters_pending_session() {
3659 let (client_write, _server_read) = tokio::io::duplex(8192);
3660 let (_server_write, client_read) = tokio::io::duplex(8192);
3661 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3662 assert!(client.startup_timings().is_none());
3663 let session_id = SessionId::new("resume-cancel-test");
3664 let handle = tokio::spawn({
3665 let client = client.clone();
3666 async move {
3667 client
3668 .resume_session(ResumeSessionConfig::new(session_id))
3669 .await
3670 }
3671 });
3672
3673 wait_for_pending_session_registration(&client).await;
3674 handle.abort();
3675 let _ = handle.await;
3676
3677 assert!(client.inner.router.session_ids().is_empty());
3678 client.force_stop();
3679 }
3680
3681 #[cfg(any(unix, windows))]
3682 #[tokio::test]
3683 async fn dropping_last_client_kills_spawned_cli() {
3684 let temp = tempfile::tempdir().unwrap();
3685 let ready = temp.path().join("ready");
3686 let survived = temp.path().join("survived");
3687 let child = test_child_command(temp.path(), &ready, &survived)
3688 .spawn()
3689 .unwrap();
3690 let (client_write, _server_read) = tokio::io::duplex(64);
3691 let (_server_write, client_read) = tokio::io::duplex(64);
3692 let client = Client::from_transport(
3693 client_read,
3694 client_write,
3695 Some(child),
3696 None,
3697 temp.path().to_path_buf(),
3698 None,
3699 None,
3700 false,
3701 false,
3702 None,
3703 None,
3704 None,
3705 ClientMode::default(),
3706 None,
3707 )
3708 .unwrap();
3709
3710 wait_for_test_child(&ready).await;
3711 drop(client);
3712
3713 assert_test_child_killed(&survived).await;
3714 }
3715
3716 #[cfg(any(unix, windows))]
3717 #[tokio::test]
3718 async fn spawned_child_is_killed_when_dropped() {
3719 let temp = tempfile::tempdir().unwrap();
3720 let ready = temp.path().join("ready");
3721 let survived = temp.path().join("survived");
3722 let child = test_child_command(temp.path(), &ready, &survived)
3723 .spawn()
3724 .unwrap();
3725
3726 wait_for_test_child(&ready).await;
3727 drop(child);
3728
3729 assert_test_child_killed(&survived).await;
3730 }
3731
3732 #[cfg(any(unix, windows))]
3733 fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command {
3734 #[cfg(unix)]
3735 let mut command = {
3736 let mut command =
3737 Client::build_command(Path::new("sh"), &ClientOptions::default(), temp);
3738 command.args([
3739 "-c",
3740 "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"",
3741 ]);
3742 command
3743 };
3744 #[cfg(windows)]
3745 let mut command = {
3746 let mut command =
3747 Client::build_command(Path::new("powershell.exe"), &ClientOptions::default(), temp);
3748 command.args([
3749 "-NoLogo",
3750 "-NoProfile",
3751 "-NonInteractive",
3752 "-Command",
3753 "Set-Content -LiteralPath $env:READY ready; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:SURVIVED survived",
3754 ]);
3755 command
3756 };
3757 command.env("READY", ready).env("SURVIVED", survived);
3758 command
3759 }
3760
3761 #[cfg(any(unix, windows))]
3762 async fn wait_for_test_child(ready: &Path) {
3763 let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
3764 while !ready.exists() {
3765 assert!(
3766 tokio::time::Instant::now() < deadline,
3767 "child did not report readiness"
3768 );
3769 tokio::time::sleep(Duration::from_millis(10)).await;
3770 }
3771 }
3772
3773 #[cfg(any(unix, windows))]
3774 async fn assert_test_child_killed(survived: &Path) {
3775 tokio::time::sleep(Duration::from_millis(1500)).await;
3776
3777 assert!(
3778 !survived.exists(),
3779 "child survived after its owner was dropped"
3780 );
3781 }
3782
3783 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3784 Client {
3785 inner: Arc::new(ClientInner {
3786 child: parking_lot::Mutex::new(None),
3787 process_tree: parking_lot::Mutex::new(None),
3788 #[cfg(feature = "bundled-in-process")]
3789 ffi_host: parking_lot::Mutex::new(None),
3790 rpc: {
3791 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3792 let (notif_tx, _notif_rx) = broadcast::channel(16);
3793 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3794 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3795 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3796 },
3797 cwd: PathBuf::from("."),
3798 request_rx: parking_lot::Mutex::new(None),
3799 notification_tx: broadcast::channel(16).0,
3800 router: router::SessionRouter::new(),
3801 github_token_registry: Arc::new(github_token::GitHubTokenRegistry::new()),
3802 negotiated_protocol_version: OnceLock::new(),
3803 state: parking_lot::Mutex::new(ConnectionState::Connected),
3804 lifecycle_tx: broadcast::channel(16).0,
3805 on_list_models: Some(handler),
3806 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3807 session_fs_configured: false,
3808 session_fs_sqlite_declared: false,
3809 llm_inference: OnceLock::new(),
3810 extension_launch_provider: Arc::new(
3811 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(None),
3812 ),
3813 on_github_telemetry: None,
3814 on_get_trace_context: None,
3815 effective_connection_token: None,
3816 mode: ClientMode::default(),
3817 client_info: None,
3818 startup_timings: OnceLock::new(),
3819 }),
3820 }
3821 }
3822
3823 async fn wait_for_pending_session_registration(client: &Client) {
3824 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3825 while client.inner.router.session_ids().is_empty() {
3826 assert!(
3827 tokio::time::Instant::now() < deadline,
3828 "session was not registered"
3829 );
3830 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3831 }
3832 }
3833}