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