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
6pub mod canvas;
8mod canvas_dispatch;
9#[cfg(feature = "bundled-cli")]
11pub(crate) mod embeddedcli;
12mod errors;
13pub mod extension_launch_provider;
15#[cfg(feature = "bundled-in-process")]
17pub(crate) mod ffi;
18pub use errors::*;
19pub mod copilot_request_handler;
23#[doc(hidden)]
26pub mod github_telemetry;
27pub mod github_token;
29pub mod handler;
31pub mod hooks;
33mod jsonrpc;
34pub mod permission;
36mod process_tree;
37pub mod provider_token;
39mod provider_token_dispatch;
40pub(crate) mod resolve;
42mod router;
43pub mod session;
45pub mod session_fs;
47mod session_fs_dispatch;
48pub mod startup_timings;
50pub mod subscription;
52pub mod tool;
54pub mod trace_context;
56pub mod transforms;
58pub mod types;
60mod wire;
61
62pub mod session_events;
64
65pub mod rpc;
68
69pub(crate) mod generated;
74
75pub mod mode;
78
79use std::ffi::OsString;
80use std::path::{Path, PathBuf};
81use std::process::Stdio;
82use std::sync::{Arc, OnceLock};
83use std::time::{Duration, Instant};
84
85use async_trait::async_trait;
86pub use github_token::{
87 GitHubToken, GitHubTokenProvider, GitHubTokenProviderArgs, GitHubTokenProviderResult,
88 GitHubTokenRequestReason,
89};
90pub use indexmap::IndexMap;
94pub(crate) use jsonrpc::{
97 JsonRpcClient, JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, error_codes,
98};
99pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet};
100pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs};
101
102#[cfg(feature = "test-support")]
104pub mod test_support {
105 pub use crate::jsonrpc::{
106 JsonRpcClient, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
107 error_codes,
108 };
109}
110use serde::{Deserialize, Serialize};
111use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader};
112use tokio::net::TcpStream;
113use tokio::process::{Child, Command};
114use tokio::sync::{broadcast, mpsc, oneshot};
115use tracing::{Instrument, debug, error, info, warn};
116pub use types::*;
117
118mod sdk_protocol_version;
119pub use sdk_protocol_version::{SDK_PROTOCOL_VERSION, get_sdk_protocol_version};
120pub use startup_timings::StartupTimings;
121pub use subscription::{EventSubscription, LifecycleSubscription};
122
123const MIN_PROTOCOL_VERSION: u32 = 3;
125const RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
126
127fn record_optional_millis(span: &tracing::Span, field: &'static str, value: Option<u64>) {
128 match value {
129 Some(value) => {
130 span.record(field, value);
131 }
132 None => {
133 span.record(field, "None");
134 }
135 }
136}
137
138#[derive(Debug, Default)]
140#[non_exhaustive]
141pub enum Transport {
142 #[default]
145 Default,
146 Stdio,
148 InProcess,
161 Tcp {
163 port: u16,
165 connection_token: Option<String>,
169 },
170 External {
172 host: String,
174 port: u16,
176 connection_token: Option<String>,
179 },
180}
181
182#[derive(Debug, Clone, Default)]
184pub enum CliProgram {
185 #[default]
190 Resolve,
191 Path(PathBuf),
193}
194
195impl From<PathBuf> for CliProgram {
196 fn from(path: PathBuf) -> Self {
197 Self::Path(path)
198 }
199}
200
201pub const HAS_BUNDLED_CLI: bool = cfg!(has_bundled_cli);
208
209pub fn install_bundled_cli() -> Option<PathBuf> {
230 #[cfg(feature = "bundled-cli")]
231 {
232 embeddedcli::path()
233 }
234 #[cfg(not(feature = "bundled-cli"))]
235 {
236 None
237 }
238}
239
240pub fn install_bundled_runtime() -> Option<PathBuf> {
251 #[cfg(feature = "bundled-cli")]
252 {
253 embeddedcli::runtime_path()
254 }
255 #[cfg(not(feature = "bundled-cli"))]
256 {
257 None
258 }
259}
260
261#[non_exhaustive]
273pub struct ClientOptions {
274 pub program: CliProgram,
276 pub prefix_args: Vec<OsString>,
278 pub working_directory: PathBuf,
282 pub env: Vec<(OsString, OsString)>,
284 pub env_remove: Vec<OsString>,
286 pub extra_args: Vec<String>,
288 pub builtin_plugin_directories: Vec<PathBuf>,
293 pub transport: Transport,
295 pub github_token: Option<String>,
300 pub use_logged_in_user: Option<bool>,
304 pub log_level: Option<LogLevel>,
308 pub session_idle_timeout_seconds: Option<u64>,
314 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
322 pub session_fs: Option<SessionFsConfig>,
330 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
339 pub extension_launch_provider:
346 Option<Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>>,
347 #[doc(hidden)]
355 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
356 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
366 pub telemetry: Option<TelemetryConfig>,
370 pub base_directory: Option<PathBuf>,
375 pub enable_remote_sessions: bool,
381 pub bundled_cli_extract_dir: Option<PathBuf>,
400 pub mode: ClientMode,
404 pub client_info: Option<ClientInfo>,
411}
412
413#[derive(Debug, Clone, Default, PartialEq, Eq)]
425#[non_exhaustive]
426pub struct ClientInfo {
427 pub application_name: Option<String>,
429 pub application_version: Option<String>,
431 pub integration_name: Option<String>,
434 pub integration_version: Option<String>,
436}
437
438impl ClientInfo {
439 pub fn new() -> Self {
442 Self::default()
443 }
444
445 pub fn with_application_name(mut self, application_name: impl Into<String>) -> Self {
447 self.application_name = Some(application_name.into());
448 self
449 }
450
451 pub fn with_application_version(mut self, application_version: impl Into<String>) -> Self {
453 self.application_version = Some(application_version.into());
454 self
455 }
456
457 pub fn with_integration_name(mut self, integration_name: impl Into<String>) -> Self {
460 self.integration_name = Some(integration_name.into());
461 self
462 }
463
464 pub fn with_integration_version(mut self, integration_version: impl Into<String>) -> Self {
467 self.integration_version = Some(integration_version.into());
468 self
469 }
470
471 fn is_empty(&self) -> bool {
475 Self::non_empty(&self.application_name).is_none()
476 && Self::non_empty(&self.application_version).is_none()
477 && Self::non_empty(&self.integration_name).is_none()
478 && Self::non_empty(&self.integration_version).is_none()
479 }
480
481 fn non_empty(value: &Option<String>) -> Option<String> {
484 value.as_ref().filter(|s| !s.is_empty()).cloned()
485 }
486
487 fn to_wire(&self) -> Option<crate::generated::api_types::ConnectClientInfo> {
490 if self.is_empty() {
491 return None;
492 }
493 Some(crate::generated::api_types::ConnectClientInfo {
494 editor_name: Self::non_empty(&self.application_name),
495 editor_version: Self::non_empty(&self.application_version),
496 extension_name: Self::non_empty(&self.integration_name),
497 extension_version: Self::non_empty(&self.integration_version),
498 })
499 }
500}
501
502impl std::fmt::Debug for ClientOptions {
503 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
504 f.debug_struct("ClientOptions")
505 .field("program", &self.program)
506 .field("prefix_args", &self.prefix_args)
507 .field("working_directory", &self.working_directory)
508 .field("env", &self.env)
509 .field("env_remove", &self.env_remove)
510 .field("extra_args", &self.extra_args)
511 .field(
512 "builtin_plugin_directories",
513 &self.builtin_plugin_directories,
514 )
515 .field("transport", &self.transport)
516 .field(
517 "github_token",
518 &self.github_token.as_ref().map(|_| "<redacted>"),
519 )
520 .field("use_logged_in_user", &self.use_logged_in_user)
521 .field("log_level", &self.log_level)
522 .field(
523 "session_idle_timeout_seconds",
524 &self.session_idle_timeout_seconds,
525 )
526 .field(
527 "on_list_models",
528 &self.on_list_models.as_ref().map(|_| "<set>"),
529 )
530 .field("session_fs", &self.session_fs)
531 .field(
532 "request_handler",
533 &self.request_handler.as_ref().map(|_| "<set>"),
534 )
535 .field(
536 "extension_launch_provider",
537 &self.extension_launch_provider.as_ref().map(|_| "<set>"),
538 )
539 .field(
540 "on_github_telemetry",
541 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
542 )
543 .field(
544 "on_get_trace_context",
545 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
546 )
547 .field("telemetry", &self.telemetry)
548 .field("base_directory", &self.base_directory)
549 .field("enable_remote_sessions", &self.enable_remote_sessions)
550 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
551 .field("client_info", &self.client_info)
552 .finish()
553 }
554}
555
556#[async_trait]
565pub trait ListModelsHandler: Send + Sync + 'static {
566 async fn list_models(&self) -> Result<Vec<Model>>;
568}
569
570#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
572#[serde(rename_all = "lowercase")]
573pub enum LogLevel {
574 None,
576 Error,
578 Warning,
580 Info,
582 Debug,
584 All,
586}
587
588impl LogLevel {
589 pub fn as_str(self) -> &'static str {
591 match self {
592 Self::None => "none",
593 Self::Error => "error",
594 Self::Warning => "warning",
595 Self::Info => "info",
596 Self::Debug => "debug",
597 Self::All => "all",
598 }
599 }
600}
601
602impl std::fmt::Display for LogLevel {
603 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
604 f.write_str(self.as_str())
605 }
606}
607
608#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
613#[serde(rename_all = "kebab-case")]
614#[non_exhaustive]
615pub enum OtelExporterType {
616 OtlpHttp,
619 File,
622}
623
624impl OtelExporterType {
625 pub fn as_str(self) -> &'static str {
627 match self {
628 Self::OtlpHttp => "otlp-http",
629 Self::File => "file",
630 }
631 }
632}
633
634#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
640#[non_exhaustive]
641pub enum OtlpHttpProtocol {
642 #[serde(rename = "http/json")]
644 HttpJson,
645 #[serde(rename = "http/protobuf")]
647 HttpProtobuf,
648}
649
650impl OtlpHttpProtocol {
651 pub fn as_str(self) -> &'static str {
653 match self {
654 Self::HttpJson => "http/json",
655 Self::HttpProtobuf => "http/protobuf",
656 }
657 }
658}
659
660#[derive(Debug, Clone, Default)]
695#[non_exhaustive]
696pub struct TelemetryConfig {
697 pub otlp_endpoint: Option<String>,
699 pub otlp_protocol: Option<OtlpHttpProtocol>,
701 pub file_path: Option<PathBuf>,
703 pub exporter_type: Option<OtelExporterType>,
706 pub source_name: Option<String>,
710 pub capture_content: Option<bool>,
714}
715
716impl TelemetryConfig {
717 pub fn new() -> Self {
720 Self::default()
721 }
722
723 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
725 self.otlp_endpoint = Some(endpoint.into());
726 self
727 }
728
729 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
731 self.otlp_protocol = Some(protocol);
732 self
733 }
734
735 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
737 self.file_path = Some(path.into());
738 self
739 }
740
741 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
743 self.exporter_type = Some(exporter_type);
744 self
745 }
746
747 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
751 self.source_name = Some(source_name.into());
752 self
753 }
754
755 pub fn with_capture_content(mut self, capture: bool) -> Self {
759 self.capture_content = Some(capture);
760 self
761 }
762
763 pub fn is_empty(&self) -> bool {
766 self.otlp_endpoint.is_none()
767 && self.otlp_protocol.is_none()
768 && self.file_path.is_none()
769 && self.exporter_type.is_none()
770 && self.source_name.is_none()
771 && self.capture_content.is_none()
772 }
773}
774
775impl Default for ClientOptions {
776 fn default() -> Self {
777 Self {
778 program: CliProgram::Resolve,
779 prefix_args: Vec::new(),
780 working_directory: PathBuf::new(),
781 env: Vec::new(),
782 env_remove: Vec::new(),
783 extra_args: Vec::new(),
784 builtin_plugin_directories: Vec::new(),
785 transport: Transport::default(),
786 github_token: None,
787 use_logged_in_user: None,
788 log_level: None,
789 session_idle_timeout_seconds: None,
790 on_list_models: None,
791 session_fs: None,
792 request_handler: None,
793 extension_launch_provider: None,
794 on_github_telemetry: None,
795 on_get_trace_context: None,
796 telemetry: None,
797 base_directory: None,
798 enable_remote_sessions: false,
799 bundled_cli_extract_dir: None,
800 mode: ClientMode::default(),
801 client_info: None,
802 }
803 }
804}
805
806impl ClientOptions {
807 pub fn new() -> Self {
823 Self::default()
824 }
825
826 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
828 self.program = program.into();
829 self
830 }
831
832 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
834 where
835 I: IntoIterator<Item = S>,
836 S: Into<OsString>,
837 {
838 self.prefix_args = args.into_iter().map(Into::into).collect();
839 self
840 }
841
842 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
844 self.working_directory = cwd.into();
845 self
846 }
847
848 pub fn with_env<I, K, V>(mut self, env: I) -> Self
850 where
851 I: IntoIterator<Item = (K, V)>,
852 K: Into<OsString>,
853 V: Into<OsString>,
854 {
855 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
856 self
857 }
858
859 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
861 where
862 I: IntoIterator<Item = S>,
863 S: Into<OsString>,
864 {
865 self.env_remove = names.into_iter().map(Into::into).collect();
866 self
867 }
868
869 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
871 where
872 I: IntoIterator<Item = S>,
873 S: Into<String>,
874 {
875 self.extra_args = args.into_iter().map(Into::into).collect();
876 self
877 }
878
879 pub fn with_builtin_plugin_directories<I, P>(mut self, paths: I) -> Self
884 where
885 I: IntoIterator<Item = P>,
886 P: Into<PathBuf>,
887 {
888 self.builtin_plugin_directories = paths.into_iter().map(Into::into).collect();
889 self
890 }
891
892 pub fn with_transport(mut self, transport: Transport) -> Self {
894 self.transport = transport;
895 self
896 }
897
898 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
901 self.github_token = Some(token.into());
902 self
903 }
904
905 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
908 self.use_logged_in_user = Some(use_logged_in);
909 self
910 }
911
912 pub fn with_log_level(mut self, level: LogLevel) -> Self {
914 self.log_level = Some(level);
915 self
916 }
917
918 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
921 self.session_idle_timeout_seconds = Some(seconds);
922 self
923 }
924
925 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
928 where
929 H: ListModelsHandler + 'static,
930 {
931 self.on_list_models = Some(Arc::new(handler));
932 self
933 }
934
935 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
937 self.session_fs = Some(config);
938 self
939 }
940
941 pub fn with_request_handler<H>(mut self, handler: H) -> Self
946 where
947 H: crate::copilot_request_handler::CopilotRequestHandler,
948 {
949 self.request_handler = Some(Arc::new(handler));
950 self
951 }
952
953 pub fn with_extension_launch_provider<P>(mut self, provider: P) -> Self
958 where
959 P: crate::extension_launch_provider::ExtensionLaunchProvider,
960 {
961 self.extension_launch_provider = Some(Arc::new(provider));
962 self
963 }
964
965 #[doc(hidden)]
971 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
972 where
973 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
974 {
975 self.on_github_telemetry = Some(Arc::new(callback));
976 self
977 }
978
979 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
983 where
984 P: TraceContextProvider + 'static,
985 {
986 self.on_get_trace_context = Some(Arc::new(provider));
987 self
988 }
989
990 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
992 self.telemetry = Some(config);
993 self
994 }
995
996 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
999 self.base_directory = Some(dir.into());
1000 self
1001 }
1002
1003 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
1006 self.enable_remote_sessions = enabled;
1007 self
1008 }
1009
1010 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
1020 self.bundled_cli_extract_dir = Some(dir.into());
1021 self
1022 }
1023
1024 pub fn with_mode(mut self, mode: ClientMode) -> Self {
1029 self.mode = mode;
1030 self
1031 }
1032
1033 pub fn with_client_info(mut self, client_info: ClientInfo) -> Self {
1037 self.client_info = Some(client_info);
1038 self
1039 }
1040}
1041
1042fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
1044 if cfg.initial_cwd.trim().is_empty() {
1045 return Err(Error::with_message(
1046 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
1047 "invalid SessionFsConfig: initial_cwd must not be empty",
1048 ));
1049 }
1050 if cfg.session_state_path.trim().is_empty() {
1051 return Err(Error::with_message(
1052 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
1053 "invalid SessionFsConfig: session_state_path must not be empty",
1054 ));
1055 }
1056 Ok(())
1057}
1058
1059fn generate_connection_token() -> String {
1066 let mut bytes = [0u8; 16];
1067 getrandom::getrandom(&mut bytes)
1068 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
1069 let mut hex = String::with_capacity(32);
1070 for byte in bytes {
1071 use std::fmt::Write;
1072 let _ = write!(hex, "{byte:02x}");
1073 }
1074 hex
1075}
1076
1077const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
1082
1083fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
1085 let configured = options
1086 .env
1087 .iter()
1088 .find(|(key, _)| {
1089 key.to_string_lossy()
1090 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
1091 })
1092 .map(|(_, value)| value.to_string_lossy().into_owned());
1093 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
1094 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
1095}
1096
1097fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
1098 match value {
1099 None => Ok(Transport::Stdio),
1100 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
1101 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
1102 Some(v) => Err(Error::with_message(
1103 ErrorKind::InvalidConfig,
1104 format!(
1105 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
1106 Expected 'inprocess', 'stdio', or unset."
1107 ),
1108 )),
1109 }
1110}
1111
1112#[cfg(any(feature = "bundled-in-process", test))]
1113fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
1114 if !matches!(&options.program, CliProgram::Resolve) {
1115 return Err(Error::with_message(
1116 ErrorKind::InvalidConfig,
1117 "ClientOptions::program is not supported with Transport::InProcess; \
1118 set COPILOT_CLI_PATH only when using an externally provisioned runtime package",
1119 ));
1120 }
1121 if !options.extra_args.is_empty() {
1122 return Err(Error::with_message(
1123 ErrorKind::InvalidConfig,
1124 "ClientOptions::extra_args is not supported with Transport::InProcess; \
1125 use typed client options instead",
1126 ));
1127 }
1128
1129 let unsupported = if !options.working_directory.as_os_str().is_empty() {
1130 Some("working_directory")
1131 } else if !options.env.is_empty() {
1132 Some("env")
1133 } else if !options.env_remove.is_empty() {
1134 Some("env_remove")
1135 } else if options.telemetry.is_some() {
1136 Some("telemetry")
1137 } else if !options.prefix_args.is_empty() {
1138 Some("prefix_args")
1139 } else {
1140 None
1141 };
1142
1143 if let Some(option) = unsupported {
1144 return Err(Error::with_message(
1145 ErrorKind::InvalidConfig,
1146 format!(
1147 "ClientOptions::{option} is not supported with Transport::InProcess; \
1148 configure process-global settings on the host process instead"
1149 ),
1150 ));
1151 }
1152
1153 Ok(())
1154}
1155
1156#[derive(Clone)]
1161pub struct Client {
1162 inner: Arc<ClientInner>,
1163}
1164
1165impl std::fmt::Debug for Client {
1166 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1167 f.debug_struct("Client")
1168 .field("working_directory", &self.inner.cwd)
1169 .field("pid", &self.pid())
1170 .finish()
1171 }
1172}
1173
1174struct ClientInner {
1175 child: parking_lot::Mutex<Option<Child>>,
1176 process_tree: parking_lot::Mutex<Option<process_tree::ProcessTree>>,
1177 #[cfg(feature = "bundled-in-process")]
1178 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
1181 rpc: JsonRpcClient,
1182 cwd: PathBuf,
1183 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
1184 notification_tx: broadcast::Sender<JsonRpcNotification>,
1185 router: router::SessionRouter,
1186 github_token_registry: Arc<github_token::GitHubTokenRegistry>,
1187 negotiated_protocol_version: OnceLock<u32>,
1188 state: parking_lot::Mutex<ConnectionState>,
1189 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
1190 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1191 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
1192 session_fs_configured: bool,
1193 session_fs_sqlite_declared: bool,
1194 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
1197 extension_launch_provider: Arc<extension_launch_provider::ExtensionLaunchProviderDispatcher>,
1198 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1203 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1204 effective_connection_token: Option<String>,
1209 client_info: Option<ClientInfo>,
1213 pub(crate) mode: ClientMode,
1216 startup_timings: OnceLock<StartupTimings>,
1220}
1221
1222impl Client {
1223 pub async fn start(options: ClientOptions) -> Result<Self> {
1236 let start_time = Instant::now();
1237 let mut timings = StartupTimings::default();
1238 let mut options = options;
1239 if matches!(options.transport, Transport::Default) {
1240 options.transport = resolve_default_transport(&options)?;
1241 }
1242 if matches!(options.transport, Transport::InProcess) {
1243 #[cfg(not(feature = "bundled-in-process"))]
1244 {
1245 return Err(Error::with_message(
1246 ErrorKind::InvalidConfig,
1247 "Transport::InProcess requires the `bundled-in-process` Cargo feature",
1248 ));
1249 }
1250 #[cfg(feature = "bundled-in-process")]
1251 validate_inprocess_options(&options)?;
1252 }
1253 if options.mode == ClientMode::Empty
1254 && options.base_directory.is_none()
1255 && options.session_fs.is_none()
1256 {
1257 return Err(Error::with_message(
1258 ErrorKind::InvalidConfig,
1259 "ClientMode::Empty requires either `base_directory` or \
1260 `session_fs` to be set (no implicit ~/.copilot fallback).",
1261 ));
1262 }
1263 if let Some(cfg) = &options.session_fs {
1264 validate_session_fs_config(cfg)?;
1265 }
1266 let builtin_plugin_directories = options
1267 .builtin_plugin_directories
1268 .iter()
1269 .map(|path| {
1270 if !path.is_absolute() {
1271 return Err(Error::with_message(
1272 ErrorKind::InvalidConfig,
1273 format!(
1274 "builtin_plugin_directories must contain only absolute paths: {}",
1275 path.display()
1276 ),
1277 ));
1278 }
1279 path.to_str().map(str::to_owned).ok_or_else(|| {
1280 Error::with_message(
1281 ErrorKind::InvalidConfig,
1282 format!(
1283 "builtin_plugin_directories must contain valid UTF-8 paths: {}",
1284 path.display()
1285 ),
1286 )
1287 })
1288 })
1289 .collect::<Result<Vec<_>>>()?;
1290 if matches!(options.transport, Transport::External { .. }) {
1293 if options.github_token.is_some() {
1294 return Err(Error::with_message(
1295 ErrorKind::InvalidConfig,
1296 "invalid client configuration: github_token cannot be used with \
1297 Transport::External (external server manages its own auth)",
1298 ));
1299 }
1300 if options.use_logged_in_user == Some(true) {
1301 return Err(Error::with_message(
1302 ErrorKind::InvalidConfig,
1303 "invalid client configuration: use_logged_in_user cannot be used with \
1304 Transport::External (external server manages its own auth)",
1305 ));
1306 }
1307 }
1308 match &options.transport {
1312 Transport::Tcp {
1313 connection_token: Some(t),
1314 ..
1315 }
1316 | Transport::External {
1317 connection_token: Some(t),
1318 ..
1319 } if t.is_empty() => {
1320 return Err(Error::with_message(
1321 ErrorKind::InvalidConfig,
1322 "invalid client configuration: connection_token must be a non-empty string",
1323 ));
1324 }
1325 _ => {}
1326 }
1327 let effective_connection_token: Option<String> = match &mut options.transport {
1332 Transport::Default => unreachable!("default transport resolved above"),
1333 Transport::Stdio | Transport::InProcess => None,
1334 Transport::Tcp {
1335 connection_token, ..
1336 } => Some(
1337 connection_token
1338 .get_or_insert_with(generate_connection_token)
1339 .clone(),
1340 ),
1341 Transport::External {
1342 connection_token, ..
1343 } => connection_token.clone(),
1344 };
1345 let session_fs_config = options.session_fs.clone();
1346 let request_handler = options.request_handler.clone();
1347 let extension_launch_provider = options.extension_launch_provider.clone();
1348 let session_fs_sqlite_declared = session_fs_config
1349 .as_ref()
1350 .and_then(|c| c.capabilities.as_ref())
1351 .is_some_and(|caps| caps.sqlite);
1352 let program = match &options.program {
1353 CliProgram::Path(path) => {
1354 info!(path = %path.display(), "using explicit copilot CLI path");
1355 path.clone()
1356 }
1357 CliProgram::Resolve => {
1358 let resolve_start = Instant::now();
1359 let resolved = resolve::copilot_binary_with_extract_dir(
1360 options.bundled_cli_extract_dir.as_deref(),
1361 true,
1362 )?;
1363 let resolve_elapsed = resolve_start.elapsed();
1364 timings.program_resolve_ms = Some(StartupTimings::millis(resolve_elapsed));
1365 debug!(
1366 elapsed_ms = resolve_elapsed.as_millis(),
1367 "Client::start CLI program resolution complete"
1368 );
1369 info!(path = %resolved.display(), "resolved copilot runtime");
1370 #[cfg(windows)]
1371 {
1372 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1373 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1374 }) {
1375 warn!(
1376 path = %resolved.display(),
1377 ext = %ext,
1378 "resolved copilot CLI is a .cmd/.bat wrapper; \
1379 this may cause console window flashes on Windows"
1380 );
1381 }
1382 }
1383 resolved
1384 }
1385 };
1386 let working_directory = {
1387 let cwd = options.working_directory.clone();
1388 if cwd.as_os_str().is_empty() {
1389 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1390 } else {
1391 cwd
1392 }
1393 };
1394
1395 let transport_setup_start = Instant::now();
1396 let client = match options.transport {
1397 Transport::Default => unreachable!("default transport resolved above"),
1398 Transport::External {
1399 ref host,
1400 port,
1401 connection_token: _,
1402 } => {
1403 info!(host = %host, port = %port, "connecting to external CLI server");
1404 let connect_start = Instant::now();
1405 let stream = TcpStream::connect((host.as_str(), port)).await?;
1406 debug!(
1407 elapsed_ms = connect_start.elapsed().as_millis(),
1408 host = %host,
1409 port,
1410 "Client::start TCP connect complete"
1411 );
1412 let (reader, writer) = tokio::io::split(stream);
1413 Self::from_transport(
1414 reader,
1415 writer,
1416 None,
1417 None,
1418 working_directory,
1419 options.on_list_models,
1420 extension_launch_provider.clone(),
1421 session_fs_config.is_some(),
1422 session_fs_sqlite_declared,
1423 options.on_get_trace_context,
1424 options.on_github_telemetry,
1425 effective_connection_token.clone(),
1426 options.mode,
1427 options.client_info,
1428 )?
1429 }
1430 Transport::Tcp {
1431 port,
1432 connection_token: _,
1433 } => {
1434 let (mut child, tree, actual_port, spawn_elapsed, port_wait_elapsed) =
1435 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1436 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1437 timings.port_wait_ms = Some(StartupTimings::millis(port_wait_elapsed));
1438 let connect_start = Instant::now();
1439 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1440 debug!(
1441 elapsed_ms = connect_start.elapsed().as_millis(),
1442 port = actual_port,
1443 "Client::start TCP connect complete"
1444 );
1445 let (reader, writer) = tokio::io::split(stream);
1446 Self::drain_stderr(&mut child);
1447 Self::from_transport(
1448 reader,
1449 writer,
1450 Some(child),
1451 tree,
1452 working_directory,
1453 options.on_list_models,
1454 extension_launch_provider.clone(),
1455 session_fs_config.is_some(),
1456 session_fs_sqlite_declared,
1457 options.on_get_trace_context,
1458 options.on_github_telemetry,
1459 effective_connection_token.clone(),
1460 options.mode,
1461 options.client_info,
1462 )?
1463 }
1464 Transport::Stdio => {
1465 let (mut child, tree, spawn_elapsed) =
1466 Self::spawn_stdio(&program, &options, &working_directory)?;
1467 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1468 let stdin = child.stdin.take().expect("stdin is piped");
1469 let stdout = child.stdout.take().expect("stdout is piped");
1470 Self::drain_stderr(&mut child);
1471 Self::from_transport(
1472 stdout,
1473 stdin,
1474 Some(child),
1475 tree,
1476 working_directory,
1477 options.on_list_models,
1478 extension_launch_provider.clone(),
1479 session_fs_config.is_some(),
1480 session_fs_sqlite_declared,
1481 options.on_get_trace_context,
1482 options.on_github_telemetry,
1483 effective_connection_token.clone(),
1484 options.mode,
1485 options.client_info,
1486 )?
1487 }
1488 Transport::InProcess => {
1489 #[cfg(feature = "bundled-in-process")]
1490 {
1491 info!(runtime_path = %program.display(), "hosting copilot runtime in-process (FFI)");
1492 let mut environment = Vec::new();
1493 if let Some(base_directory) = &options.base_directory {
1494 let value = base_directory.to_str().ok_or_else(|| {
1495 Error::with_message(
1496 ErrorKind::InvalidConfig,
1497 "base_directory must be valid UTF-8 for Transport::InProcess",
1498 )
1499 })?;
1500 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1501 }
1502 if options.mode == ClientMode::Empty {
1503 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1504 }
1505 if let Some(github_token) = &options.github_token {
1506 environment
1507 .push(("COPILOT_SDK_AUTH_TOKEN".to_string(), github_token.clone()));
1508 }
1509 let mut args = Vec::new();
1510 args.extend(
1511 Self::log_level_args(&options)
1512 .into_iter()
1513 .map(str::to_string),
1514 );
1515 args.extend(Self::session_idle_timeout_args(&options));
1516 args.extend(Self::remote_args(&options));
1517 if options.github_token.is_some() {
1518 args.extend([
1519 "--auth-token-env".to_string(),
1520 "COPILOT_SDK_AUTH_TOKEN".to_string(),
1521 ]);
1522 }
1523 let use_logged_in_user = options
1524 .use_logged_in_user
1525 .unwrap_or(options.github_token.is_none());
1526 if !use_logged_in_user {
1527 args.push("--no-auto-login".to_string());
1528 }
1529 let explicit_cli = std::env::var_os("COPILOT_CLI_PATH")
1530 .map(PathBuf::from)
1531 .filter(|path| path.is_file());
1532 let host = crate::ffi::FfiHost::create(
1533 &program,
1534 explicit_cli.as_deref(),
1535 environment,
1536 args,
1537 )?;
1538 let (reader, writer, shared) = host.start().await?;
1539 let client = Self::from_transport(
1540 reader,
1541 writer,
1542 None,
1543 None,
1544 working_directory,
1545 options.on_list_models,
1546 extension_launch_provider.clone(),
1547 session_fs_config.is_some(),
1548 session_fs_sqlite_declared,
1549 options.on_get_trace_context,
1550 options.on_github_telemetry,
1551 effective_connection_token.clone(),
1552 options.mode,
1553 options.client_info,
1554 )?;
1555 *client.inner.ffi_host.lock() = Some(shared);
1556 client
1557 }
1558 #[cfg(not(feature = "bundled-in-process"))]
1559 unreachable!("in-process feature validation returned above")
1560 }
1561 };
1562 timings.transport_setup_ms = StartupTimings::millis(transport_setup_start.elapsed());
1563 debug!(
1564 elapsed_ms = start_time.elapsed().as_millis(),
1565 "Client::start transport setup complete"
1566 );
1567 let handshake_start = Instant::now();
1568 client.verify_protocol_version().await?;
1569 timings.handshake_ms = StartupTimings::millis(handshake_start.elapsed());
1570 debug!(
1571 elapsed_ms = start_time.elapsed().as_millis(),
1572 "Client::start protocol verification complete"
1573 );
1574 let request_dispatcher = request_handler.map(|handler| {
1575 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1576 handler,
1577 ));
1578 dispatcher.set_client(Arc::downgrade(&client.inner));
1579 let _ = client.inner.llm_inference.set(dispatcher.clone());
1580 dispatcher
1581 });
1582 if client.inner.extension_launch_provider.is_configured() {
1583 client.inner.router.ensure_started(
1584 &client.inner.notification_tx,
1585 &client.inner.request_rx,
1586 client.inner.extension_launch_provider.clone(),
1587 request_dispatcher.clone(),
1588 client.inner.on_github_telemetry.clone(),
1589 client.inner.github_token_registry.clone(),
1590 );
1591 client.rpc().register_extension_launch_provider().await?;
1592 }
1593 if !builtin_plugin_directories.is_empty() {
1594 client
1595 .call(
1596 "plugins.builtin.set",
1597 Some(serde_json::json!({ "paths": builtin_plugin_directories })),
1598 )
1599 .await?;
1600 }
1601 if let Some(cfg) = session_fs_config {
1602 let session_fs_start = Instant::now();
1603 let capabilities = cfg.capabilities.as_ref().map(|c| {
1604 crate::generated::api_types::SessionFsSetProviderCapabilities {
1605 sqlite: Some(c.sqlite),
1606 }
1607 });
1608 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1609 capabilities,
1610 conventions: cfg.conventions.into_wire(),
1611 initial_cwd: cfg.initial_cwd,
1612 session_state_path: cfg.session_state_path,
1613 };
1614 client.rpc().session_fs().set_provider(request).await?;
1615 let session_fs_elapsed = session_fs_start.elapsed();
1616 timings.session_fs_ms = Some(StartupTimings::millis(session_fs_elapsed));
1617 debug!(
1618 elapsed_ms = session_fs_elapsed.as_millis(),
1619 "Client::start session filesystem setup complete"
1620 );
1621 }
1622 if let Some(dispatcher) = request_dispatcher {
1623 let llm_inference_start = Instant::now();
1624 client.inner.router.ensure_started(
1627 &client.inner.notification_tx,
1628 &client.inner.request_rx,
1629 client.inner.extension_launch_provider.clone(),
1630 Some(dispatcher.clone()),
1631 client.inner.on_github_telemetry.clone(),
1632 client.inner.github_token_registry.clone(),
1633 );
1634 client.rpc().llm_inference().set_provider().await?;
1635 let llm_inference_elapsed = llm_inference_start.elapsed();
1636 timings.llm_handler_ms = Some(StartupTimings::millis(llm_inference_elapsed));
1637 debug!(
1638 elapsed_ms = llm_inference_elapsed.as_millis(),
1639 "Client::start Copilot request handler registration complete"
1640 );
1641 }
1642 timings.total_ms = StartupTimings::millis(start_time.elapsed());
1643 let timings_span = tracing::debug_span!(
1646 "Client::start timings",
1647 program_resolve_ms = tracing::field::Empty,
1648 process_spawn_ms = tracing::field::Empty,
1649 port_wait_ms = tracing::field::Empty,
1650 transport_setup_ms = timings.transport_setup_ms,
1651 handshake_ms = timings.handshake_ms,
1652 session_fs_ms = tracing::field::Empty,
1653 llm_handler_ms = tracing::field::Empty,
1654 total_ms = timings.total_ms,
1655 );
1656 record_optional_millis(
1657 &timings_span,
1658 "program_resolve_ms",
1659 timings.program_resolve_ms,
1660 );
1661 record_optional_millis(&timings_span, "process_spawn_ms", timings.process_spawn_ms);
1662 record_optional_millis(&timings_span, "port_wait_ms", timings.port_wait_ms);
1663 record_optional_millis(&timings_span, "session_fs_ms", timings.session_fs_ms);
1664 record_optional_millis(&timings_span, "llm_handler_ms", timings.llm_handler_ms);
1665 timings_span.in_scope(|| debug!("Client::start timings"));
1666 let _ = client.inner.startup_timings.set(timings);
1667 debug!(
1668 elapsed_ms = start_time.elapsed().as_millis(),
1669 "Client::start complete"
1670 );
1671 Ok(client)
1672 }
1673
1674 pub fn from_streams(
1678 reader: impl AsyncRead + Unpin + Send + 'static,
1679 writer: impl AsyncWrite + Unpin + Send + 'static,
1680 cwd: PathBuf,
1681 ) -> Result<Self> {
1682 Self::from_transport(
1683 reader,
1684 writer,
1685 None,
1686 None,
1687 cwd,
1688 None,
1689 None,
1690 false,
1691 false,
1692 None,
1693 None,
1694 None,
1695 ClientMode::default(),
1696 None,
1697 )
1698 }
1699
1700 #[doc(hidden)]
1703 #[cfg(any(test, feature = "test-support"))]
1704 pub fn from_streams_with_extension_launch_provider(
1705 reader: impl AsyncRead + Unpin + Send + 'static,
1706 writer: impl AsyncWrite + Unpin + Send + 'static,
1707 cwd: PathBuf,
1708 provider: Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1709 ) -> Result<Self> {
1710 Self::from_transport(
1711 reader,
1712 writer,
1713 None,
1714 None,
1715 cwd,
1716 None,
1717 Some(provider),
1718 false,
1719 false,
1720 None,
1721 None,
1722 None,
1723 ClientMode::default(),
1724 None,
1725 )
1726 }
1727
1728 #[cfg(any(test, feature = "test-support"))]
1736 pub fn from_streams_with_trace_provider(
1737 reader: impl AsyncRead + Unpin + Send + 'static,
1738 writer: impl AsyncWrite + Unpin + Send + 'static,
1739 cwd: PathBuf,
1740 provider: Arc<dyn TraceContextProvider>,
1741 ) -> Result<Self> {
1742 Self::from_transport(
1743 reader,
1744 writer,
1745 None,
1746 None,
1747 cwd,
1748 None,
1749 None,
1750 false,
1751 false,
1752 Some(provider),
1753 None,
1754 None,
1755 ClientMode::default(),
1756 None,
1757 )
1758 }
1759
1760 #[cfg(any(test, feature = "test-support"))]
1764 pub fn from_streams_with_connection_token(
1765 reader: impl AsyncRead + Unpin + Send + 'static,
1766 writer: impl AsyncWrite + Unpin + Send + 'static,
1767 cwd: PathBuf,
1768 token: Option<String>,
1769 ) -> Result<Self> {
1770 Self::from_transport(
1771 reader,
1772 writer,
1773 None,
1774 None,
1775 cwd,
1776 None,
1777 None,
1778 false,
1779 false,
1780 None,
1781 None,
1782 token,
1783 ClientMode::default(),
1784 None,
1785 )
1786 }
1787
1788 #[doc(hidden)]
1791 #[cfg(any(test, feature = "test-support"))]
1792 pub fn from_streams_with_github_telemetry(
1793 reader: impl AsyncRead + Unpin + Send + 'static,
1794 writer: impl AsyncWrite + Unpin + Send + 'static,
1795 cwd: PathBuf,
1796 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1797 ) -> Result<Self> {
1798 Self::from_transport(
1799 reader,
1800 writer,
1801 None,
1802 None,
1803 cwd,
1804 None,
1805 None,
1806 false,
1807 false,
1808 None,
1809 Some(on_github_telemetry),
1810 None,
1811 ClientMode::default(),
1812 None,
1813 )
1814 }
1815
1816 #[cfg(any(test, feature = "test-support"))]
1822 pub fn generate_connection_token_for_test() -> String {
1823 generate_connection_token()
1824 }
1825
1826 #[doc(hidden)]
1830 #[cfg(any(test, feature = "test-support"))]
1831 pub fn from_streams_with_client_info(
1832 reader: impl AsyncRead + Unpin + Send + 'static,
1833 writer: impl AsyncWrite + Unpin + Send + 'static,
1834 cwd: PathBuf,
1835 client_info: Option<ClientInfo>,
1836 ) -> Result<Self> {
1837 Self::from_transport(
1838 reader,
1839 writer,
1840 None,
1841 None,
1842 cwd,
1843 None,
1844 None,
1845 false,
1846 false,
1847 None,
1848 None,
1849 None,
1850 ClientMode::default(),
1851 client_info,
1852 )
1853 }
1854
1855 #[allow(clippy::too_many_arguments)]
1856 fn from_transport(
1857 reader: impl AsyncRead + Unpin + Send + 'static,
1858 writer: impl AsyncWrite + Unpin + Send + 'static,
1859 child: Option<Child>,
1860 process_tree: Option<process_tree::ProcessTree>,
1861 cwd: PathBuf,
1862 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1863 extension_launch_provider: Option<
1864 Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1865 >,
1866 session_fs_configured: bool,
1867 session_fs_sqlite_declared: bool,
1868 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1869 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1870 effective_connection_token: Option<String>,
1871 mode: ClientMode,
1872 client_info: Option<ClientInfo>,
1873 ) -> Result<Self> {
1874 let setup_start = Instant::now();
1875 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1876 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1877 let rpc = JsonRpcClient::new(
1878 writer,
1879 reader,
1880 notification_broadcast_tx.clone(),
1881 request_tx,
1882 );
1883
1884 let pid = child.as_ref().and_then(|c| c.id());
1885 info!(pid = ?pid, "copilot CLI client ready");
1886
1887 let github_token_registry = Arc::new(github_token::GitHubTokenRegistry::new());
1888 let extension_launch_provider = Arc::new(
1889 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(
1890 extension_launch_provider,
1891 ),
1892 );
1893 let client = Self {
1894 inner: Arc::new(ClientInner {
1895 child: parking_lot::Mutex::new(child),
1896 process_tree: parking_lot::Mutex::new(process_tree),
1897 #[cfg(feature = "bundled-in-process")]
1898 ffi_host: parking_lot::Mutex::new(None),
1899 rpc,
1900 cwd,
1901 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1902 notification_tx: notification_broadcast_tx,
1903 router: router::SessionRouter::new(),
1904 github_token_registry: github_token_registry.clone(),
1905 negotiated_protocol_version: OnceLock::new(),
1906 state: parking_lot::Mutex::new(ConnectionState::Connected),
1907 lifecycle_tx: broadcast::channel(256).0,
1908 on_list_models,
1909 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1910 session_fs_configured,
1911 session_fs_sqlite_declared,
1912 llm_inference: OnceLock::new(),
1913 extension_launch_provider: extension_launch_provider.clone(),
1914 on_github_telemetry,
1915 on_get_trace_context,
1916 effective_connection_token,
1917 mode,
1918 client_info,
1919 startup_timings: OnceLock::new(),
1920 }),
1921 };
1922 github_token_registry.set_client(Arc::downgrade(&client.inner));
1923 extension_launch_provider.set_client(Arc::downgrade(&client.inner));
1924 client.spawn_lifecycle_dispatcher();
1925 debug!(
1926 elapsed_ms = setup_start.elapsed().as_millis(),
1927 pid = ?pid,
1928 "Client::from_transport setup complete"
1929 );
1930 Ok(client)
1931 }
1932
1933 fn spawn_lifecycle_dispatcher(&self) {
1937 let mut notif_rx = self.inner.notification_tx.subscribe();
1938 let lifecycle_tx = self.inner.lifecycle_tx.clone();
1939 tokio::spawn(async move {
1940 loop {
1941 match notif_rx.recv().await {
1942 Ok(notification) => {
1943 if notification.method != "session.lifecycle" {
1944 continue;
1945 }
1946 let Some(params) = notification.params.as_ref() else {
1947 continue;
1948 };
1949 let event: SessionLifecycleEvent =
1950 match serde_json::from_value(params.clone()) {
1951 Ok(e) => e,
1952 Err(e) => {
1953 warn!(
1954 error = %e,
1955 "failed to deserialize session.lifecycle notification"
1956 );
1957 continue;
1958 }
1959 };
1960 let _ = lifecycle_tx.send(event);
1963 }
1964 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
1965 warn!(missed = n, "lifecycle dispatcher lagged");
1966 }
1967 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
1968 }
1969 }
1970 });
1971 }
1972
1973 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
1974 let mut command = Command::new(program);
1975 command.kill_on_drop(true);
1976 for arg in &options.prefix_args {
1977 command.arg(arg);
1978 }
1979 if let Some(token) = &options.github_token {
1982 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1983 }
1984 if let Some(telemetry) = &options.telemetry {
1987 command.env("COPILOT_OTEL_ENABLED", "true");
1988 if let Some(endpoint) = &telemetry.otlp_endpoint {
1989 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1990 }
1991 if let Some(protocol) = telemetry.otlp_protocol {
1992 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
1993 }
1994 if let Some(path) = &telemetry.file_path {
1995 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
1996 }
1997 if let Some(exporter) = telemetry.exporter_type {
1998 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
1999 }
2000 if let Some(source) = &telemetry.source_name {
2001 command.env("COPILOT_OTEL_SOURCE_NAME", source);
2002 }
2003 if let Some(capture) = telemetry.capture_content {
2004 command.env(
2005 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2006 if capture { "true" } else { "false" },
2007 );
2008 }
2009 }
2010 if let Some(dir) = &options.base_directory {
2011 command.env("COPILOT_HOME", dir);
2012 }
2013 if options.mode == ClientMode::Empty {
2016 command.env("COPILOT_DISABLE_KEYTAR", "1");
2017 }
2018 if let Transport::Tcp {
2019 connection_token: Some(token),
2020 ..
2021 } = &options.transport
2022 {
2023 command.env("COPILOT_CONNECTION_TOKEN", token);
2024 }
2025 for (key, value) in &options.env {
2026 command.env(key, value);
2027 }
2028 for key in &options.env_remove {
2029 command.env_remove(key);
2030 }
2031 command
2032 .current_dir(working_directory)
2033 .stdout(Stdio::piped())
2034 .stderr(Stdio::piped());
2035
2036 command
2037 }
2038
2039 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
2047 let mut args: Vec<&'static str> = Vec::new();
2048 if options.github_token.is_some() {
2049 args.push("--auth-token-env");
2050 args.push("COPILOT_SDK_AUTH_TOKEN");
2051 }
2052 let use_logged_in = options
2053 .use_logged_in_user
2054 .unwrap_or(options.github_token.is_none());
2055 if !use_logged_in {
2056 args.push("--no-auto-login");
2057 }
2058 args
2059 }
2060
2061 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
2065 match options.session_idle_timeout_seconds {
2066 Some(secs) if secs > 0 => {
2067 vec!["--session-idle-timeout".to_string(), secs.to_string()]
2068 }
2069 _ => Vec::new(),
2070 }
2071 }
2072
2073 fn remote_args(options: &ClientOptions) -> Vec<String> {
2074 if options.enable_remote_sessions {
2075 vec!["--remote".to_string()]
2076 } else {
2077 Vec::new()
2078 }
2079 }
2080
2081 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
2082 match options.log_level {
2083 Some(level) => vec!["--log-level", level.as_str()],
2084 None => Vec::new(),
2085 }
2086 }
2087
2088 fn spawn_stdio(
2089 program: &Path,
2090 options: &ClientOptions,
2091 working_directory: &Path,
2092 ) -> Result<(Child, Option<process_tree::ProcessTree>, Duration)> {
2093 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
2094 let mut command = Self::build_command(program, options, working_directory);
2095 command
2096 .args(["--server", "--stdio", "--no-auto-update"])
2097 .args(Self::log_level_args(options))
2098 .args(Self::auth_args(options))
2099 .args(Self::session_idle_timeout_args(options))
2100 .args(Self::remote_args(options))
2101 .args(&options.extra_args)
2102 .stdin(Stdio::piped());
2103 let spawn_start = Instant::now();
2104 let (child, tree) = process_tree::spawn(&mut command)?;
2105 let spawn_elapsed = spawn_start.elapsed();
2106 debug!(
2107 elapsed_ms = spawn_elapsed.as_millis(),
2108 "Client::spawn_stdio subprocess spawned"
2109 );
2110 Ok((child, tree, spawn_elapsed))
2111 }
2112
2113 async fn spawn_tcp(
2114 program: &Path,
2115 options: &ClientOptions,
2116 working_directory: &Path,
2117 port: u16,
2118 ) -> Result<(
2119 Child,
2120 Option<process_tree::ProcessTree>,
2121 u16,
2122 Duration,
2123 Duration,
2124 )> {
2125 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
2126 let mut command = Self::build_command(program, options, working_directory);
2127 command
2128 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
2129 .args(Self::log_level_args(options))
2130 .args(Self::auth_args(options))
2131 .args(Self::session_idle_timeout_args(options))
2132 .args(Self::remote_args(options))
2133 .args(&options.extra_args)
2134 .stdin(Stdio::null());
2135 let spawn_start = Instant::now();
2136 let (mut child, tree) = process_tree::spawn(&mut command)?;
2137 let spawn_elapsed = spawn_start.elapsed();
2138 debug!(
2139 elapsed_ms = spawn_elapsed.as_millis(),
2140 "Client::spawn_tcp subprocess spawned"
2141 );
2142 let stdout = child.stdout.take().expect("stdout is piped");
2143
2144 let (port_tx, port_rx) = oneshot::channel::<u16>();
2145 let span = tracing::error_span!("copilot_cli_port_scan");
2146 tokio::spawn(
2147 async move {
2148 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
2150 let mut lines = BufReader::new(stdout).lines();
2151 let mut port_tx = Some(port_tx);
2152 while let Ok(Some(line)) = lines.next_line().await {
2153 debug!(line = %line, "CLI stdout");
2154 if let Some(tx) = port_tx.take() {
2155 if let Some(caps) = port_re.captures(&line)
2156 && let Some(p) =
2157 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
2158 {
2159 let _ = tx.send(p);
2160 continue;
2161 }
2162 port_tx = Some(tx);
2164 }
2165 }
2166 }
2167 .instrument(span),
2168 );
2169
2170 let port_wait_start = Instant::now();
2171 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
2172 .await
2173 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
2174 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
2175
2176 let port_wait_elapsed = port_wait_start.elapsed();
2177 debug!(
2178 elapsed_ms = port_wait_elapsed.as_millis(),
2179 port = actual_port,
2180 "Client::spawn_tcp TCP port wait complete"
2181 );
2182 info!(port = %actual_port, "CLI server listening");
2183 Ok((child, tree, actual_port, spawn_elapsed, port_wait_elapsed))
2184 }
2185
2186 fn drain_stderr(child: &mut Child) {
2187 if let Some(stderr) = child.stderr.take() {
2188 let span = tracing::error_span!("copilot_cli");
2189 tokio::spawn(
2190 async move {
2191 let mut reader = BufReader::new(stderr).lines();
2192 while let Ok(Some(line)) = reader.next_line().await {
2193 warn!(line = %line, "CLI stderr");
2194 }
2195 }
2196 .instrument(span),
2197 );
2198 }
2199 }
2200
2201 pub fn cwd(&self) -> &PathBuf {
2203 &self.inner.cwd
2204 }
2205
2206 pub fn mode(&self) -> ClientMode {
2208 self.inner.mode
2209 }
2210
2211 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
2222 crate::generated::rpc::ClientRpc { client: self }
2223 }
2224
2225 #[allow(dead_code, reason = "convenience for future internal use")]
2227 pub(crate) async fn send_request(
2228 &self,
2229 method: &str,
2230 params: Option<serde_json::Value>,
2231 ) -> Result<JsonRpcResponse> {
2232 self.inner.rpc.send_request(method, params).await
2233 }
2234
2235 pub async fn call(
2255 &self,
2256 method: &str,
2257 params: Option<serde_json::Value>,
2258 ) -> Result<serde_json::Value> {
2259 self.call_with_inline_callback(method, params, None).await
2260 }
2261
2262 pub(crate) async fn call_with_inline_callback(
2277 &self,
2278 method: &str,
2279 params: Option<serde_json::Value>,
2280 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
2281 ) -> Result<serde_json::Value> {
2282 let session_id: Option<SessionId> = params
2283 .as_ref()
2284 .and_then(|p| p.get("sessionId"))
2285 .and_then(|v| v.as_str())
2286 .map(SessionId::from);
2287 let response = self
2288 .inner
2289 .rpc
2290 .send_request_with_inline_callback(method, params, inline_callback)
2291 .await?;
2292 if let Some(err) = response.error {
2293 if err.message.contains("Session not found") {
2294 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
2295 session_id.unwrap_or_else(|| "unknown".into()),
2296 ))
2297 .into());
2298 }
2299 return Err(Error::with_message(
2300 ErrorKind::Rpc { code: err.code },
2301 err.message,
2302 ));
2303 }
2304 Ok(response.result.unwrap_or(serde_json::Value::Null))
2305 }
2306
2307 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
2309 self.inner.rpc.write(response).await
2310 }
2311
2312 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
2314 Self { inner }
2315 }
2316
2317 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
2321 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
2322 self.inner.request_rx.lock().take()
2323 }
2324
2325 pub(crate) fn register_session(
2333 &self,
2334 session_id: &SessionId,
2335 ) -> crate::router::SessionChannels {
2336 self.inner.router.ensure_started(
2337 &self.inner.notification_tx,
2338 &self.inner.request_rx,
2339 self.inner.extension_launch_provider.clone(),
2340 self.inner.llm_inference.get().cloned(),
2341 self.inner.on_github_telemetry.clone(),
2342 self.inner.github_token_registry.clone(),
2343 );
2344 self.inner.router.register(session_id)
2345 }
2346
2347 pub(crate) fn unregister_session(&self, session_id: &SessionId) {
2349 self.inner.router.unregister(session_id);
2350 }
2351
2352 pub(crate) fn register_github_token_provider(
2353 &self,
2354 provider: Arc<dyn GitHubTokenProvider>,
2355 ) -> github_token::GitHubTokenRegistration {
2356 self.inner.router.ensure_started(
2357 &self.inner.notification_tx,
2358 &self.inner.request_rx,
2359 self.inner.extension_launch_provider.clone(),
2360 self.inner.llm_inference.get().cloned(),
2361 self.inner.on_github_telemetry.clone(),
2362 self.inner.github_token_registry.clone(),
2363 );
2364 let id = self.inner.github_token_registry.register(provider);
2365 github_token::GitHubTokenRegistration::new(self.inner.github_token_registry.clone(), id)
2366 }
2367
2368 pub(crate) fn retire_github_token_provider(&self, session_id: &SessionId) {
2369 self.inner.github_token_registry.retire_session(session_id);
2370 }
2371
2372 pub fn protocol_version(&self) -> Option<u32> {
2379 self.inner.negotiated_protocol_version.get().copied()
2380 }
2381
2382 pub fn startup_timings(&self) -> Option<StartupTimings> {
2389 self.inner.startup_timings.get().cloned()
2390 }
2391
2392 pub async fn verify_protocol_version(&self) -> Result<()> {
2416 let handshake_start = Instant::now();
2417 let mut used_fallback_ping = false;
2418 let server_version = match self.connect_handshake().await {
2422 Ok(v) => v,
2423 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2424 used_fallback_ping = true;
2425 self.ping(None).await?.protocol_version
2426 }
2427 Err(e) => return Err(e),
2428 };
2429
2430 match server_version {
2431 None => {
2432 warn!("CLI server did not report protocolVersion; skipping version check");
2433 }
2434 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2435 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2436 server: v,
2437 min: MIN_PROTOCOL_VERSION,
2438 max: SDK_PROTOCOL_VERSION,
2439 })
2440 .into());
2441 }
2442 Some(v) => {
2443 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2444 if existing != v {
2445 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2446 previous: existing,
2447 current: v,
2448 })
2449 .into());
2450 }
2451 } else {
2452 let _ = self.inner.negotiated_protocol_version.set(v);
2453 }
2454 }
2455 }
2456
2457 debug!(
2458 elapsed_ms = handshake_start.elapsed().as_millis(),
2459 protocol_version = ?server_version,
2460 used_fallback_ping,
2461 "Client::verify_protocol_version protocol handshake complete"
2462 );
2463 Ok(())
2464 }
2465
2466 async fn connect_handshake(&self) -> Result<Option<u32>> {
2473 let params = crate::generated::api_types::ConnectRequest {
2474 token: self.inner.effective_connection_token.clone(),
2475 enable_git_hub_telemetry_forwarding: self
2476 .inner
2477 .on_github_telemetry
2478 .is_some()
2479 .then_some(true),
2480 client_info: self
2485 .inner
2486 .client_info
2487 .as_ref()
2488 .and_then(ClientInfo::to_wire),
2489 };
2490 let value = self
2491 .call(
2492 crate::generated::api_types::rpc_methods::CONNECT,
2493 Some(serde_json::to_value(params)?),
2494 )
2495 .await?;
2496 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2497 Ok(Some(u32::try_from(result.protocol_version).map_err(
2498 |_| ProtocolErrorKind::InvalidProtocolVersion {
2499 server: result.protocol_version,
2500 },
2501 )?))
2502 }
2503
2504 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2512 let params = match message {
2513 Some(m) => serde_json::json!({ "message": m }),
2514 None => serde_json::json!({}),
2515 };
2516 let value = self
2517 .call(generated::api_types::rpc_methods::PING, Some(params))
2518 .await?;
2519 Ok(serde_json::from_value(value)?)
2520 }
2521
2522 pub async fn list_sessions(
2525 &self,
2526 filter: Option<SessionListFilter>,
2527 ) -> Result<Vec<SessionMetadata>> {
2528 let params = match filter {
2529 Some(f) => serde_json::json!({ "filter": f }),
2530 None => serde_json::json!({}),
2531 };
2532 let result = self.call("session.list", Some(params)).await?;
2533 let response: ListSessionsResponse = serde_json::from_value(result)?;
2534 Ok(response.sessions)
2535 }
2536
2537 pub async fn get_session_metadata(
2555 &self,
2556 session_id: &SessionId,
2557 ) -> Result<Option<SessionMetadata>> {
2558 let result = self
2559 .call(
2560 "session.getMetadata",
2561 Some(serde_json::json!({ "sessionId": session_id })),
2562 )
2563 .await?;
2564 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2565 Ok(response.session)
2566 }
2567
2568 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2570 self.call(
2571 "session.delete",
2572 Some(serde_json::json!({ "sessionId": session_id })),
2573 )
2574 .await?;
2575 self.retire_github_token_provider(session_id);
2576 Ok(())
2577 }
2578
2579 #[cfg(feature = "test-support")]
2582 #[doc(hidden)]
2583 pub fn start_router_for_test(&self) {
2584 self.inner.router.ensure_started(
2585 &self.inner.notification_tx,
2586 &self.inner.request_rx,
2587 self.inner.extension_launch_provider.clone(),
2588 self.inner.llm_inference.get().cloned(),
2589 self.inner.on_github_telemetry.clone(),
2590 self.inner.github_token_registry.clone(),
2591 );
2592 }
2593
2594 #[cfg(feature = "test-support")]
2595 #[doc(hidden)]
2596 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2599 let mut first_error = None;
2600
2601 for session_id in self.inner.router.session_ids() {
2602 if let Err(error) = self
2603 .call(
2604 "session.destroy",
2605 Some(serde_json::json!({ "sessionId": session_id })),
2606 )
2607 .await
2608 && first_error.is_none()
2609 {
2610 first_error = Some(error);
2611 }
2612 self.inner.router.unregister(&session_id);
2613 }
2614 self.inner.github_token_registry.clear();
2615
2616 match self.list_sessions(None).await {
2617 Ok(sessions) => {
2618 for session in sessions {
2619 if let Err(error) = self.delete_session(&session.session_id).await
2620 && first_error.is_none()
2621 {
2622 first_error = Some(error);
2623 }
2624 }
2625 }
2626 Err(error) if first_error.is_none() => first_error = Some(error),
2627 Err(_) => {}
2628 }
2629
2630 match first_error {
2631 Some(error) => Err(error),
2632 None => Ok(()),
2633 }
2634 }
2635
2636 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2652 let result = self
2653 .call("session.getLastId", Some(serde_json::json!({})))
2654 .await?;
2655 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2656 Ok(response.session_id)
2657 }
2658
2659 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2664 let result = self
2665 .call("session.getForeground", Some(serde_json::json!({})))
2666 .await?;
2667 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2668 Ok(response.session_id)
2669 }
2670
2671 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2676 self.call(
2677 "session.setForeground",
2678 Some(serde_json::json!({ "sessionId": session_id })),
2679 )
2680 .await?;
2681 Ok(())
2682 }
2683
2684 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2686 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2687 Ok(serde_json::from_value(result)?)
2688 }
2689
2690 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2692 let result = self
2693 .call("auth.getStatus", Some(serde_json::json!({})))
2694 .await?;
2695 Ok(serde_json::from_value(result)?)
2696 }
2697
2698 pub async fn list_models(&self) -> Result<Vec<Model>> {
2703 let cache = self.inner.models_cache.lock().clone();
2704 let models = cache
2705 .get_or_try_init(|| async {
2706 if let Some(handler) = &self.inner.on_list_models {
2707 handler.list_models().await
2708 } else {
2709 Ok(self.rpc().models().list().await?.models)
2710 }
2711 })
2712 .await?;
2713 Ok(models.clone())
2714 }
2715
2716 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2719 if let Some(provider) = &self.inner.on_get_trace_context {
2720 provider.get_trace_context().await
2721 } else {
2722 TraceContext::default()
2723 }
2724 }
2725
2726 pub fn pid(&self) -> Option<u32> {
2728 self.inner.child.lock().as_ref().and_then(|c| c.id())
2729 }
2730
2731 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2759 let pid = self.pid();
2760 info!(pid = ?pid, "stopping CLI process");
2761 let mut errors: Vec<Error> = Vec::new();
2762 self.inner.extension_launch_provider.clear();
2763
2764 for session_id in self.inner.router.session_ids() {
2767 match self
2768 .call(
2769 "session.destroy",
2770 Some(serde_json::json!({ "sessionId": session_id })),
2771 )
2772 .await
2773 {
2774 Ok(_) => {}
2775 Err(e) => {
2776 warn!(
2777 session_id = %session_id,
2778 error = %e,
2779 "session.destroy failed during Client::stop",
2780 );
2781 errors.push(e);
2782 }
2783 }
2784 self.inner.router.unregister(&session_id);
2785 }
2786 self.inner.github_token_registry.clear();
2787
2788 let should_shutdown_runtime = self.inner.child.lock().is_some();
2789 #[cfg(feature = "bundled-in-process")]
2790 let should_shutdown_runtime =
2791 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2792 if should_shutdown_runtime {
2793 let runtime_shutdown_start = Instant::now();
2794 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2795 .await
2796 {
2797 Ok(Ok(())) => {
2798 debug!(
2799 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2800 "Client::stop runtime shutdown complete"
2801 );
2802 }
2803 Ok(Err(e)) => {
2804 warn!(
2805 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2806 error = %e,
2807 "runtime.shutdown failed during Client::stop",
2808 );
2809 errors.push(e);
2810 }
2811 Err(_) => {
2812 let e = std::io::Error::new(
2813 std::io::ErrorKind::TimedOut,
2814 "runtime.shutdown timed out during Client::stop",
2815 );
2816 warn!(
2817 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2818 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2819 error = %e,
2820 "runtime.shutdown timed out during Client::stop",
2821 );
2822 errors.push(e.into());
2823 }
2824 }
2825 }
2826
2827 let child = self.inner.child.lock().take();
2828 let process_tree = self.inner.process_tree.lock().take();
2829 *self.inner.state.lock() = ConnectionState::Disconnected;
2830 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2831 if let Some(process_tree) = process_tree
2832 && let Err(error) = process_tree.terminate()
2833 {
2834 errors.push(error.into());
2835 }
2836 if let Some(mut child) = child {
2837 match child.try_wait() {
2838 Ok(Some(_status)) => {}
2839 Ok(None) => {
2840 if let Err(e) = child.kill().await {
2847 errors.push(e.into());
2848 }
2849 }
2850 Err(e) => errors.push(e.into()),
2851 }
2852 }
2853
2854 self.inner.rpc.force_close();
2857 #[cfg(feature = "bundled-in-process")]
2858 {
2859 if let Some(host) = self.inner.ffi_host.lock().take() {
2860 host.close();
2861 }
2862 }
2863
2864 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2865 if errors.is_empty() {
2866 Ok(())
2867 } else {
2868 Err(StopErrors(errors))
2869 }
2870 }
2871
2872 pub fn force_stop(&self) {
2901 let pid = self.pid();
2902 info!(pid = ?pid, "force-stopping CLI process");
2903 self.inner.extension_launch_provider.clear();
2904 if let Some(process_tree) = self.inner.process_tree.lock().take()
2905 && let Err(error) = process_tree.terminate()
2906 {
2907 error!(pid = ?pid, %error, "failed to terminate CLI process tree");
2908 }
2909 if let Some(mut child) = self.inner.child.lock().take()
2910 && let Err(e) = child.start_kill()
2911 {
2912 error!(pid = ?pid, error = %e, "failed to send kill signal");
2913 }
2914 self.inner.rpc.force_close();
2915 #[cfg(feature = "bundled-in-process")]
2916 {
2917 if let Some(host) = self.inner.ffi_host.lock().take() {
2918 host.close();
2919 }
2920 }
2921 self.inner.router.clear();
2924 self.inner.github_token_registry.clear();
2925 *self.inner.state.lock() = ConnectionState::Disconnected;
2926 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2927 }
2928
2929 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2964 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2965 }
2966}
2967
2968impl Drop for ClientInner {
2969 fn drop(&mut self) {
2970 let pid = self.child.lock().as_ref().and_then(Child::id);
2971 if let Some(process_tree) = self.process_tree.lock().take()
2972 && let Err(error) = process_tree.terminate()
2973 {
2974 error!(pid = ?pid, %error, "failed to terminate CLI process tree on drop");
2975 }
2976 if let Some(ref mut child) = *self.child.lock() {
2977 if let Err(e) = child.start_kill() {
2978 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2979 } else {
2980 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2981 }
2982 }
2983 #[cfg(feature = "bundled-in-process")]
2984 {
2985 if let Some(host) = self.ffi_host.lock().take() {
2986 self.rpc.force_close();
2987 host.close();
2988 }
2989 }
2990 }
2991}
2992
2993#[cfg(test)]
2994mod tests {
2995 use super::*;
2996
2997 #[test]
2998 fn is_transport_failure_matches_request_cancelled() {
2999 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
3000 assert!(err.is_transport_failure());
3001 }
3002
3003 #[test]
3004 fn is_transport_failure_matches_io_error() {
3005 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
3006 assert!(err.is_transport_failure());
3007 }
3008
3009 #[test]
3010 fn is_transport_failure_rejects_rpc_error() {
3011 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
3012 assert!(!err.is_transport_failure());
3013 }
3014
3015 #[test]
3016 fn is_transport_failure_rejects_session_error() {
3017 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
3018 assert!(!err.is_transport_failure());
3019 }
3020
3021 #[test]
3022 fn client_options_builder_composes() {
3023 let opts = ClientOptions::new()
3024 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
3025 .with_prefix_args(["node"])
3026 .with_cwd(PathBuf::from("/tmp"))
3027 .with_env([("KEY", "value")])
3028 .with_env_remove(["UNWANTED"])
3029 .with_extra_args(["--quiet"])
3030 .with_github_token("ghp_test")
3031 .with_use_logged_in_user(false)
3032 .with_log_level(LogLevel::Debug)
3033 .with_session_idle_timeout_seconds(120)
3034 .with_enable_remote_sessions(true);
3035 assert!(matches!(opts.program, CliProgram::Path(_)));
3036 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
3037 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
3038 assert_eq!(
3039 opts.env,
3040 vec![(
3041 std::ffi::OsString::from("KEY"),
3042 std::ffi::OsString::from("value")
3043 )]
3044 );
3045 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
3046 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
3047 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
3048 assert_eq!(opts.use_logged_in_user, Some(false));
3049 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
3050 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
3051 assert!(opts.enable_remote_sessions);
3052 }
3053
3054 #[test]
3055 fn default_transport_values_resolve_without_process_state() {
3056 assert!(matches!(
3057 resolve_default_transport_value(None).unwrap(),
3058 Transport::Stdio
3059 ));
3060 assert!(matches!(
3061 resolve_default_transport_value(Some("stdio")).unwrap(),
3062 Transport::Stdio
3063 ));
3064 assert!(matches!(
3065 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
3066 Transport::InProcess
3067 ));
3068 assert!(resolve_default_transport_value(Some("tcp")).is_err());
3069 }
3070
3071 #[test]
3072 fn inprocess_rejects_process_scoped_options() {
3073 let invalid = [
3074 ClientOptions::new().with_cwd("."),
3075 ClientOptions::new().with_env([("KEY", "value")]),
3076 ClientOptions::new().with_env_remove(["KEY"]),
3077 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
3078 ClientOptions::new().with_prefix_args(["index.js"]),
3079 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
3080 ClientOptions::new().with_extra_args(["--verbose"]),
3081 ];
3082
3083 for options in invalid {
3084 assert!(validate_inprocess_options(&options).is_err());
3085 }
3086 }
3087
3088 #[test]
3089 fn inprocess_allows_typed_runtime_options() {
3090 let options = ClientOptions::new()
3091 .with_base_directory("state")
3092 .with_log_level(LogLevel::Debug)
3093 .with_session_idle_timeout_seconds(10)
3094 .with_github_token("token")
3095 .with_use_logged_in_user(false)
3096 .with_enable_remote_sessions(true);
3097
3098 assert!(validate_inprocess_options(&options).is_ok());
3099 }
3100
3101 #[cfg(not(feature = "bundled-in-process"))]
3102 #[tokio::test]
3103 async fn inprocess_requires_cargo_feature() {
3104 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
3105 .await
3106 .unwrap_err();
3107
3108 assert!(error.to_string().contains("bundled-in-process"));
3109 }
3110
3111 #[test]
3112 fn is_transport_failure_rejects_other_protocol_errors() {
3113 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
3114 assert!(!err.is_transport_failure());
3115 }
3116
3117 #[test]
3118 fn build_command_lets_env_remove_strip_injected_token() {
3119 let opts = ClientOptions {
3120 github_token: Some("secret".to_string()),
3121 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
3122 ..Default::default()
3123 };
3124 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3125 let action = cmd
3127 .as_std()
3128 .get_envs()
3129 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3130 .map(|(_, v)| v);
3131 assert_eq!(
3132 action,
3133 Some(None),
3134 "env_remove should win over github_token"
3135 );
3136 }
3137
3138 #[test]
3139 fn build_command_lets_env_override_injected_token() {
3140 let opts = ClientOptions {
3141 github_token: Some("from-options".to_string()),
3142 env: vec![(
3143 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
3144 std::ffi::OsString::from("from-env"),
3145 )],
3146 ..Default::default()
3147 };
3148 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3149 let value = cmd
3150 .as_std()
3151 .get_envs()
3152 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3153 .and_then(|(_, v)| v);
3154 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
3155 }
3156
3157 #[test]
3158 fn build_command_injects_github_token_by_default() {
3159 let opts = ClientOptions {
3160 github_token: Some("just-the-token".to_string()),
3161 ..Default::default()
3162 };
3163 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3164 let value = cmd
3165 .as_std()
3166 .get_envs()
3167 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3168 .and_then(|(_, v)| v);
3169 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
3170 }
3171
3172 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
3173 cmd.as_std()
3174 .get_envs()
3175 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
3176 .and_then(|(_, v)| v)
3177 }
3178
3179 #[test]
3180 fn telemetry_config_builder_composes() {
3181 let cfg = TelemetryConfig::new()
3182 .with_otlp_endpoint("http://collector:4318")
3183 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
3184 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
3185 .with_exporter_type(OtelExporterType::OtlpHttp)
3186 .with_source_name("my-app")
3187 .with_capture_content(true);
3188
3189 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
3190 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
3191 assert_eq!(
3192 cfg.file_path.as_deref(),
3193 Some(Path::new("/var/log/copilot.jsonl")),
3194 );
3195 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
3196 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
3197 assert_eq!(cfg.capture_content, Some(true));
3198 assert!(!cfg.is_empty());
3199 assert!(TelemetryConfig::new().is_empty());
3200 }
3201
3202 #[test]
3203 fn otlp_http_protocol_serde_matches_env_value() {
3204 for (protocol, wire) in [
3205 (OtlpHttpProtocol::HttpJson, "http/json"),
3206 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
3207 ] {
3208 assert_eq!(protocol.as_str(), wire);
3209
3210 let serialized = serde_json::to_string(&protocol).unwrap();
3211 assert_eq!(serialized, format!("\"{wire}\""));
3212
3213 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
3214 assert_eq!(deserialized, protocol);
3215 }
3216 }
3217
3218 #[test]
3219 fn build_command_sets_otel_env_when_telemetry_enabled() {
3220 let opts = ClientOptions {
3221 telemetry: Some(TelemetryConfig {
3222 otlp_endpoint: Some("http://collector:4318".to_string()),
3223 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
3224 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
3225 exporter_type: Some(OtelExporterType::OtlpHttp),
3226 source_name: Some("my-app".to_string()),
3227 capture_content: Some(true),
3228 }),
3229 ..Default::default()
3230 };
3231 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3232 assert_eq!(
3233 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
3234 Some(std::ffi::OsStr::new("true")),
3235 );
3236 assert_eq!(
3237 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3238 Some(std::ffi::OsStr::new("http://collector:4318")),
3239 );
3240 assert_eq!(
3241 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
3242 Some(std::ffi::OsStr::new("http/protobuf")),
3243 );
3244 assert_eq!(
3245 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
3246 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
3247 );
3248 assert_eq!(
3249 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
3250 Some(std::ffi::OsStr::new("otlp-http")),
3251 );
3252 assert_eq!(
3253 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
3254 Some(std::ffi::OsStr::new("my-app")),
3255 );
3256 assert_eq!(
3257 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
3258 Some(std::ffi::OsStr::new("true")),
3259 );
3260 }
3261
3262 #[test]
3263 fn build_command_omits_otel_env_when_telemetry_none() {
3264 let opts = ClientOptions::default();
3265 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3266 for key in [
3267 "COPILOT_OTEL_ENABLED",
3268 "OTEL_EXPORTER_OTLP_ENDPOINT",
3269 "OTEL_EXPORTER_OTLP_PROTOCOL",
3270 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3271 "COPILOT_OTEL_EXPORTER_TYPE",
3272 "COPILOT_OTEL_SOURCE_NAME",
3273 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3274 ] {
3275 assert!(
3276 env_value(&cmd, key).is_none(),
3277 "expected {key} to be unset when telemetry is None",
3278 );
3279 }
3280 }
3281
3282 #[test]
3283 fn build_command_omits_unset_telemetry_fields() {
3284 let opts = ClientOptions {
3285 telemetry: Some(TelemetryConfig {
3286 otlp_endpoint: Some("http://collector:4318".to_string()),
3287 ..Default::default()
3288 }),
3289 ..Default::default()
3290 };
3291 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3292 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 for key in [
3303 "OTEL_EXPORTER_OTLP_PROTOCOL",
3304 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3305 "COPILOT_OTEL_EXPORTER_TYPE",
3306 "COPILOT_OTEL_SOURCE_NAME",
3307 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3308 ] {
3309 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
3310 }
3311 }
3312
3313 #[test]
3314 fn build_command_lets_user_env_override_telemetry() {
3315 let opts = ClientOptions {
3316 telemetry: Some(TelemetryConfig {
3317 otlp_endpoint: Some("http://from-config:4318".to_string()),
3318 ..Default::default()
3319 }),
3320 env: vec![(
3321 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
3322 std::ffi::OsString::from("http://from-user-env:4318"),
3323 )],
3324 ..Default::default()
3325 };
3326 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3327 assert_eq!(
3328 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3329 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
3330 "user-supplied options.env should override telemetry config",
3331 );
3332 }
3333
3334 #[test]
3335 fn build_command_sets_copilot_home_env_when_configured() {
3336 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
3337 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3338 assert_eq!(
3339 env_value(&cmd, "COPILOT_HOME"),
3340 Some(std::ffi::OsStr::new("/custom/copilot")),
3341 );
3342
3343 let opts = ClientOptions::default();
3344 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3345 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
3346 }
3347
3348 #[test]
3349 fn build_command_sets_connection_token_env_when_configured() {
3350 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3351 port: 0,
3352 connection_token: Some("secret-token".to_string()),
3353 });
3354 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3355 assert_eq!(
3356 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3357 Some(std::ffi::OsStr::new("secret-token")),
3358 );
3359
3360 let opts = ClientOptions::default();
3361 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3362 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3363 }
3364
3365 #[tokio::test]
3366 async fn start_rejects_empty_connection_token() {
3367 let opts = ClientOptions::new()
3368 .with_transport(Transport::Tcp {
3369 port: 0,
3370 connection_token: Some(String::new()),
3371 })
3372 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3373 let err = Client::start(opts).await.unwrap_err();
3374 assert!(
3375 matches!(err.kind(), ErrorKind::InvalidConfig),
3376 "got {err:?}"
3377 );
3378 }
3379
3380 #[tokio::test]
3381 async fn start_rejects_empty_external_connection_token() {
3382 let opts = ClientOptions::new()
3383 .with_transport(Transport::External {
3384 host: "127.0.0.1".to_string(),
3385 port: 1,
3386 connection_token: Some(String::new()),
3387 })
3388 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3389 let err = Client::start(opts).await.unwrap_err();
3390 assert!(
3391 matches!(err.kind(), ErrorKind::InvalidConfig),
3392 "got {err:?}"
3393 );
3394 }
3395
3396 #[test]
3397 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3398 let opts_true = ClientOptions {
3399 telemetry: Some(TelemetryConfig {
3400 capture_content: Some(true),
3401 ..Default::default()
3402 }),
3403 ..Default::default()
3404 };
3405 let opts_false = ClientOptions {
3406 telemetry: Some(TelemetryConfig {
3407 capture_content: Some(false),
3408 ..Default::default()
3409 }),
3410 ..Default::default()
3411 };
3412 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3413 let cmd_false =
3414 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3415 assert_eq!(
3416 env_value(
3417 &cmd_true,
3418 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3419 ),
3420 Some(std::ffi::OsStr::new("true")),
3421 );
3422 assert_eq!(
3423 env_value(
3424 &cmd_false,
3425 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3426 ),
3427 Some(std::ffi::OsStr::new("false")),
3428 );
3429 }
3430
3431 #[test]
3432 fn session_idle_timeout_args_are_omitted_by_default() {
3433 let opts = ClientOptions::default();
3434 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3435 }
3436
3437 #[test]
3438 fn session_idle_timeout_args_omitted_for_zero() {
3439 let opts = ClientOptions {
3440 session_idle_timeout_seconds: Some(0),
3441 ..Default::default()
3442 };
3443 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3444 }
3445
3446 #[test]
3447 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3448 let opts = ClientOptions {
3449 session_idle_timeout_seconds: Some(300),
3450 ..Default::default()
3451 };
3452 assert_eq!(
3453 Client::session_idle_timeout_args(&opts),
3454 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3455 );
3456 }
3457
3458 #[test]
3459 fn remote_args_omitted_by_default() {
3460 let opts = ClientOptions::default();
3461 assert!(Client::remote_args(&opts).is_empty());
3462 }
3463
3464 #[test]
3465 fn remote_args_emit_flag_when_enabled() {
3466 let opts = ClientOptions {
3467 enable_remote_sessions: true,
3468 ..Default::default()
3469 };
3470 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3471 }
3472
3473 #[test]
3474 fn log_level_args_omitted_when_unset() {
3475 let opts = ClientOptions::default();
3476 assert!(opts.log_level.is_none());
3477 assert!(
3478 Client::log_level_args(&opts).is_empty(),
3479 "with no caller-supplied log_level the SDK must not pass --log-level"
3480 );
3481 }
3482
3483 #[test]
3484 fn log_level_args_emit_flag_when_set() {
3485 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3486 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3487 }
3488
3489 #[test]
3490 fn log_level_str_round_trips() {
3491 for level in [
3492 LogLevel::None,
3493 LogLevel::Error,
3494 LogLevel::Warning,
3495 LogLevel::Info,
3496 LogLevel::Debug,
3497 LogLevel::All,
3498 ] {
3499 let s = level.as_str();
3500 let json = serde_json::to_string(&level).unwrap();
3501 assert_eq!(json, format!("\"{s}\""));
3502 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3503 assert_eq!(parsed, level);
3504 }
3505 }
3506
3507 #[test]
3508 fn client_options_debug_redacts_handler() {
3509 struct StubHandler;
3510 #[async_trait]
3511 impl ListModelsHandler for StubHandler {
3512 async fn list_models(&self) -> Result<Vec<Model>> {
3513 Ok(vec![])
3514 }
3515 }
3516 let opts = ClientOptions {
3517 on_list_models: Some(Arc::new(StubHandler)),
3518 github_token: Some("secret-token".into()),
3519 ..Default::default()
3520 };
3521 let debug = format!("{opts:?}");
3522 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3523 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3524 assert!(!debug.contains("secret-token"));
3525 }
3526
3527 #[tokio::test]
3528 async fn list_models_uses_on_list_models_handler_when_set() {
3529 use std::sync::atomic::{AtomicUsize, Ordering};
3530
3531 struct CountingHandler {
3532 calls: Arc<AtomicUsize>,
3533 models: Vec<Model>,
3534 }
3535 #[async_trait]
3536 impl ListModelsHandler for CountingHandler {
3537 async fn list_models(&self) -> Result<Vec<Model>> {
3538 self.calls.fetch_add(1, Ordering::SeqCst);
3539 Ok(self.models.clone())
3540 }
3541 }
3542
3543 let calls = Arc::new(AtomicUsize::new(0));
3544 let model = Model {
3545 id: "byok-gpt-4".into(),
3546 name: "BYOK GPT-4".into(),
3547 ..Default::default()
3548 };
3549 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3550 calls: Arc::clone(&calls),
3551 models: vec![model.clone()],
3552 });
3553
3554 let client = client_with_list_models_handler(handler);
3555
3556 let result = client.list_models().await.unwrap();
3557 assert_eq!(result.len(), 1);
3558 assert_eq!(result[0].id, "byok-gpt-4");
3559 assert_eq!(calls.load(Ordering::SeqCst), 1);
3560 }
3561
3562 #[tokio::test]
3563 async fn list_models_serializes_concurrent_cache_misses() {
3564 use std::sync::atomic::{AtomicUsize, Ordering};
3565
3566 struct SlowCountingHandler {
3567 calls: Arc<AtomicUsize>,
3568 models: Vec<Model>,
3569 }
3570 #[async_trait]
3571 impl ListModelsHandler for SlowCountingHandler {
3572 async fn list_models(&self) -> Result<Vec<Model>> {
3573 self.calls.fetch_add(1, Ordering::SeqCst);
3574 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3575 Ok(self.models.clone())
3576 }
3577 }
3578
3579 let calls = Arc::new(AtomicUsize::new(0));
3580 let model = Model {
3581 id: "single-flight-model".into(),
3582 name: "Single Flight Model".into(),
3583 ..Default::default()
3584 };
3585 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3586 calls: Arc::clone(&calls),
3587 models: vec![model],
3588 });
3589 let client = client_with_list_models_handler(handler);
3590
3591 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3592 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3593 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3594 assert_eq!(calls.load(Ordering::SeqCst), 1);
3595 }
3596
3597 #[tokio::test]
3598 async fn cancelled_resume_session_unregisters_pending_session() {
3599 let (client_write, _server_read) = tokio::io::duplex(8192);
3600 let (_server_write, client_read) = tokio::io::duplex(8192);
3601 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3602 assert!(client.startup_timings().is_none());
3603 let session_id = SessionId::new("resume-cancel-test");
3604 let handle = tokio::spawn({
3605 let client = client.clone();
3606 async move {
3607 client
3608 .resume_session(ResumeSessionConfig::new(session_id))
3609 .await
3610 }
3611 });
3612
3613 wait_for_pending_session_registration(&client).await;
3614 handle.abort();
3615 let _ = handle.await;
3616
3617 assert!(client.inner.router.session_ids().is_empty());
3618 client.force_stop();
3619 }
3620
3621 #[cfg(any(unix, windows))]
3622 #[tokio::test]
3623 async fn dropping_last_client_kills_spawned_cli() {
3624 let temp = tempfile::tempdir().unwrap();
3625 let ready = temp.path().join("ready");
3626 let survived = temp.path().join("survived");
3627 let child = test_child_command(temp.path(), &ready, &survived)
3628 .spawn()
3629 .unwrap();
3630 let (client_write, _server_read) = tokio::io::duplex(64);
3631 let (_server_write, client_read) = tokio::io::duplex(64);
3632 let client = Client::from_transport(
3633 client_read,
3634 client_write,
3635 Some(child),
3636 None,
3637 temp.path().to_path_buf(),
3638 None,
3639 None,
3640 false,
3641 false,
3642 None,
3643 None,
3644 None,
3645 ClientMode::default(),
3646 None,
3647 )
3648 .unwrap();
3649
3650 wait_for_test_child(&ready).await;
3651 drop(client);
3652
3653 assert_test_child_killed(&survived).await;
3654 }
3655
3656 #[cfg(any(unix, windows))]
3657 #[tokio::test]
3658 async fn spawned_child_is_killed_when_dropped() {
3659 let temp = tempfile::tempdir().unwrap();
3660 let ready = temp.path().join("ready");
3661 let survived = temp.path().join("survived");
3662 let child = test_child_command(temp.path(), &ready, &survived)
3663 .spawn()
3664 .unwrap();
3665
3666 wait_for_test_child(&ready).await;
3667 drop(child);
3668
3669 assert_test_child_killed(&survived).await;
3670 }
3671
3672 #[cfg(any(unix, windows))]
3673 fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command {
3674 #[cfg(unix)]
3675 let mut command = {
3676 let mut command =
3677 Client::build_command(Path::new("sh"), &ClientOptions::default(), temp);
3678 command.args([
3679 "-c",
3680 "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"",
3681 ]);
3682 command
3683 };
3684 #[cfg(windows)]
3685 let mut command = {
3686 let mut command =
3687 Client::build_command(Path::new("powershell.exe"), &ClientOptions::default(), temp);
3688 command.args([
3689 "-NoLogo",
3690 "-NoProfile",
3691 "-NonInteractive",
3692 "-Command",
3693 "Set-Content -LiteralPath $env:READY ready; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:SURVIVED survived",
3694 ]);
3695 command
3696 };
3697 command.env("READY", ready).env("SURVIVED", survived);
3698 command
3699 }
3700
3701 #[cfg(any(unix, windows))]
3702 async fn wait_for_test_child(ready: &Path) {
3703 let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
3704 while !ready.exists() {
3705 assert!(
3706 tokio::time::Instant::now() < deadline,
3707 "child did not report readiness"
3708 );
3709 tokio::time::sleep(Duration::from_millis(10)).await;
3710 }
3711 }
3712
3713 #[cfg(any(unix, windows))]
3714 async fn assert_test_child_killed(survived: &Path) {
3715 tokio::time::sleep(Duration::from_millis(1500)).await;
3716
3717 assert!(
3718 !survived.exists(),
3719 "child survived after its owner was dropped"
3720 );
3721 }
3722
3723 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3724 Client {
3725 inner: Arc::new(ClientInner {
3726 child: parking_lot::Mutex::new(None),
3727 process_tree: parking_lot::Mutex::new(None),
3728 #[cfg(feature = "bundled-in-process")]
3729 ffi_host: parking_lot::Mutex::new(None),
3730 rpc: {
3731 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3732 let (notif_tx, _notif_rx) = broadcast::channel(16);
3733 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3734 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3735 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3736 },
3737 cwd: PathBuf::from("."),
3738 request_rx: parking_lot::Mutex::new(None),
3739 notification_tx: broadcast::channel(16).0,
3740 router: router::SessionRouter::new(),
3741 github_token_registry: Arc::new(github_token::GitHubTokenRegistry::new()),
3742 negotiated_protocol_version: OnceLock::new(),
3743 state: parking_lot::Mutex::new(ConnectionState::Connected),
3744 lifecycle_tx: broadcast::channel(16).0,
3745 on_list_models: Some(handler),
3746 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3747 session_fs_configured: false,
3748 session_fs_sqlite_declared: false,
3749 llm_inference: OnceLock::new(),
3750 extension_launch_provider: Arc::new(
3751 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(None),
3752 ),
3753 on_github_telemetry: None,
3754 on_get_trace_context: None,
3755 effective_connection_token: None,
3756 mode: ClientMode::default(),
3757 client_info: None,
3758 startup_timings: OnceLock::new(),
3759 }),
3760 }
3761 }
3762
3763 async fn wait_for_pending_session_registration(client: &Client) {
3764 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3765 while client.inner.router.session_ids().is_empty() {
3766 assert!(
3767 tokio::time::Instant::now() < deadline,
3768 "session was not registered"
3769 );
3770 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3771 }
3772 }
3773}