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;
35pub mod installation_confirmation;
37mod jsonrpc;
38pub mod permission;
40mod process_tree;
41pub mod provider_token;
43mod provider_token_dispatch;
44pub(crate) mod resolve;
46mod router;
47pub mod session;
49pub mod session_fs;
51mod session_fs_dispatch;
52pub mod startup_timings;
54pub mod subscription;
56pub mod tool;
58pub mod trace_context;
60pub mod transforms;
62pub mod types;
64mod wire;
65
66pub mod session_events;
68
69pub mod rpc;
72
73#[derive(serde::Deserialize)]
74struct SessionDetachResponse {
75 success: bool,
76 error: Option<String>,
77}
78
79pub(crate) mod generated;
84
85pub mod mode;
88
89use std::ffi::OsString;
90use std::path::{Path, PathBuf};
91use std::process::Stdio;
92use std::sync::{Arc, OnceLock};
93use std::time::{Duration, Instant};
94
95use async_trait::async_trait;
96pub use github_token::{
97 GitHubToken, GitHubTokenProvider, GitHubTokenProviderArgs, GitHubTokenProviderResult,
98 GitHubTokenRequestReason,
99};
100pub use indexmap::IndexMap;
104pub(crate) use jsonrpc::{
107 JsonRpcClient, JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, error_codes,
108};
109pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet};
110pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs};
111
112#[cfg(feature = "test-support")]
114pub mod test_support {
115 pub use crate::jsonrpc::{
116 JsonRpcClient, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
117 error_codes,
118 };
119}
120use serde::{Deserialize, Serialize};
121use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader};
122use tokio::net::TcpStream;
123use tokio::process::{Child, Command};
124use tokio::sync::{broadcast, mpsc, oneshot};
125use tracing::{Instrument, debug, error, info, warn};
126pub use types::*;
127
128mod sdk_protocol_version;
129pub use sdk_protocol_version::{SDK_PROTOCOL_VERSION, get_sdk_protocol_version};
130pub use startup_timings::StartupTimings;
131pub use subscription::{EventSubscription, LifecycleSubscription};
132
133const MIN_PROTOCOL_VERSION: u32 = 3;
135const RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
136
137fn record_optional_millis(span: &tracing::Span, field: &'static str, value: Option<u64>) {
138 match value {
139 Some(value) => {
140 span.record(field, value);
141 }
142 None => {
143 span.record(field, "None");
144 }
145 }
146}
147
148#[derive(Debug, Default)]
150#[non_exhaustive]
151pub enum Transport {
152 #[default]
155 Default,
156 Stdio,
158 InProcess,
172 Tcp {
174 port: u16,
176 connection_token: Option<String>,
180 },
181 External {
183 host: String,
185 port: u16,
187 connection_token: Option<String>,
190 },
191}
192
193#[derive(Debug, Clone, Default)]
195pub enum CliProgram {
196 #[default]
201 Resolve,
202 Path(PathBuf),
204}
205
206impl From<PathBuf> for CliProgram {
207 fn from(path: PathBuf) -> Self {
208 Self::Path(path)
209 }
210}
211
212pub const HAS_BUNDLED_CLI: bool = cfg!(has_bundled_cli);
219
220pub fn install_bundled_cli() -> Option<PathBuf> {
241 #[cfg(feature = "bundled-cli")]
242 {
243 embeddedcli::path()
244 }
245 #[cfg(not(feature = "bundled-cli"))]
246 {
247 None
248 }
249}
250
251pub fn install_bundled_runtime() -> Option<PathBuf> {
262 #[cfg(feature = "bundled-cli")]
263 {
264 embeddedcli::runtime_path()
265 }
266 #[cfg(not(feature = "bundled-cli"))]
267 {
268 None
269 }
270}
271
272#[non_exhaustive]
284pub struct ClientOptions {
285 pub program: CliProgram,
287 pub prefix_args: Vec<OsString>,
289 pub working_directory: PathBuf,
293 pub env: Vec<(OsString, OsString)>,
295 pub env_remove: Vec<OsString>,
297 pub extra_args: Vec<String>,
299 pub builtin_plugin_directories: Vec<PathBuf>,
304 pub transport: Transport,
306 pub github_token: Option<String>,
311 pub use_logged_in_user: Option<bool>,
315 pub log_level: Option<LogLevel>,
319 pub session_idle_timeout_seconds: Option<u64>,
325 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
333 pub session_fs: Option<SessionFsConfig>,
341 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
350 pub extension_launch_provider:
357 Option<Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>>,
358 pub installation_confirmation_handler:
363 Option<Arc<dyn crate::installation_confirmation::InstallationConfirmationHandler>>,
364 #[doc(hidden)]
372 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
373 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
383 pub telemetry: Option<TelemetryConfig>,
387 pub base_directory: Option<PathBuf>,
392 pub enable_remote_sessions: bool,
398 pub bundled_cli_extract_dir: Option<PathBuf>,
417 pub mode: ClientMode,
421 pub client_info: Option<ClientInfo>,
428}
429
430#[derive(Debug, Clone, Default, PartialEq, Eq)]
442#[non_exhaustive]
443pub struct ClientInfo {
444 pub application_name: Option<String>,
446 pub application_version: Option<String>,
448 pub integration_name: Option<String>,
451 pub integration_version: Option<String>,
453}
454
455impl ClientInfo {
456 pub fn new() -> Self {
459 Self::default()
460 }
461
462 pub fn with_application_name(mut self, application_name: impl Into<String>) -> Self {
464 self.application_name = Some(application_name.into());
465 self
466 }
467
468 pub fn with_application_version(mut self, application_version: impl Into<String>) -> Self {
470 self.application_version = Some(application_version.into());
471 self
472 }
473
474 pub fn with_integration_name(mut self, integration_name: impl Into<String>) -> Self {
477 self.integration_name = Some(integration_name.into());
478 self
479 }
480
481 pub fn with_integration_version(mut self, integration_version: impl Into<String>) -> Self {
484 self.integration_version = Some(integration_version.into());
485 self
486 }
487
488 fn is_empty(&self) -> bool {
492 Self::non_empty(&self.application_name).is_none()
493 && Self::non_empty(&self.application_version).is_none()
494 && Self::non_empty(&self.integration_name).is_none()
495 && Self::non_empty(&self.integration_version).is_none()
496 }
497
498 fn non_empty(value: &Option<String>) -> Option<String> {
501 value.as_ref().filter(|s| !s.is_empty()).cloned()
502 }
503
504 fn to_wire(&self) -> Option<crate::generated::api_types::ConnectClientInfo> {
507 if self.is_empty() {
508 return None;
509 }
510 Some(crate::generated::api_types::ConnectClientInfo {
511 editor_name: Self::non_empty(&self.application_name),
512 editor_version: Self::non_empty(&self.application_version),
513 extension_name: Self::non_empty(&self.integration_name),
514 extension_version: Self::non_empty(&self.integration_version),
515 })
516 }
517}
518
519impl std::fmt::Debug for ClientOptions {
520 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
521 f.debug_struct("ClientOptions")
522 .field("program", &self.program)
523 .field("prefix_args", &self.prefix_args)
524 .field("working_directory", &self.working_directory)
525 .field("env", &self.env)
526 .field("env_remove", &self.env_remove)
527 .field("extra_args", &self.extra_args)
528 .field(
529 "builtin_plugin_directories",
530 &self.builtin_plugin_directories,
531 )
532 .field("transport", &self.transport)
533 .field(
534 "github_token",
535 &self.github_token.as_ref().map(|_| "<redacted>"),
536 )
537 .field("use_logged_in_user", &self.use_logged_in_user)
538 .field("log_level", &self.log_level)
539 .field(
540 "session_idle_timeout_seconds",
541 &self.session_idle_timeout_seconds,
542 )
543 .field(
544 "on_list_models",
545 &self.on_list_models.as_ref().map(|_| "<set>"),
546 )
547 .field("session_fs", &self.session_fs)
548 .field(
549 "request_handler",
550 &self.request_handler.as_ref().map(|_| "<set>"),
551 )
552 .field(
553 "extension_launch_provider",
554 &self.extension_launch_provider.as_ref().map(|_| "<set>"),
555 )
556 .field(
557 "installation_confirmation_handler",
558 &self
559 .installation_confirmation_handler
560 .as_ref()
561 .map(|_| "<set>"),
562 )
563 .field(
564 "on_github_telemetry",
565 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
566 )
567 .field(
568 "on_get_trace_context",
569 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
570 )
571 .field("telemetry", &self.telemetry)
572 .field("base_directory", &self.base_directory)
573 .field("enable_remote_sessions", &self.enable_remote_sessions)
574 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
575 .field("client_info", &self.client_info)
576 .finish()
577 }
578}
579
580#[async_trait]
589pub trait ListModelsHandler: Send + Sync + 'static {
590 async fn list_models(&self) -> Result<Vec<Model>>;
592}
593
594#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
596#[serde(rename_all = "lowercase")]
597pub enum LogLevel {
598 None,
600 Error,
602 Warning,
604 Info,
606 Debug,
608 All,
610}
611
612impl LogLevel {
613 pub fn as_str(self) -> &'static str {
615 match self {
616 Self::None => "none",
617 Self::Error => "error",
618 Self::Warning => "warning",
619 Self::Info => "info",
620 Self::Debug => "debug",
621 Self::All => "all",
622 }
623 }
624}
625
626impl std::fmt::Display for LogLevel {
627 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
628 f.write_str(self.as_str())
629 }
630}
631
632#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
637#[serde(rename_all = "kebab-case")]
638#[non_exhaustive]
639pub enum OtelExporterType {
640 OtlpHttp,
643 File,
646}
647
648impl OtelExporterType {
649 pub fn as_str(self) -> &'static str {
651 match self {
652 Self::OtlpHttp => "otlp-http",
653 Self::File => "file",
654 }
655 }
656}
657
658#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
664#[non_exhaustive]
665pub enum OtlpHttpProtocol {
666 #[serde(rename = "http/json")]
668 HttpJson,
669 #[serde(rename = "http/protobuf")]
671 HttpProtobuf,
672}
673
674impl OtlpHttpProtocol {
675 pub fn as_str(self) -> &'static str {
677 match self {
678 Self::HttpJson => "http/json",
679 Self::HttpProtobuf => "http/protobuf",
680 }
681 }
682}
683
684#[derive(Debug, Clone, Default)]
719#[non_exhaustive]
720pub struct TelemetryConfig {
721 pub otlp_endpoint: Option<String>,
723 pub otlp_protocol: Option<OtlpHttpProtocol>,
725 pub file_path: Option<PathBuf>,
727 pub exporter_type: Option<OtelExporterType>,
730 pub source_name: Option<String>,
734 pub capture_content: Option<bool>,
738}
739
740impl TelemetryConfig {
741 pub fn new() -> Self {
744 Self::default()
745 }
746
747 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
749 self.otlp_endpoint = Some(endpoint.into());
750 self
751 }
752
753 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
755 self.otlp_protocol = Some(protocol);
756 self
757 }
758
759 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
761 self.file_path = Some(path.into());
762 self
763 }
764
765 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
767 self.exporter_type = Some(exporter_type);
768 self
769 }
770
771 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
775 self.source_name = Some(source_name.into());
776 self
777 }
778
779 pub fn with_capture_content(mut self, capture: bool) -> Self {
783 self.capture_content = Some(capture);
784 self
785 }
786
787 pub fn is_empty(&self) -> bool {
790 self.otlp_endpoint.is_none()
791 && self.otlp_protocol.is_none()
792 && self.file_path.is_none()
793 && self.exporter_type.is_none()
794 && self.source_name.is_none()
795 && self.capture_content.is_none()
796 }
797}
798
799impl Default for ClientOptions {
800 fn default() -> Self {
801 Self {
802 program: CliProgram::Resolve,
803 prefix_args: Vec::new(),
804 working_directory: PathBuf::new(),
805 env: Vec::new(),
806 env_remove: Vec::new(),
807 extra_args: Vec::new(),
808 builtin_plugin_directories: Vec::new(),
809 transport: Transport::default(),
810 github_token: None,
811 use_logged_in_user: None,
812 log_level: None,
813 session_idle_timeout_seconds: None,
814 on_list_models: None,
815 session_fs: None,
816 request_handler: None,
817 extension_launch_provider: None,
818 installation_confirmation_handler: None,
819 on_github_telemetry: None,
820 on_get_trace_context: None,
821 telemetry: None,
822 base_directory: None,
823 enable_remote_sessions: false,
824 bundled_cli_extract_dir: None,
825 mode: ClientMode::default(),
826 client_info: None,
827 }
828 }
829}
830
831impl ClientOptions {
832 pub fn new() -> Self {
848 Self::default()
849 }
850
851 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
853 self.program = program.into();
854 self
855 }
856
857 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
859 where
860 I: IntoIterator<Item = S>,
861 S: Into<OsString>,
862 {
863 self.prefix_args = args.into_iter().map(Into::into).collect();
864 self
865 }
866
867 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
869 self.working_directory = cwd.into();
870 self
871 }
872
873 pub fn with_env<I, K, V>(mut self, env: I) -> Self
875 where
876 I: IntoIterator<Item = (K, V)>,
877 K: Into<OsString>,
878 V: Into<OsString>,
879 {
880 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
881 self
882 }
883
884 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
886 where
887 I: IntoIterator<Item = S>,
888 S: Into<OsString>,
889 {
890 self.env_remove = names.into_iter().map(Into::into).collect();
891 self
892 }
893
894 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
896 where
897 I: IntoIterator<Item = S>,
898 S: Into<String>,
899 {
900 self.extra_args = args.into_iter().map(Into::into).collect();
901 self
902 }
903
904 pub fn with_builtin_plugin_directories<I, P>(mut self, paths: I) -> Self
909 where
910 I: IntoIterator<Item = P>,
911 P: Into<PathBuf>,
912 {
913 self.builtin_plugin_directories = paths.into_iter().map(Into::into).collect();
914 self
915 }
916
917 pub fn with_transport(mut self, transport: Transport) -> Self {
919 self.transport = transport;
920 self
921 }
922
923 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
926 self.github_token = Some(token.into());
927 self
928 }
929
930 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
933 self.use_logged_in_user = Some(use_logged_in);
934 self
935 }
936
937 pub fn with_log_level(mut self, level: LogLevel) -> Self {
939 self.log_level = Some(level);
940 self
941 }
942
943 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
946 self.session_idle_timeout_seconds = Some(seconds);
947 self
948 }
949
950 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
953 where
954 H: ListModelsHandler + 'static,
955 {
956 self.on_list_models = Some(Arc::new(handler));
957 self
958 }
959
960 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
962 self.session_fs = Some(config);
963 self
964 }
965
966 pub fn with_request_handler<H>(mut self, handler: H) -> Self
971 where
972 H: crate::copilot_request_handler::CopilotRequestHandler,
973 {
974 self.request_handler = Some(Arc::new(handler));
975 self
976 }
977
978 pub fn with_extension_launch_provider<P>(mut self, provider: P) -> Self
983 where
984 P: crate::extension_launch_provider::ExtensionLaunchProvider,
985 {
986 self.extension_launch_provider = Some(Arc::new(provider));
987 self
988 }
989
990 pub fn with_installation_confirmation_handler<H>(mut self, handler: H) -> Self
992 where
993 H: crate::installation_confirmation::InstallationConfirmationHandler,
994 {
995 self.installation_confirmation_handler = Some(Arc::new(handler));
996 self
997 }
998
999 #[doc(hidden)]
1005 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
1006 where
1007 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
1008 {
1009 self.on_github_telemetry = Some(Arc::new(callback));
1010 self
1011 }
1012
1013 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
1017 where
1018 P: TraceContextProvider + 'static,
1019 {
1020 self.on_get_trace_context = Some(Arc::new(provider));
1021 self
1022 }
1023
1024 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
1026 self.telemetry = Some(config);
1027 self
1028 }
1029
1030 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
1033 self.base_directory = Some(dir.into());
1034 self
1035 }
1036
1037 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
1040 self.enable_remote_sessions = enabled;
1041 self
1042 }
1043
1044 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
1054 self.bundled_cli_extract_dir = Some(dir.into());
1055 self
1056 }
1057
1058 pub fn with_mode(mut self, mode: ClientMode) -> Self {
1063 self.mode = mode;
1064 self
1065 }
1066
1067 pub fn with_client_info(mut self, client_info: ClientInfo) -> Self {
1071 self.client_info = Some(client_info);
1072 self
1073 }
1074}
1075
1076fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
1078 if cfg.initial_cwd.trim().is_empty() {
1079 return Err(Error::with_message(
1080 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
1081 "invalid SessionFsConfig: initial_cwd must not be empty",
1082 ));
1083 }
1084 if cfg.session_state_path.trim().is_empty() {
1085 return Err(Error::with_message(
1086 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
1087 "invalid SessionFsConfig: session_state_path must not be empty",
1088 ));
1089 }
1090 Ok(())
1091}
1092
1093fn generate_connection_token() -> String {
1100 let mut bytes = [0u8; 16];
1101 getrandom::fill(&mut bytes)
1102 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
1103 let mut hex = String::with_capacity(32);
1104 for byte in bytes {
1105 use std::fmt::Write;
1106 let _ = write!(hex, "{byte:02x}");
1107 }
1108 hex
1109}
1110
1111const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
1116
1117fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
1119 let configured = options
1120 .env
1121 .iter()
1122 .find(|(key, _)| {
1123 key.to_string_lossy()
1124 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
1125 })
1126 .map(|(_, value)| value.to_string_lossy().into_owned());
1127 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
1128 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
1129}
1130
1131fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
1132 match value {
1133 None => Ok(Transport::Stdio),
1134 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
1135 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
1136 Some(v) => Err(Error::with_message(
1137 ErrorKind::InvalidConfig,
1138 format!(
1139 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
1140 Expected 'inprocess', 'stdio', or unset."
1141 ),
1142 )),
1143 }
1144}
1145
1146#[cfg(any(feature = "in-process", test))]
1147fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
1148 if !matches!(&options.program, CliProgram::Resolve) {
1149 return Err(Error::with_message(
1150 ErrorKind::InvalidConfig,
1151 "ClientOptions::program is not supported with Transport::InProcess; \
1152 set COPILOT_CLI_PATH only when using an externally provisioned runtime package",
1153 ));
1154 }
1155 if !options.extra_args.is_empty() {
1156 return Err(Error::with_message(
1157 ErrorKind::InvalidConfig,
1158 "ClientOptions::extra_args is not supported with Transport::InProcess; \
1159 use typed client options instead",
1160 ));
1161 }
1162
1163 let unsupported = if !options.working_directory.as_os_str().is_empty() {
1164 Some("working_directory")
1165 } else if !options.env.is_empty() {
1166 Some("env")
1167 } else if !options.env_remove.is_empty() {
1168 Some("env_remove")
1169 } else if options.telemetry.is_some() {
1170 Some("telemetry")
1171 } else if !options.prefix_args.is_empty() {
1172 Some("prefix_args")
1173 } else {
1174 None
1175 };
1176
1177 if let Some(option) = unsupported {
1178 return Err(Error::with_message(
1179 ErrorKind::InvalidConfig,
1180 format!(
1181 "ClientOptions::{option} is not supported with Transport::InProcess; \
1182 configure process-global settings on the host process instead"
1183 ),
1184 ));
1185 }
1186
1187 Ok(())
1188}
1189
1190#[derive(Clone)]
1195pub struct Client {
1196 inner: Arc<ClientInner>,
1197}
1198
1199impl std::fmt::Debug for Client {
1200 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1201 f.debug_struct("Client")
1202 .field("working_directory", &self.inner.cwd)
1203 .field("pid", &self.pid())
1204 .finish()
1205 }
1206}
1207
1208struct ClientInner {
1209 child: parking_lot::Mutex<Option<Child>>,
1210 process_tree: parking_lot::Mutex<Option<process_tree::ProcessTree>>,
1211 #[cfg(feature = "in-process")]
1212 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
1215 rpc: JsonRpcClient,
1216 cwd: PathBuf,
1217 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
1218 notification_tx: broadcast::Sender<JsonRpcNotification>,
1219 router: router::SessionRouter,
1220 github_token_registry: Arc<github_token::GitHubTokenRegistry>,
1221 negotiated_protocol_version: OnceLock<u32>,
1222 state: parking_lot::Mutex<ConnectionState>,
1223 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
1224 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1225 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
1226 session_fs_configured: bool,
1227 session_fs_sqlite_declared: bool,
1228 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
1231 extension_launch_provider: Arc<extension_launch_provider::ExtensionLaunchProviderDispatcher>,
1232 installation_confirmation: Arc<installation_confirmation::InstallationConfirmationDispatcher>,
1233 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1238 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1239 effective_connection_token: Option<String>,
1244 client_info: Option<ClientInfo>,
1248 pub(crate) mode: ClientMode,
1251 startup_timings: OnceLock<StartupTimings>,
1255}
1256
1257impl Client {
1258 pub async fn start(options: ClientOptions) -> Result<Self> {
1271 let start_time = Instant::now();
1272 let mut timings = StartupTimings::default();
1273 let mut options = options;
1274 if matches!(options.transport, Transport::Default) {
1275 options.transport = resolve_default_transport(&options)?;
1276 }
1277 if matches!(options.transport, Transport::InProcess) {
1278 #[cfg(not(feature = "in-process"))]
1279 {
1280 return Err(Error::with_message(
1281 ErrorKind::InvalidConfig,
1282 "Transport::InProcess requires the `in-process` Cargo feature",
1283 ));
1284 }
1285 #[cfg(feature = "in-process")]
1286 validate_inprocess_options(&options)?;
1287 }
1288 if options.mode == ClientMode::Empty
1289 && options.base_directory.is_none()
1290 && options.session_fs.is_none()
1291 {
1292 return Err(Error::with_message(
1293 ErrorKind::InvalidConfig,
1294 "ClientMode::Empty requires either `base_directory` or \
1295 `session_fs` to be set (no implicit ~/.copilot fallback).",
1296 ));
1297 }
1298 if let Some(cfg) = &options.session_fs {
1299 validate_session_fs_config(cfg)?;
1300 }
1301 let builtin_plugin_directories = options
1302 .builtin_plugin_directories
1303 .iter()
1304 .map(|path| {
1305 if !path.is_absolute() {
1306 return Err(Error::with_message(
1307 ErrorKind::InvalidConfig,
1308 format!(
1309 "builtin_plugin_directories must contain only absolute paths: {}",
1310 path.display()
1311 ),
1312 ));
1313 }
1314 path.to_str().map(str::to_owned).ok_or_else(|| {
1315 Error::with_message(
1316 ErrorKind::InvalidConfig,
1317 format!(
1318 "builtin_plugin_directories must contain valid UTF-8 paths: {}",
1319 path.display()
1320 ),
1321 )
1322 })
1323 })
1324 .collect::<Result<Vec<_>>>()?;
1325 if matches!(options.transport, Transport::External { .. }) {
1328 if options.github_token.is_some() {
1329 return Err(Error::with_message(
1330 ErrorKind::InvalidConfig,
1331 "invalid client configuration: github_token cannot be used with \
1332 Transport::External (external server manages its own auth)",
1333 ));
1334 }
1335 if options.use_logged_in_user == Some(true) {
1336 return Err(Error::with_message(
1337 ErrorKind::InvalidConfig,
1338 "invalid client configuration: use_logged_in_user cannot be used with \
1339 Transport::External (external server manages its own auth)",
1340 ));
1341 }
1342 }
1343 match &options.transport {
1347 Transport::Tcp {
1348 connection_token: Some(t),
1349 ..
1350 }
1351 | Transport::External {
1352 connection_token: Some(t),
1353 ..
1354 } if t.is_empty() => {
1355 return Err(Error::with_message(
1356 ErrorKind::InvalidConfig,
1357 "invalid client configuration: connection_token must be a non-empty string",
1358 ));
1359 }
1360 _ => {}
1361 }
1362 let effective_connection_token: Option<String> = match &mut options.transport {
1367 Transport::Default => unreachable!("default transport resolved above"),
1368 Transport::Stdio | Transport::InProcess => None,
1369 Transport::Tcp {
1370 connection_token, ..
1371 } => Some(
1372 connection_token
1373 .get_or_insert_with(generate_connection_token)
1374 .clone(),
1375 ),
1376 Transport::External {
1377 connection_token, ..
1378 } => connection_token.clone(),
1379 };
1380 let session_fs_config = options.session_fs.clone();
1381 let request_handler = options.request_handler.clone();
1382 let extension_launch_provider = options.extension_launch_provider.clone();
1383 let installation_confirmation_handler = options.installation_confirmation_handler.clone();
1384 let session_fs_sqlite_declared = session_fs_config
1385 .as_ref()
1386 .and_then(|c| c.capabilities.as_ref())
1387 .is_some_and(|caps| caps.sqlite);
1388 let program = match &options.program {
1389 CliProgram::Path(path) => {
1390 info!(path = %path.display(), "using explicit copilot CLI path");
1391 path.clone()
1392 }
1393 CliProgram::Resolve => {
1394 let resolve_start = Instant::now();
1395 let resolved = resolve::copilot_binary_with_extract_dir(
1396 options.bundled_cli_extract_dir.as_deref(),
1397 true,
1398 )?;
1399 let resolve_elapsed = resolve_start.elapsed();
1400 timings.program_resolve_ms = Some(StartupTimings::millis(resolve_elapsed));
1401 debug!(
1402 elapsed_ms = resolve_elapsed.as_millis(),
1403 "Client::start CLI program resolution complete"
1404 );
1405 info!(path = %resolved.display(), "resolved copilot runtime");
1406 #[cfg(windows)]
1407 {
1408 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1409 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1410 }) {
1411 warn!(
1412 path = %resolved.display(),
1413 ext = %ext,
1414 "resolved copilot CLI is a .cmd/.bat wrapper; \
1415 this may cause console window flashes on Windows"
1416 );
1417 }
1418 }
1419 resolved
1420 }
1421 };
1422 let working_directory = {
1423 let cwd = options.working_directory.clone();
1424 if cwd.as_os_str().is_empty() {
1425 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1426 } else {
1427 cwd
1428 }
1429 };
1430
1431 let transport_setup_start = Instant::now();
1432 let client = match options.transport {
1433 Transport::Default => unreachable!("default transport resolved above"),
1434 Transport::External {
1435 ref host,
1436 port,
1437 connection_token: _,
1438 } => {
1439 info!(host = %host, port = %port, "connecting to external CLI server");
1440 let connect_start = Instant::now();
1441 let stream = TcpStream::connect((host.as_str(), port)).await?;
1442 debug!(
1443 elapsed_ms = connect_start.elapsed().as_millis(),
1444 host = %host,
1445 port,
1446 "Client::start TCP connect complete"
1447 );
1448 let (reader, writer) = tokio::io::split(stream);
1449 Self::from_transport(
1450 reader,
1451 writer,
1452 None,
1453 None,
1454 working_directory,
1455 options.on_list_models,
1456 extension_launch_provider.clone(),
1457 session_fs_config.is_some(),
1458 session_fs_sqlite_declared,
1459 options.on_get_trace_context,
1460 options.on_github_telemetry,
1461 effective_connection_token.clone(),
1462 options.mode,
1463 options.client_info,
1464 )?
1465 }
1466 Transport::Tcp {
1467 port,
1468 connection_token: _,
1469 } => {
1470 let (mut child, tree, actual_port, spawn_elapsed, port_wait_elapsed) =
1471 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1472 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1473 timings.port_wait_ms = Some(StartupTimings::millis(port_wait_elapsed));
1474 let connect_start = Instant::now();
1475 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1476 debug!(
1477 elapsed_ms = connect_start.elapsed().as_millis(),
1478 port = actual_port,
1479 "Client::start TCP connect complete"
1480 );
1481 let (reader, writer) = tokio::io::split(stream);
1482 Self::drain_stderr(&mut child);
1483 Self::from_transport(
1484 reader,
1485 writer,
1486 Some(child),
1487 tree,
1488 working_directory,
1489 options.on_list_models,
1490 extension_launch_provider.clone(),
1491 session_fs_config.is_some(),
1492 session_fs_sqlite_declared,
1493 options.on_get_trace_context,
1494 options.on_github_telemetry,
1495 effective_connection_token.clone(),
1496 options.mode,
1497 options.client_info,
1498 )?
1499 }
1500 Transport::Stdio => {
1501 let (mut child, tree, spawn_elapsed) =
1502 Self::spawn_stdio(&program, &options, &working_directory)?;
1503 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1504 let stdin = child.stdin.take().expect("stdin is piped");
1505 let stdout = child.stdout.take().expect("stdout is piped");
1506 Self::drain_stderr(&mut child);
1507 Self::from_transport(
1508 stdout,
1509 stdin,
1510 Some(child),
1511 tree,
1512 working_directory,
1513 options.on_list_models,
1514 extension_launch_provider.clone(),
1515 session_fs_config.is_some(),
1516 session_fs_sqlite_declared,
1517 options.on_get_trace_context,
1518 options.on_github_telemetry,
1519 effective_connection_token.clone(),
1520 options.mode,
1521 options.client_info,
1522 )?
1523 }
1524 Transport::InProcess => {
1525 #[cfg(feature = "in-process")]
1526 {
1527 info!(runtime_path = %program.display(), "hosting copilot runtime in-process (FFI)");
1528 let mut environment = Vec::new();
1529 if let Some(base_directory) = &options.base_directory {
1530 let value = base_directory.to_str().ok_or_else(|| {
1531 Error::with_message(
1532 ErrorKind::InvalidConfig,
1533 "base_directory must be valid UTF-8 for Transport::InProcess",
1534 )
1535 })?;
1536 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1537 }
1538 if options.mode == ClientMode::Empty {
1539 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1540 }
1541 if let Some(github_token) = &options.github_token {
1542 environment
1543 .push(("COPILOT_SDK_AUTH_TOKEN".to_string(), github_token.clone()));
1544 }
1545 let mut args = Vec::new();
1546 args.extend(
1547 Self::log_level_args(&options)
1548 .into_iter()
1549 .map(str::to_string),
1550 );
1551 args.extend(Self::session_idle_timeout_args(&options));
1552 args.extend(Self::remote_args(&options));
1553 if options.github_token.is_some() {
1554 args.extend([
1555 "--auth-token-env".to_string(),
1556 "COPILOT_SDK_AUTH_TOKEN".to_string(),
1557 ]);
1558 }
1559 let use_logged_in_user = options
1560 .use_logged_in_user
1561 .unwrap_or(options.github_token.is_none());
1562 if !use_logged_in_user {
1563 args.push("--no-auto-login".to_string());
1564 }
1565 let explicit_cli = std::env::var_os("COPILOT_CLI_PATH")
1566 .map(PathBuf::from)
1567 .filter(|path| path.is_file());
1568 let host = crate::ffi::FfiHost::create(
1569 &program,
1570 explicit_cli.as_deref(),
1571 environment,
1572 args,
1573 )?;
1574 let (reader, writer, shared) = host.start().await?;
1575 let client = Self::from_transport(
1576 reader,
1577 writer,
1578 None,
1579 None,
1580 working_directory,
1581 options.on_list_models,
1582 extension_launch_provider.clone(),
1583 session_fs_config.is_some(),
1584 session_fs_sqlite_declared,
1585 options.on_get_trace_context,
1586 options.on_github_telemetry,
1587 effective_connection_token.clone(),
1588 options.mode,
1589 options.client_info,
1590 )?;
1591 *client.inner.ffi_host.lock() = Some(shared);
1592 client
1593 }
1594 #[cfg(not(feature = "in-process"))]
1595 unreachable!("in-process feature validation returned above")
1596 }
1597 };
1598 timings.transport_setup_ms = StartupTimings::millis(transport_setup_start.elapsed());
1599 client
1600 .inner
1601 .installation_confirmation
1602 .set_handler(installation_confirmation_handler);
1603 debug!(
1604 elapsed_ms = start_time.elapsed().as_millis(),
1605 "Client::start transport setup complete"
1606 );
1607 let handshake_start = Instant::now();
1608 client.verify_protocol_version().await?;
1609 timings.handshake_ms = StartupTimings::millis(handshake_start.elapsed());
1610 debug!(
1611 elapsed_ms = start_time.elapsed().as_millis(),
1612 "Client::start protocol verification complete"
1613 );
1614 let request_dispatcher = request_handler.map(|handler| {
1615 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1616 handler,
1617 ));
1618 dispatcher.set_client(Arc::downgrade(&client.inner));
1619 let _ = client.inner.llm_inference.set(dispatcher.clone());
1620 dispatcher
1621 });
1622 client.inner.router.ensure_started(&client.inner);
1623 if client.inner.extension_launch_provider.is_configured() {
1624 client.rpc().register_extension_launch_provider().await?;
1625 }
1626 if !builtin_plugin_directories.is_empty() {
1627 client
1628 .call(
1629 "plugins.builtin.set",
1630 Some(serde_json::json!({ "paths": builtin_plugin_directories })),
1631 )
1632 .await?;
1633 }
1634 if let Some(cfg) = session_fs_config {
1635 let session_fs_start = Instant::now();
1636 let capabilities = cfg.capabilities.as_ref().map(|c| {
1637 crate::generated::api_types::SessionFsSetProviderCapabilities {
1638 sqlite: Some(c.sqlite),
1639 }
1640 });
1641 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1642 capabilities,
1643 conventions: cfg.conventions.into_wire(),
1644 initial_cwd: cfg.initial_cwd,
1645 session_state_path: cfg.session_state_path,
1646 };
1647 client.rpc().session_fs().set_provider(request).await?;
1648 let session_fs_elapsed = session_fs_start.elapsed();
1649 timings.session_fs_ms = Some(StartupTimings::millis(session_fs_elapsed));
1650 debug!(
1651 elapsed_ms = session_fs_elapsed.as_millis(),
1652 "Client::start session filesystem setup complete"
1653 );
1654 }
1655 if request_dispatcher.is_some() {
1656 let llm_inference_start = Instant::now();
1657 client.rpc().llm_inference().set_provider().await?;
1658 let llm_inference_elapsed = llm_inference_start.elapsed();
1659 timings.llm_handler_ms = Some(StartupTimings::millis(llm_inference_elapsed));
1660 debug!(
1661 elapsed_ms = llm_inference_elapsed.as_millis(),
1662 "Client::start Copilot request handler registration complete"
1663 );
1664 }
1665 timings.total_ms = StartupTimings::millis(start_time.elapsed());
1666 let timings_span = tracing::debug_span!(
1669 "Client::start timings",
1670 program_resolve_ms = tracing::field::Empty,
1671 process_spawn_ms = tracing::field::Empty,
1672 port_wait_ms = tracing::field::Empty,
1673 transport_setup_ms = timings.transport_setup_ms,
1674 handshake_ms = timings.handshake_ms,
1675 session_fs_ms = tracing::field::Empty,
1676 llm_handler_ms = tracing::field::Empty,
1677 total_ms = timings.total_ms,
1678 );
1679 record_optional_millis(
1680 &timings_span,
1681 "program_resolve_ms",
1682 timings.program_resolve_ms,
1683 );
1684 record_optional_millis(&timings_span, "process_spawn_ms", timings.process_spawn_ms);
1685 record_optional_millis(&timings_span, "port_wait_ms", timings.port_wait_ms);
1686 record_optional_millis(&timings_span, "session_fs_ms", timings.session_fs_ms);
1687 record_optional_millis(&timings_span, "llm_handler_ms", timings.llm_handler_ms);
1688 timings_span.in_scope(|| debug!("Client::start timings"));
1689 let _ = client.inner.startup_timings.set(timings);
1690 debug!(
1691 elapsed_ms = start_time.elapsed().as_millis(),
1692 "Client::start complete"
1693 );
1694 Ok(client)
1695 }
1696
1697 pub fn from_streams(
1701 reader: impl AsyncRead + Unpin + Send + 'static,
1702 writer: impl AsyncWrite + Unpin + Send + 'static,
1703 cwd: PathBuf,
1704 ) -> Result<Self> {
1705 Self::from_transport(
1706 reader,
1707 writer,
1708 None,
1709 None,
1710 cwd,
1711 None,
1712 None,
1713 false,
1714 false,
1715 None,
1716 None,
1717 None,
1718 ClientMode::default(),
1719 None,
1720 )
1721 }
1722
1723 #[doc(hidden)]
1726 #[cfg(any(test, feature = "test-support"))]
1727 pub fn from_streams_with_extension_launch_provider(
1728 reader: impl AsyncRead + Unpin + Send + 'static,
1729 writer: impl AsyncWrite + Unpin + Send + 'static,
1730 cwd: PathBuf,
1731 provider: Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1732 ) -> Result<Self> {
1733 Self::from_transport(
1734 reader,
1735 writer,
1736 None,
1737 None,
1738 cwd,
1739 None,
1740 Some(provider),
1741 false,
1742 false,
1743 None,
1744 None,
1745 None,
1746 ClientMode::default(),
1747 None,
1748 )
1749 }
1750
1751 #[doc(hidden)]
1753 #[cfg(any(test, feature = "test-support"))]
1754 pub fn from_streams_with_installation_confirmation_handler(
1755 reader: impl AsyncRead + Unpin + Send + 'static,
1756 writer: impl AsyncWrite + Unpin + Send + 'static,
1757 cwd: PathBuf,
1758 handler: Arc<dyn crate::installation_confirmation::InstallationConfirmationHandler>,
1759 ) -> Result<Self> {
1760 let client = Self::from_streams(reader, writer, cwd)?;
1761 client
1762 .inner
1763 .installation_confirmation
1764 .set_handler(Some(handler));
1765 Ok(client)
1766 }
1767
1768 #[cfg(any(test, feature = "test-support"))]
1776 pub fn from_streams_with_trace_provider(
1777 reader: impl AsyncRead + Unpin + Send + 'static,
1778 writer: impl AsyncWrite + Unpin + Send + 'static,
1779 cwd: PathBuf,
1780 provider: Arc<dyn TraceContextProvider>,
1781 ) -> Result<Self> {
1782 Self::from_transport(
1783 reader,
1784 writer,
1785 None,
1786 None,
1787 cwd,
1788 None,
1789 None,
1790 false,
1791 false,
1792 Some(provider),
1793 None,
1794 None,
1795 ClientMode::default(),
1796 None,
1797 )
1798 }
1799
1800 #[cfg(any(test, feature = "test-support"))]
1804 pub fn from_streams_with_connection_token(
1805 reader: impl AsyncRead + Unpin + Send + 'static,
1806 writer: impl AsyncWrite + Unpin + Send + 'static,
1807 cwd: PathBuf,
1808 token: Option<String>,
1809 ) -> Result<Self> {
1810 Self::from_transport(
1811 reader,
1812 writer,
1813 None,
1814 None,
1815 cwd,
1816 None,
1817 None,
1818 false,
1819 false,
1820 None,
1821 None,
1822 token,
1823 ClientMode::default(),
1824 None,
1825 )
1826 }
1827
1828 #[doc(hidden)]
1831 #[cfg(any(test, feature = "test-support"))]
1832 pub fn from_streams_with_github_telemetry(
1833 reader: impl AsyncRead + Unpin + Send + 'static,
1834 writer: impl AsyncWrite + Unpin + Send + 'static,
1835 cwd: PathBuf,
1836 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1837 ) -> Result<Self> {
1838 Self::from_transport(
1839 reader,
1840 writer,
1841 None,
1842 None,
1843 cwd,
1844 None,
1845 None,
1846 false,
1847 false,
1848 None,
1849 Some(on_github_telemetry),
1850 None,
1851 ClientMode::default(),
1852 None,
1853 )
1854 }
1855
1856 #[cfg(any(test, feature = "test-support"))]
1862 pub fn generate_connection_token_for_test() -> String {
1863 generate_connection_token()
1864 }
1865
1866 #[doc(hidden)]
1870 #[cfg(any(test, feature = "test-support"))]
1871 pub fn from_streams_with_client_info(
1872 reader: impl AsyncRead + Unpin + Send + 'static,
1873 writer: impl AsyncWrite + Unpin + Send + 'static,
1874 cwd: PathBuf,
1875 client_info: Option<ClientInfo>,
1876 ) -> Result<Self> {
1877 Self::from_transport(
1878 reader,
1879 writer,
1880 None,
1881 None,
1882 cwd,
1883 None,
1884 None,
1885 false,
1886 false,
1887 None,
1888 None,
1889 None,
1890 ClientMode::default(),
1891 client_info,
1892 )
1893 }
1894
1895 #[allow(clippy::too_many_arguments)]
1896 fn from_transport(
1897 reader: impl AsyncRead + Unpin + Send + 'static,
1898 writer: impl AsyncWrite + Unpin + Send + 'static,
1899 child: Option<Child>,
1900 process_tree: Option<process_tree::ProcessTree>,
1901 cwd: PathBuf,
1902 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1903 extension_launch_provider: Option<
1904 Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1905 >,
1906 session_fs_configured: bool,
1907 session_fs_sqlite_declared: bool,
1908 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1909 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1910 effective_connection_token: Option<String>,
1911 mode: ClientMode,
1912 client_info: Option<ClientInfo>,
1913 ) -> Result<Self> {
1914 let setup_start = Instant::now();
1915 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1916 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1917 let rpc = JsonRpcClient::new(
1918 writer,
1919 reader,
1920 notification_broadcast_tx.clone(),
1921 request_tx,
1922 );
1923
1924 let pid = child.as_ref().and_then(|c| c.id());
1925 info!(pid = ?pid, "copilot CLI client ready");
1926
1927 let github_token_registry = Arc::new(github_token::GitHubTokenRegistry::new());
1928 let extension_launch_provider = Arc::new(
1929 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(
1930 extension_launch_provider,
1931 ),
1932 );
1933 let installation_confirmation =
1934 Arc::new(installation_confirmation::InstallationConfirmationDispatcher::new());
1935 let client = Self {
1936 inner: Arc::new(ClientInner {
1937 child: parking_lot::Mutex::new(child),
1938 process_tree: parking_lot::Mutex::new(process_tree),
1939 #[cfg(feature = "in-process")]
1940 ffi_host: parking_lot::Mutex::new(None),
1941 rpc,
1942 cwd,
1943 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1944 notification_tx: notification_broadcast_tx,
1945 router: router::SessionRouter::new(),
1946 github_token_registry: github_token_registry.clone(),
1947 negotiated_protocol_version: OnceLock::new(),
1948 state: parking_lot::Mutex::new(ConnectionState::Connected),
1949 lifecycle_tx: broadcast::channel(256).0,
1950 on_list_models,
1951 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1952 session_fs_configured,
1953 session_fs_sqlite_declared,
1954 llm_inference: OnceLock::new(),
1955 extension_launch_provider: extension_launch_provider.clone(),
1956 installation_confirmation: installation_confirmation.clone(),
1957 on_github_telemetry,
1958 on_get_trace_context,
1959 effective_connection_token,
1960 mode,
1961 client_info,
1962 startup_timings: OnceLock::new(),
1963 }),
1964 };
1965 github_token_registry.set_client(Arc::downgrade(&client.inner));
1966 extension_launch_provider.set_client(Arc::downgrade(&client.inner));
1967 installation_confirmation.set_client(Arc::downgrade(&client.inner));
1968 client.spawn_lifecycle_dispatcher();
1969 debug!(
1970 elapsed_ms = setup_start.elapsed().as_millis(),
1971 pid = ?pid,
1972 "Client::from_transport setup complete"
1973 );
1974 Ok(client)
1975 }
1976
1977 fn spawn_lifecycle_dispatcher(&self) {
1981 let mut notif_rx = self.inner.notification_tx.subscribe();
1982 let lifecycle_tx = self.inner.lifecycle_tx.clone();
1983 tokio::spawn(async move {
1984 loop {
1985 match notif_rx.recv().await {
1986 Ok(notification) => {
1987 if notification.method != "session.lifecycle" {
1988 continue;
1989 }
1990 let Some(params) = notification.params.as_ref() else {
1991 continue;
1992 };
1993 let event: SessionLifecycleEvent =
1994 match serde_json::from_value(params.clone()) {
1995 Ok(e) => e,
1996 Err(e) => {
1997 warn!(
1998 error = %e,
1999 "failed to deserialize session.lifecycle notification"
2000 );
2001 continue;
2002 }
2003 };
2004 let _ = lifecycle_tx.send(event);
2007 }
2008 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
2009 warn!(missed = n, "lifecycle dispatcher lagged");
2010 }
2011 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
2012 }
2013 }
2014 });
2015 }
2016
2017 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
2018 let mut command = Command::new(program);
2019 command.kill_on_drop(true);
2020 for arg in &options.prefix_args {
2021 command.arg(arg);
2022 }
2023 if let Some(token) = &options.github_token {
2026 command.env("COPILOT_SDK_AUTH_TOKEN", token);
2027 }
2028 if let Some(telemetry) = &options.telemetry {
2031 command.env("COPILOT_OTEL_ENABLED", "true");
2032 if let Some(endpoint) = &telemetry.otlp_endpoint {
2033 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
2034 }
2035 if let Some(protocol) = telemetry.otlp_protocol {
2036 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
2037 }
2038 if let Some(path) = &telemetry.file_path {
2039 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
2040 }
2041 if let Some(exporter) = telemetry.exporter_type {
2042 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
2043 }
2044 if let Some(source) = &telemetry.source_name {
2045 command.env("COPILOT_OTEL_SOURCE_NAME", source);
2046 }
2047 if let Some(capture) = telemetry.capture_content {
2048 command.env(
2049 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2050 if capture { "true" } else { "false" },
2051 );
2052 }
2053 }
2054 if let Some(dir) = &options.base_directory {
2055 command.env("COPILOT_HOME", dir);
2056 }
2057 if options.mode == ClientMode::Empty {
2060 command.env("COPILOT_DISABLE_KEYTAR", "1");
2061 }
2062 if let Transport::Tcp {
2063 connection_token: Some(token),
2064 ..
2065 } = &options.transport
2066 {
2067 command.env("COPILOT_CONNECTION_TOKEN", token);
2068 }
2069 for (key, value) in &options.env {
2070 command.env(key, value);
2071 }
2072 for key in &options.env_remove {
2073 command.env_remove(key);
2074 }
2075 if options.mode != ClientMode::Empty {
2076 command.env("COPILOT_RUNTIME_PROCESS_FILE_LOGGING", "1");
2077 }
2078 command
2079 .current_dir(working_directory)
2080 .stdout(Stdio::piped())
2081 .stderr(Stdio::piped());
2082
2083 command
2084 }
2085
2086 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
2094 let mut args: Vec<&'static str> = Vec::new();
2095 if options.github_token.is_some() {
2096 args.push("--auth-token-env");
2097 args.push("COPILOT_SDK_AUTH_TOKEN");
2098 }
2099 let use_logged_in = options
2100 .use_logged_in_user
2101 .unwrap_or(options.github_token.is_none());
2102 if !use_logged_in {
2103 args.push("--no-auto-login");
2104 }
2105 args
2106 }
2107
2108 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
2112 match options.session_idle_timeout_seconds {
2113 Some(secs) if secs > 0 => {
2114 vec!["--session-idle-timeout".to_string(), secs.to_string()]
2115 }
2116 _ => Vec::new(),
2117 }
2118 }
2119
2120 fn remote_args(options: &ClientOptions) -> Vec<String> {
2121 if options.enable_remote_sessions {
2122 vec!["--remote".to_string()]
2123 } else {
2124 Vec::new()
2125 }
2126 }
2127
2128 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
2129 match options.log_level {
2130 Some(level) => vec!["--log-level", level.as_str()],
2131 None => Vec::new(),
2132 }
2133 }
2134
2135 fn spawn_stdio(
2136 program: &Path,
2137 options: &ClientOptions,
2138 working_directory: &Path,
2139 ) -> Result<(Child, Option<process_tree::ProcessTree>, Duration)> {
2140 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
2141 let mut command = Self::build_command(program, options, working_directory);
2142 command
2143 .args(["--server", "--stdio", "--no-auto-update"])
2144 .args(Self::log_level_args(options))
2145 .args(Self::auth_args(options))
2146 .args(Self::session_idle_timeout_args(options))
2147 .args(Self::remote_args(options))
2148 .args(&options.extra_args)
2149 .stdin(Stdio::piped());
2150 let spawn_start = Instant::now();
2151 let (child, tree) = process_tree::spawn(&mut command)?;
2152 let spawn_elapsed = spawn_start.elapsed();
2153 debug!(
2154 elapsed_ms = spawn_elapsed.as_millis(),
2155 "Client::spawn_stdio subprocess spawned"
2156 );
2157 Ok((child, tree, spawn_elapsed))
2158 }
2159
2160 async fn spawn_tcp(
2161 program: &Path,
2162 options: &ClientOptions,
2163 working_directory: &Path,
2164 port: u16,
2165 ) -> Result<(
2166 Child,
2167 Option<process_tree::ProcessTree>,
2168 u16,
2169 Duration,
2170 Duration,
2171 )> {
2172 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
2173 let mut command = Self::build_command(program, options, working_directory);
2174 command
2175 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
2176 .args(Self::log_level_args(options))
2177 .args(Self::auth_args(options))
2178 .args(Self::session_idle_timeout_args(options))
2179 .args(Self::remote_args(options))
2180 .args(&options.extra_args)
2181 .stdin(Stdio::null());
2182 let spawn_start = Instant::now();
2183 let (mut child, tree) = process_tree::spawn(&mut command)?;
2184 let spawn_elapsed = spawn_start.elapsed();
2185 debug!(
2186 elapsed_ms = spawn_elapsed.as_millis(),
2187 "Client::spawn_tcp subprocess spawned"
2188 );
2189 let stdout = child.stdout.take().expect("stdout is piped");
2190
2191 let (port_tx, port_rx) = oneshot::channel::<u16>();
2192 let span = tracing::error_span!("copilot_cli_port_scan");
2193 tokio::spawn(
2194 async move {
2195 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
2197 let mut lines = BufReader::new(stdout).lines();
2198 let mut port_tx = Some(port_tx);
2199 while let Ok(Some(line)) = lines.next_line().await {
2200 debug!(line = %line, "CLI stdout");
2201 if let Some(tx) = port_tx.take() {
2202 if let Some(caps) = port_re.captures(&line)
2203 && let Some(p) =
2204 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
2205 {
2206 let _ = tx.send(p);
2207 continue;
2208 }
2209 port_tx = Some(tx);
2211 }
2212 }
2213 }
2214 .instrument(span),
2215 );
2216
2217 let port_wait_start = Instant::now();
2218 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
2219 .await
2220 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
2221 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
2222
2223 let port_wait_elapsed = port_wait_start.elapsed();
2224 debug!(
2225 elapsed_ms = port_wait_elapsed.as_millis(),
2226 port = actual_port,
2227 "Client::spawn_tcp TCP port wait complete"
2228 );
2229 info!(port = %actual_port, "CLI server listening");
2230 Ok((child, tree, actual_port, spawn_elapsed, port_wait_elapsed))
2231 }
2232
2233 fn drain_stderr(child: &mut Child) {
2234 if let Some(stderr) = child.stderr.take() {
2235 let span = tracing::error_span!("copilot_cli");
2236 tokio::spawn(
2237 async move {
2238 let mut reader = BufReader::new(stderr).lines();
2239 while let Ok(Some(line)) = reader.next_line().await {
2240 warn!(line = %line, "CLI stderr");
2241 }
2242 }
2243 .instrument(span),
2244 );
2245 }
2246 }
2247
2248 pub fn cwd(&self) -> &PathBuf {
2250 &self.inner.cwd
2251 }
2252
2253 pub fn mode(&self) -> ClientMode {
2255 self.inner.mode
2256 }
2257
2258 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
2269 crate::generated::rpc::ClientRpc { client: self }
2270 }
2271
2272 #[allow(dead_code, reason = "convenience for future internal use")]
2274 pub(crate) async fn send_request(
2275 &self,
2276 method: &str,
2277 params: Option<serde_json::Value>,
2278 ) -> Result<JsonRpcResponse> {
2279 self.inner.rpc.send_request(method, params).await
2280 }
2281
2282 pub async fn call(
2302 &self,
2303 method: &str,
2304 params: Option<serde_json::Value>,
2305 ) -> Result<serde_json::Value> {
2306 self.call_with_inline_callback(method, params, None).await
2307 }
2308
2309 pub(crate) async fn detach_session(&self, session_id: &str) -> Result<()> {
2310 let value = self
2311 .call(
2312 "session.detach",
2313 Some(serde_json::json!({ "sessionId": session_id })),
2314 )
2315 .await?;
2316 let response: SessionDetachResponse = serde_json::from_value(value)?;
2317 if response.success {
2318 return Ok(());
2319 }
2320 Err(Error::with_message(
2321 ErrorKind::Session(SessionErrorKind::DetachFailed),
2322 response
2323 .error
2324 .unwrap_or_else(|| "unknown error".to_string()),
2325 ))
2326 }
2327
2328 pub(crate) async fn call_with_inline_callback(
2343 &self,
2344 method: &str,
2345 params: Option<serde_json::Value>,
2346 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
2347 ) -> Result<serde_json::Value> {
2348 let session_id: Option<SessionId> = params
2349 .as_ref()
2350 .and_then(|p| p.get("sessionId"))
2351 .and_then(|v| v.as_str())
2352 .map(SessionId::from);
2353 let response = self
2354 .inner
2355 .rpc
2356 .send_request_with_inline_callback(method, params, inline_callback)
2357 .await?;
2358 if let Some(err) = response.error {
2359 if err.message.contains("Session not found") {
2360 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
2361 session_id.unwrap_or_else(|| "unknown".into()),
2362 ))
2363 .into());
2364 }
2365 return Err(Error::from_rpc(err.code, err.message, err.data));
2366 }
2367 Ok(response.result.unwrap_or(serde_json::Value::Null))
2368 }
2369
2370 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
2372 self.inner.rpc.write(response).await
2373 }
2374
2375 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
2377 Self { inner }
2378 }
2379
2380 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
2384 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
2385 self.inner.request_rx.lock().take()
2386 }
2387
2388 pub(crate) fn register_session(
2399 &self,
2400 session_id: &SessionId,
2401 ) -> crate::router::SessionRegistration {
2402 self.inner.router.ensure_started(&self.inner);
2403 self.inner.router.register(session_id)
2404 }
2405
2406 pub(crate) fn unregister_session_owned(
2414 &self,
2415 session_id: &SessionId,
2416 token: crate::router::RegistrationToken,
2417 ) {
2418 self.inner.router.unregister_owned(session_id, token);
2419 }
2420
2421 #[cfg(any(test, feature = "test-support"))]
2428 pub(crate) fn registered_session_ids(&self) -> Vec<SessionId> {
2429 self.inner.router.session_ids()
2430 }
2431
2432 pub(crate) fn register_github_token_provider(
2433 &self,
2434 provider: Arc<dyn GitHubTokenProvider>,
2435 ) -> github_token::GitHubTokenRegistration {
2436 self.inner.router.ensure_started(&self.inner);
2437 let id = self.inner.github_token_registry.register(provider);
2438 github_token::GitHubTokenRegistration::new(self.inner.github_token_registry.clone(), id)
2439 }
2440
2441 pub(crate) fn retire_github_token_provider(&self, session_id: &SessionId) {
2442 self.inner.github_token_registry.retire_session(session_id);
2443 }
2444
2445 pub fn protocol_version(&self) -> Option<u32> {
2452 self.inner.negotiated_protocol_version.get().copied()
2453 }
2454
2455 pub fn startup_timings(&self) -> Option<StartupTimings> {
2462 self.inner.startup_timings.get().cloned()
2463 }
2464
2465 pub async fn verify_protocol_version(&self) -> Result<()> {
2489 let handshake_start = Instant::now();
2490 let mut used_fallback_ping = false;
2491 let server_version = match self.connect_handshake().await {
2495 Ok(v) => v,
2496 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2497 used_fallback_ping = true;
2498 self.ping(None).await?.protocol_version
2499 }
2500 Err(e) => return Err(e),
2501 };
2502
2503 match server_version {
2504 None => {
2505 warn!("CLI server did not report protocolVersion; skipping version check");
2506 }
2507 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2508 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2509 server: v,
2510 min: MIN_PROTOCOL_VERSION,
2511 max: SDK_PROTOCOL_VERSION,
2512 })
2513 .into());
2514 }
2515 Some(v) => {
2516 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2517 if existing != v {
2518 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2519 previous: existing,
2520 current: v,
2521 })
2522 .into());
2523 }
2524 } else {
2525 let _ = self.inner.negotiated_protocol_version.set(v);
2526 }
2527 }
2528 }
2529
2530 debug!(
2531 elapsed_ms = handshake_start.elapsed().as_millis(),
2532 protocol_version = ?server_version,
2533 used_fallback_ping,
2534 "Client::verify_protocol_version protocol handshake complete"
2535 );
2536 Ok(())
2537 }
2538
2539 async fn connect_handshake(&self) -> Result<Option<u32>> {
2546 let params = crate::generated::api_types::ConnectRequest {
2547 token: self.inner.effective_connection_token.clone(),
2548 enable_git_hub_telemetry_forwarding: self
2549 .inner
2550 .on_github_telemetry
2551 .is_some()
2552 .then_some(true),
2553 supported_task_kinds: Some(vec![
2554 crate::generated::api_types::TaskKind::Agent,
2555 crate::generated::api_types::TaskKind::Client,
2556 crate::generated::api_types::TaskKind::Shell,
2557 ]),
2558 client_info: self
2563 .inner
2564 .client_info
2565 .as_ref()
2566 .and_then(ClientInfo::to_wire),
2567 };
2568 let value = self
2569 .call(
2570 crate::generated::api_types::rpc_methods::CONNECT,
2571 Some(serde_json::to_value(params)?),
2572 )
2573 .await?;
2574 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2575 Ok(Some(u32::try_from(result.protocol_version).map_err(
2576 |_| ProtocolErrorKind::InvalidProtocolVersion {
2577 server: result.protocol_version,
2578 },
2579 )?))
2580 }
2581
2582 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2590 let params = match message {
2591 Some(m) => serde_json::json!({ "message": m }),
2592 None => serde_json::json!({}),
2593 };
2594 let value = self
2595 .call(generated::api_types::rpc_methods::PING, Some(params))
2596 .await?;
2597 Ok(serde_json::from_value(value)?)
2598 }
2599
2600 pub async fn list_sessions(
2603 &self,
2604 filter: Option<SessionListFilter>,
2605 ) -> Result<Vec<SessionMetadata>> {
2606 let params = match filter {
2607 Some(f) => serde_json::json!({ "filter": f }),
2608 None => serde_json::json!({}),
2609 };
2610 let result = self.call("session.list", Some(params)).await?;
2611 let response: ListSessionsResponse = serde_json::from_value(result)?;
2612 Ok(response.sessions)
2613 }
2614
2615 pub async fn get_session_metadata(
2633 &self,
2634 session_id: &SessionId,
2635 ) -> Result<Option<SessionMetadata>> {
2636 let result = self
2637 .call(
2638 "session.getMetadata",
2639 Some(serde_json::json!({ "sessionId": session_id })),
2640 )
2641 .await?;
2642 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2643 Ok(response.session)
2644 }
2645
2646 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2648 self.call(
2649 "session.delete",
2650 Some(serde_json::json!({ "sessionId": session_id })),
2651 )
2652 .await?;
2653 self.retire_github_token_provider(session_id);
2654 Ok(())
2655 }
2656
2657 #[cfg(feature = "test-support")]
2660 #[doc(hidden)]
2661 pub fn start_router_for_test(&self) {
2662 self.inner.router.ensure_started(&self.inner);
2663 }
2664
2665 #[cfg(feature = "test-support")]
2666 #[doc(hidden)]
2667 pub fn registered_session_ids_for_test(&self) -> Vec<SessionId> {
2671 self.registered_session_ids()
2672 }
2673
2674 #[cfg(feature = "test-support")]
2675 #[doc(hidden)]
2676 pub fn registered_session_count_for_test(&self) -> usize {
2680 self.inner.router.session_count()
2681 }
2682
2683 #[cfg(feature = "test-support")]
2684 #[doc(hidden)]
2685 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2688 let mut first_error = None;
2689
2690 for session_id in self.inner.router.session_ids() {
2691 if let Err(error) = self.detach_session(&session_id).await
2692 && first_error.is_none()
2693 {
2694 first_error = Some(error);
2695 }
2696 self.inner.router.unregister(&session_id);
2697 }
2698 self.inner.github_token_registry.clear();
2699
2700 match self.list_sessions(None).await {
2701 Ok(sessions) => {
2702 for session in sessions {
2703 if let Err(error) = self.delete_session(&session.session_id).await
2704 && first_error.is_none()
2705 {
2706 first_error = Some(error);
2707 }
2708 }
2709 }
2710 Err(error) if first_error.is_none() => first_error = Some(error),
2711 Err(_) => {}
2712 }
2713
2714 match first_error {
2715 Some(error) => Err(error),
2716 None => Ok(()),
2717 }
2718 }
2719
2720 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2736 let result = self
2737 .call("session.getLastId", Some(serde_json::json!({})))
2738 .await?;
2739 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2740 Ok(response.session_id)
2741 }
2742
2743 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2748 let result = self
2749 .call("session.getForeground", Some(serde_json::json!({})))
2750 .await?;
2751 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2752 Ok(response.session_id)
2753 }
2754
2755 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2760 self.call(
2761 "session.setForeground",
2762 Some(serde_json::json!({ "sessionId": session_id })),
2763 )
2764 .await?;
2765 Ok(())
2766 }
2767
2768 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2770 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2771 Ok(serde_json::from_value(result)?)
2772 }
2773
2774 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2776 let result = self
2777 .call("auth.getStatus", Some(serde_json::json!({})))
2778 .await?;
2779 Ok(serde_json::from_value(result)?)
2780 }
2781
2782 pub async fn list_models(&self) -> Result<Vec<Model>> {
2787 let cache = self.inner.models_cache.lock().clone();
2788 let models = cache
2789 .get_or_try_init(|| async {
2790 if let Some(handler) = &self.inner.on_list_models {
2791 handler.list_models().await
2792 } else {
2793 Ok(self.rpc().models().list().await?.models)
2794 }
2795 })
2796 .await?;
2797 Ok(models.clone())
2798 }
2799
2800 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2803 if let Some(provider) = &self.inner.on_get_trace_context {
2804 provider.get_trace_context().await
2805 } else {
2806 TraceContext::default()
2807 }
2808 }
2809
2810 pub fn pid(&self) -> Option<u32> {
2812 self.inner.child.lock().as_ref().and_then(|c| c.id())
2813 }
2814
2815 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2843 let pid = self.pid();
2844 info!(pid = ?pid, "stopping CLI process");
2845 let mut errors: Vec<Error> = Vec::new();
2846 self.inner.extension_launch_provider.clear();
2847 self.inner.installation_confirmation.clear();
2848
2849 for session_id in self.inner.router.session_ids() {
2852 match self.detach_session(&session_id).await {
2853 Ok(_) => {}
2854 Err(e) => {
2855 warn!(
2856 session_id = %session_id,
2857 error = %e,
2858 "session.detach failed during Client::stop",
2859 );
2860 errors.push(e);
2861 }
2862 }
2863 self.inner.router.unregister(&session_id);
2864 }
2865 self.inner.github_token_registry.clear();
2866
2867 let should_shutdown_runtime = self.inner.child.lock().is_some();
2868 #[cfg(feature = "in-process")]
2869 let should_shutdown_runtime =
2870 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2871 if should_shutdown_runtime {
2872 let runtime_shutdown_start = Instant::now();
2873 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2874 .await
2875 {
2876 Ok(Ok(())) => {
2877 debug!(
2878 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2879 "Client::stop runtime shutdown complete"
2880 );
2881 }
2882 Ok(Err(e)) => {
2883 warn!(
2884 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2885 error = %e,
2886 "runtime.shutdown failed during Client::stop",
2887 );
2888 errors.push(e);
2889 }
2890 Err(_) => {
2891 let e = std::io::Error::new(
2892 std::io::ErrorKind::TimedOut,
2893 "runtime.shutdown timed out during Client::stop",
2894 );
2895 warn!(
2896 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2897 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2898 error = %e,
2899 "runtime.shutdown timed out during Client::stop",
2900 );
2901 errors.push(e.into());
2902 }
2903 }
2904 }
2905
2906 let child = self.inner.child.lock().take();
2907 let process_tree = self.inner.process_tree.lock().take();
2908 *self.inner.state.lock() = ConnectionState::Disconnected;
2909 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2910 if let Some(process_tree) = process_tree
2911 && let Err(error) = process_tree.terminate()
2912 {
2913 errors.push(error.into());
2914 }
2915 if let Some(mut child) = child {
2916 match child.try_wait() {
2917 Ok(Some(_status)) => {}
2918 Ok(None) => {
2919 if let Err(e) = child.kill().await {
2926 errors.push(e.into());
2927 }
2928 }
2929 Err(e) => errors.push(e.into()),
2930 }
2931 }
2932
2933 self.inner.rpc.force_close();
2936 #[cfg(feature = "in-process")]
2937 {
2938 if let Some(host) = self.inner.ffi_host.lock().take() {
2939 host.close();
2940 }
2941 }
2942
2943 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2944 if errors.is_empty() {
2945 Ok(())
2946 } else {
2947 Err(StopErrors(errors))
2948 }
2949 }
2950
2951 pub fn force_stop(&self) {
2980 let pid = self.pid();
2981 info!(pid = ?pid, "force-stopping CLI process");
2982 self.inner.extension_launch_provider.clear();
2983 self.inner.installation_confirmation.clear();
2984 if let Some(process_tree) = self.inner.process_tree.lock().take()
2985 && let Err(error) = process_tree.terminate()
2986 {
2987 error!(pid = ?pid, %error, "failed to terminate CLI process tree");
2988 }
2989 if let Some(mut child) = self.inner.child.lock().take()
2990 && let Err(e) = child.start_kill()
2991 {
2992 error!(pid = ?pid, error = %e, "failed to send kill signal");
2993 }
2994 self.inner.rpc.force_close();
2995 #[cfg(feature = "in-process")]
2996 {
2997 if let Some(host) = self.inner.ffi_host.lock().take() {
2998 host.close();
2999 }
3000 }
3001 self.inner.router.clear();
3004 self.inner.github_token_registry.clear();
3005 *self.inner.state.lock() = ConnectionState::Disconnected;
3006 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
3007 }
3008
3009 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
3044 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
3045 }
3046}
3047
3048impl Drop for ClientInner {
3049 fn drop(&mut self) {
3050 self.rpc.force_close();
3051 let pid = self.child.lock().as_ref().and_then(Child::id);
3052 if let Some(process_tree) = self.process_tree.lock().take()
3053 && let Err(error) = process_tree.terminate()
3054 {
3055 error!(pid = ?pid, %error, "failed to terminate CLI process tree on drop");
3056 }
3057 if let Some(ref mut child) = *self.child.lock() {
3058 if let Err(e) = child.start_kill() {
3059 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
3060 } else {
3061 info!(pid = ?pid, "kill signal sent for CLI process on drop");
3062 }
3063 }
3064 #[cfg(feature = "in-process")]
3065 {
3066 if let Some(host) = self.ffi_host.lock().take() {
3067 host.close();
3068 }
3069 }
3070 }
3071}
3072
3073#[cfg(test)]
3074mod tests {
3075 use super::*;
3076
3077 #[test]
3078 fn is_transport_failure_matches_request_cancelled() {
3079 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
3080 assert!(err.is_transport_failure());
3081 }
3082
3083 #[test]
3084 fn is_transport_failure_matches_io_error() {
3085 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
3086 assert!(err.is_transport_failure());
3087 }
3088
3089 #[test]
3090 fn is_transport_failure_rejects_rpc_error() {
3091 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
3092 assert!(!err.is_transport_failure());
3093 }
3094
3095 #[test]
3096 fn is_transport_failure_rejects_session_error() {
3097 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
3098 assert!(!err.is_transport_failure());
3099 }
3100
3101 #[test]
3102 fn client_options_builder_composes() {
3103 let opts = ClientOptions::new()
3104 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
3105 .with_prefix_args(["node"])
3106 .with_cwd(PathBuf::from("/tmp"))
3107 .with_env([("KEY", "value")])
3108 .with_env_remove(["UNWANTED"])
3109 .with_extra_args(["--quiet"])
3110 .with_github_token("ghp_test")
3111 .with_use_logged_in_user(false)
3112 .with_log_level(LogLevel::Debug)
3113 .with_session_idle_timeout_seconds(120)
3114 .with_enable_remote_sessions(true);
3115 assert!(matches!(opts.program, CliProgram::Path(_)));
3116 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
3117 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
3118 assert_eq!(
3119 opts.env,
3120 vec![(
3121 std::ffi::OsString::from("KEY"),
3122 std::ffi::OsString::from("value")
3123 )]
3124 );
3125 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
3126 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
3127 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
3128 assert_eq!(opts.use_logged_in_user, Some(false));
3129 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
3130 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
3131 assert!(opts.enable_remote_sessions);
3132 }
3133
3134 #[test]
3135 fn default_transport_values_resolve_without_process_state() {
3136 assert!(matches!(
3137 resolve_default_transport_value(None).unwrap(),
3138 Transport::Stdio
3139 ));
3140 assert!(matches!(
3141 resolve_default_transport_value(Some("stdio")).unwrap(),
3142 Transport::Stdio
3143 ));
3144 assert!(matches!(
3145 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
3146 Transport::InProcess
3147 ));
3148 assert!(resolve_default_transport_value(Some("tcp")).is_err());
3149 }
3150
3151 #[test]
3152 fn inprocess_rejects_process_scoped_options() {
3153 let invalid = [
3154 ClientOptions::new().with_cwd("."),
3155 ClientOptions::new().with_env([("KEY", "value")]),
3156 ClientOptions::new().with_env_remove(["KEY"]),
3157 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
3158 ClientOptions::new().with_prefix_args(["index.js"]),
3159 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
3160 ClientOptions::new().with_extra_args(["--verbose"]),
3161 ];
3162
3163 for options in invalid {
3164 assert!(validate_inprocess_options(&options).is_err());
3165 }
3166 }
3167
3168 #[test]
3169 fn inprocess_allows_typed_runtime_options() {
3170 let options = ClientOptions::new()
3171 .with_base_directory("state")
3172 .with_log_level(LogLevel::Debug)
3173 .with_session_idle_timeout_seconds(10)
3174 .with_github_token("token")
3175 .with_use_logged_in_user(false)
3176 .with_enable_remote_sessions(true);
3177
3178 assert!(validate_inprocess_options(&options).is_ok());
3179 }
3180
3181 #[cfg(not(feature = "in-process"))]
3182 #[tokio::test]
3183 async fn inprocess_requires_cargo_feature() {
3184 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
3185 .await
3186 .unwrap_err();
3187
3188 assert!(error.to_string().contains("in-process"));
3189 }
3190
3191 #[test]
3192 fn is_transport_failure_rejects_other_protocol_errors() {
3193 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
3194 assert!(!err.is_transport_failure());
3195 }
3196
3197 #[test]
3198 fn build_command_lets_env_remove_strip_injected_token() {
3199 let opts = ClientOptions {
3200 github_token: Some("secret".to_string()),
3201 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
3202 ..Default::default()
3203 };
3204 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3205 let action = cmd
3207 .as_std()
3208 .get_envs()
3209 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3210 .map(|(_, v)| v);
3211 assert_eq!(
3212 action,
3213 Some(None),
3214 "env_remove should win over github_token"
3215 );
3216 }
3217
3218 #[test]
3219 fn build_command_lets_env_override_injected_token() {
3220 let opts = ClientOptions {
3221 github_token: Some("from-options".to_string()),
3222 env: vec![(
3223 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
3224 std::ffi::OsString::from("from-env"),
3225 )],
3226 ..Default::default()
3227 };
3228 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3229 let value = cmd
3230 .as_std()
3231 .get_envs()
3232 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3233 .and_then(|(_, v)| v);
3234 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
3235 }
3236
3237 #[test]
3238 fn build_command_injects_github_token_by_default() {
3239 let opts = ClientOptions {
3240 github_token: Some("just-the-token".to_string()),
3241 ..Default::default()
3242 };
3243 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3244 let value = cmd
3245 .as_std()
3246 .get_envs()
3247 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
3248 .and_then(|(_, v)| v);
3249 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
3250 }
3251
3252 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
3253 cmd.as_std()
3254 .get_envs()
3255 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
3256 .and_then(|(_, v)| v)
3257 }
3258
3259 #[test]
3260 fn telemetry_config_builder_composes() {
3261 let cfg = TelemetryConfig::new()
3262 .with_otlp_endpoint("http://collector:4318")
3263 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
3264 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
3265 .with_exporter_type(OtelExporterType::OtlpHttp)
3266 .with_source_name("my-app")
3267 .with_capture_content(true);
3268
3269 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
3270 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
3271 assert_eq!(
3272 cfg.file_path.as_deref(),
3273 Some(Path::new("/var/log/copilot.jsonl")),
3274 );
3275 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
3276 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
3277 assert_eq!(cfg.capture_content, Some(true));
3278 assert!(!cfg.is_empty());
3279 assert!(TelemetryConfig::new().is_empty());
3280 }
3281
3282 #[test]
3283 fn otlp_http_protocol_serde_matches_env_value() {
3284 for (protocol, wire) in [
3285 (OtlpHttpProtocol::HttpJson, "http/json"),
3286 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
3287 ] {
3288 assert_eq!(protocol.as_str(), wire);
3289
3290 let serialized = serde_json::to_string(&protocol).unwrap();
3291 assert_eq!(serialized, format!("\"{wire}\""));
3292
3293 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
3294 assert_eq!(deserialized, protocol);
3295 }
3296 }
3297
3298 #[test]
3299 fn build_command_sets_otel_env_when_telemetry_enabled() {
3300 let opts = ClientOptions {
3301 telemetry: Some(TelemetryConfig {
3302 otlp_endpoint: Some("http://collector:4318".to_string()),
3303 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
3304 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
3305 exporter_type: Some(OtelExporterType::OtlpHttp),
3306 source_name: Some("my-app".to_string()),
3307 capture_content: Some(true),
3308 }),
3309 ..Default::default()
3310 };
3311 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3312 assert_eq!(
3313 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
3314 Some(std::ffi::OsStr::new("true")),
3315 );
3316 assert_eq!(
3317 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3318 Some(std::ffi::OsStr::new("http://collector:4318")),
3319 );
3320 assert_eq!(
3321 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
3322 Some(std::ffi::OsStr::new("http/protobuf")),
3323 );
3324 assert_eq!(
3325 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
3326 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
3327 );
3328 assert_eq!(
3329 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
3330 Some(std::ffi::OsStr::new("otlp-http")),
3331 );
3332 assert_eq!(
3333 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
3334 Some(std::ffi::OsStr::new("my-app")),
3335 );
3336 assert_eq!(
3337 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
3338 Some(std::ffi::OsStr::new("true")),
3339 );
3340 }
3341
3342 #[test]
3343 fn build_command_omits_otel_env_when_telemetry_none() {
3344 let opts = ClientOptions::default();
3345 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3346 for key in [
3347 "COPILOT_OTEL_ENABLED",
3348 "OTEL_EXPORTER_OTLP_ENDPOINT",
3349 "OTEL_EXPORTER_OTLP_PROTOCOL",
3350 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3351 "COPILOT_OTEL_EXPORTER_TYPE",
3352 "COPILOT_OTEL_SOURCE_NAME",
3353 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3354 ] {
3355 assert!(
3356 env_value(&cmd, key).is_none(),
3357 "expected {key} to be unset when telemetry is None",
3358 );
3359 }
3360 }
3361
3362 #[test]
3363 fn build_command_omits_unset_telemetry_fields() {
3364 let opts = ClientOptions {
3365 telemetry: Some(TelemetryConfig {
3366 otlp_endpoint: Some("http://collector:4318".to_string()),
3367 ..Default::default()
3368 }),
3369 ..Default::default()
3370 };
3371 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3372 assert_eq!(
3374 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
3375 Some(std::ffi::OsStr::new("true")),
3376 );
3377 assert_eq!(
3378 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3379 Some(std::ffi::OsStr::new("http://collector:4318")),
3380 );
3381 for key in [
3383 "OTEL_EXPORTER_OTLP_PROTOCOL",
3384 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3385 "COPILOT_OTEL_EXPORTER_TYPE",
3386 "COPILOT_OTEL_SOURCE_NAME",
3387 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3388 ] {
3389 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
3390 }
3391 }
3392
3393 #[test]
3394 fn build_command_lets_user_env_override_telemetry() {
3395 let opts = ClientOptions {
3396 telemetry: Some(TelemetryConfig {
3397 otlp_endpoint: Some("http://from-config:4318".to_string()),
3398 ..Default::default()
3399 }),
3400 env: vec![(
3401 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
3402 std::ffi::OsString::from("http://from-user-env:4318"),
3403 )],
3404 ..Default::default()
3405 };
3406 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3407 assert_eq!(
3408 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3409 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
3410 "user-supplied options.env should override telemetry config",
3411 );
3412 }
3413
3414 #[test]
3415 fn build_command_sets_copilot_home_env_when_configured() {
3416 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
3417 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3418 assert_eq!(
3419 env_value(&cmd, "COPILOT_HOME"),
3420 Some(std::ffi::OsStr::new("/custom/copilot")),
3421 );
3422
3423 let opts = ClientOptions::default();
3424 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3425 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
3426 }
3427
3428 #[test]
3429 fn build_command_sets_connection_token_env_when_configured() {
3430 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3431 port: 0,
3432 connection_token: Some("secret-token".to_string()),
3433 });
3434 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3435 assert_eq!(
3436 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3437 Some(std::ffi::OsStr::new("secret-token")),
3438 );
3439
3440 let opts = ClientOptions::default();
3441 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3442 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3443 }
3444
3445 #[tokio::test]
3446 async fn start_rejects_empty_connection_token() {
3447 let opts = ClientOptions::new()
3448 .with_transport(Transport::Tcp {
3449 port: 0,
3450 connection_token: Some(String::new()),
3451 })
3452 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3453 let err = Client::start(opts).await.unwrap_err();
3454 assert!(
3455 matches!(err.kind(), ErrorKind::InvalidConfig),
3456 "got {err:?}"
3457 );
3458 }
3459
3460 #[tokio::test]
3461 async fn start_rejects_empty_external_connection_token() {
3462 let opts = ClientOptions::new()
3463 .with_transport(Transport::External {
3464 host: "127.0.0.1".to_string(),
3465 port: 1,
3466 connection_token: Some(String::new()),
3467 })
3468 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3469 let err = Client::start(opts).await.unwrap_err();
3470 assert!(
3471 matches!(err.kind(), ErrorKind::InvalidConfig),
3472 "got {err:?}"
3473 );
3474 }
3475
3476 #[test]
3477 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3478 let opts_true = ClientOptions {
3479 telemetry: Some(TelemetryConfig {
3480 capture_content: Some(true),
3481 ..Default::default()
3482 }),
3483 ..Default::default()
3484 };
3485 let opts_false = ClientOptions {
3486 telemetry: Some(TelemetryConfig {
3487 capture_content: Some(false),
3488 ..Default::default()
3489 }),
3490 ..Default::default()
3491 };
3492 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3493 let cmd_false =
3494 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3495 assert_eq!(
3496 env_value(
3497 &cmd_true,
3498 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3499 ),
3500 Some(std::ffi::OsStr::new("true")),
3501 );
3502 assert_eq!(
3503 env_value(
3504 &cmd_false,
3505 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3506 ),
3507 Some(std::ffi::OsStr::new("false")),
3508 );
3509 }
3510
3511 #[test]
3512 fn session_idle_timeout_args_are_omitted_by_default() {
3513 let opts = ClientOptions::default();
3514 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3515 }
3516
3517 #[test]
3518 fn session_idle_timeout_args_omitted_for_zero() {
3519 let opts = ClientOptions {
3520 session_idle_timeout_seconds: Some(0),
3521 ..Default::default()
3522 };
3523 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3524 }
3525
3526 #[test]
3527 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3528 let opts = ClientOptions {
3529 session_idle_timeout_seconds: Some(300),
3530 ..Default::default()
3531 };
3532 assert_eq!(
3533 Client::session_idle_timeout_args(&opts),
3534 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3535 );
3536 }
3537
3538 #[test]
3539 fn remote_args_omitted_by_default() {
3540 let opts = ClientOptions::default();
3541 assert!(Client::remote_args(&opts).is_empty());
3542 }
3543
3544 #[test]
3545 fn remote_args_emit_flag_when_enabled() {
3546 let opts = ClientOptions {
3547 enable_remote_sessions: true,
3548 ..Default::default()
3549 };
3550 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3551 }
3552
3553 #[test]
3554 fn log_level_args_omitted_when_unset() {
3555 let opts = ClientOptions::default();
3556 assert!(opts.log_level.is_none());
3557 assert!(
3558 Client::log_level_args(&opts).is_empty(),
3559 "with no caller-supplied log_level the SDK must not pass --log-level"
3560 );
3561 }
3562
3563 #[test]
3564 fn log_level_args_emit_flag_when_set() {
3565 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3566 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3567 }
3568
3569 #[test]
3570 fn cli_mode_opts_into_process_logging_without_changing_empty_mode_environment() {
3571 for (mode, expected) in [
3572 (ClientMode::Empty, None),
3573 (ClientMode::CopilotCli, Some("1")),
3574 ] {
3575 let mut options = ClientOptions::default().with_mode(mode);
3576 options.env.push((
3577 std::ffi::OsString::from("COPILOT_RUNTIME_PROCESS_FILE_LOGGING"),
3578 std::ffi::OsString::from("opposite"),
3579 ));
3580 let command =
3581 Client::build_command(Path::new("copilot-runtime"), &options, Path::new("."));
3582 let actual = command
3583 .as_std()
3584 .get_envs()
3585 .find(|(key, _)| {
3586 *key == std::ffi::OsStr::new("COPILOT_RUNTIME_PROCESS_FILE_LOGGING")
3587 })
3588 .and_then(|(_, value)| value);
3589 assert_eq!(
3590 actual,
3591 Some(std::ffi::OsStr::new(expected.unwrap_or("opposite"))),
3592 "mode: {mode:?}"
3593 );
3594 }
3595 }
3596
3597 #[test]
3598 fn log_level_str_round_trips() {
3599 for level in [
3600 LogLevel::None,
3601 LogLevel::Error,
3602 LogLevel::Warning,
3603 LogLevel::Info,
3604 LogLevel::Debug,
3605 LogLevel::All,
3606 ] {
3607 let s = level.as_str();
3608 let json = serde_json::to_string(&level).unwrap();
3609 assert_eq!(json, format!("\"{s}\""));
3610 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3611 assert_eq!(parsed, level);
3612 }
3613 }
3614
3615 #[test]
3616 fn client_options_debug_redacts_handler() {
3617 struct StubHandler;
3618 #[async_trait]
3619 impl ListModelsHandler for StubHandler {
3620 async fn list_models(&self) -> Result<Vec<Model>> {
3621 Ok(vec![])
3622 }
3623 }
3624 let opts = ClientOptions {
3625 on_list_models: Some(Arc::new(StubHandler)),
3626 github_token: Some("secret-token".into()),
3627 ..Default::default()
3628 };
3629 let debug = format!("{opts:?}");
3630 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3631 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3632 assert!(!debug.contains("secret-token"));
3633 }
3634
3635 #[tokio::test]
3636 async fn list_models_uses_on_list_models_handler_when_set() {
3637 use std::sync::atomic::{AtomicUsize, Ordering};
3638
3639 struct CountingHandler {
3640 calls: Arc<AtomicUsize>,
3641 models: Vec<Model>,
3642 }
3643 #[async_trait]
3644 impl ListModelsHandler for CountingHandler {
3645 async fn list_models(&self) -> Result<Vec<Model>> {
3646 self.calls.fetch_add(1, Ordering::SeqCst);
3647 Ok(self.models.clone())
3648 }
3649 }
3650
3651 let calls = Arc::new(AtomicUsize::new(0));
3652 let model = Model {
3653 id: "byok-gpt-4".into(),
3654 name: "BYOK GPT-4".into(),
3655 ..Default::default()
3656 };
3657 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3658 calls: Arc::clone(&calls),
3659 models: vec![model.clone()],
3660 });
3661
3662 let client = client_with_list_models_handler(handler);
3663
3664 let result = client.list_models().await.unwrap();
3665 assert_eq!(result.len(), 1);
3666 assert_eq!(result[0].id, "byok-gpt-4");
3667 assert_eq!(calls.load(Ordering::SeqCst), 1);
3668 }
3669
3670 #[tokio::test]
3671 async fn list_models_serializes_concurrent_cache_misses() {
3672 use std::sync::atomic::{AtomicUsize, Ordering};
3673
3674 struct SlowCountingHandler {
3675 calls: Arc<AtomicUsize>,
3676 models: Vec<Model>,
3677 }
3678 #[async_trait]
3679 impl ListModelsHandler for SlowCountingHandler {
3680 async fn list_models(&self) -> Result<Vec<Model>> {
3681 self.calls.fetch_add(1, Ordering::SeqCst);
3682 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3683 Ok(self.models.clone())
3684 }
3685 }
3686
3687 let calls = Arc::new(AtomicUsize::new(0));
3688 let model = Model {
3689 id: "single-flight-model".into(),
3690 name: "Single Flight Model".into(),
3691 ..Default::default()
3692 };
3693 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3694 calls: Arc::clone(&calls),
3695 models: vec![model],
3696 });
3697 let client = client_with_list_models_handler(handler);
3698
3699 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3700 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3701 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3702 assert_eq!(calls.load(Ordering::SeqCst), 1);
3703 }
3704
3705 #[tokio::test]
3706 async fn cancelled_resume_session_unregisters_pending_session() {
3707 let (client_write, _server_read) = tokio::io::duplex(8192);
3708 let (_server_write, client_read) = tokio::io::duplex(8192);
3709 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3710 assert!(client.startup_timings().is_none());
3711 let session_id = SessionId::new("resume-cancel-test");
3712 let handle = tokio::spawn({
3713 let client = client.clone();
3714 async move {
3715 client
3716 .resume_session(ResumeSessionConfig::new(session_id))
3717 .await
3718 }
3719 });
3720
3721 wait_for_pending_session_registration(&client).await;
3722 handle.abort();
3723 let _ = handle.await;
3724
3725 assert!(client.inner.router.session_ids().is_empty());
3726 client.force_stop();
3727 }
3728
3729 #[cfg(any(unix, windows))]
3730 #[tokio::test]
3731 async fn dropping_last_client_kills_spawned_cli() {
3732 let temp = tempfile::tempdir().unwrap();
3733 let ready = temp.path().join("ready");
3734 let survived = temp.path().join("survived");
3735 let child = test_child_command(temp.path(), &ready, &survived)
3736 .spawn()
3737 .unwrap();
3738 let (client_write, _server_read) = tokio::io::duplex(64);
3739 let (_server_write, client_read) = tokio::io::duplex(64);
3740 let client = Client::from_transport(
3741 client_read,
3742 client_write,
3743 Some(child),
3744 None,
3745 temp.path().to_path_buf(),
3746 None,
3747 None,
3748 false,
3749 false,
3750 None,
3751 None,
3752 None,
3753 ClientMode::default(),
3754 None,
3755 )
3756 .unwrap();
3757
3758 wait_for_test_child(&ready).await;
3759 drop(client);
3760
3761 assert_test_child_killed(&survived).await;
3762 }
3763
3764 #[cfg(any(unix, windows))]
3765 #[tokio::test]
3766 async fn spawned_child_is_killed_when_dropped() {
3767 let temp = tempfile::tempdir().unwrap();
3768 let ready = temp.path().join("ready");
3769 let survived = temp.path().join("survived");
3770 let child = test_child_command(temp.path(), &ready, &survived)
3771 .spawn()
3772 .unwrap();
3773
3774 wait_for_test_child(&ready).await;
3775 drop(child);
3776
3777 assert_test_child_killed(&survived).await;
3778 }
3779
3780 #[cfg(any(unix, windows))]
3781 fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command {
3782 let mut command = Client::build_command(Path::new("node"), &ClientOptions::default(), temp);
3783 #[cfg(windows)]
3784 {
3785 const CREATE_NO_WINDOW: u32 = 0x0800_0000;
3786 command.creation_flags(CREATE_NO_WINDOW);
3787 }
3788 command
3789 .args([
3790 "-e",
3791 r#"
3792 const fs = require("node:fs");
3793 fs.writeFileSync(process.env.READY, "ready");
3794 setTimeout(() => fs.writeFileSync(process.env.SURVIVED, "survived"), 1000);
3795 "#,
3796 ])
3797 .env("READY", ready)
3798 .env("SURVIVED", survived)
3799 .stderr(Stdio::inherit());
3800 command
3801 }
3802
3803 #[cfg(any(unix, windows))]
3804 async fn wait_for_test_child(ready: &Path) {
3805 let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
3806 while !ready.exists() {
3807 assert!(
3808 tokio::time::Instant::now() < deadline,
3809 "child did not report readiness"
3810 );
3811 tokio::time::sleep(Duration::from_millis(10)).await;
3812 }
3813 }
3814
3815 #[cfg(any(unix, windows))]
3816 async fn assert_test_child_killed(survived: &Path) {
3817 tokio::time::sleep(Duration::from_millis(1500)).await;
3818
3819 assert!(
3820 !survived.exists(),
3821 "child survived after its owner was dropped"
3822 );
3823 }
3824
3825 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3826 Client {
3827 inner: Arc::new(ClientInner {
3828 child: parking_lot::Mutex::new(None),
3829 process_tree: parking_lot::Mutex::new(None),
3830 #[cfg(feature = "in-process")]
3831 ffi_host: parking_lot::Mutex::new(None),
3832 rpc: {
3833 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3834 let (notif_tx, _notif_rx) = broadcast::channel(16);
3835 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3836 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3837 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3838 },
3839 cwd: PathBuf::from("."),
3840 request_rx: parking_lot::Mutex::new(None),
3841 notification_tx: broadcast::channel(16).0,
3842 router: router::SessionRouter::new(),
3843 github_token_registry: Arc::new(github_token::GitHubTokenRegistry::new()),
3844 negotiated_protocol_version: OnceLock::new(),
3845 state: parking_lot::Mutex::new(ConnectionState::Connected),
3846 lifecycle_tx: broadcast::channel(16).0,
3847 on_list_models: Some(handler),
3848 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3849 session_fs_configured: false,
3850 session_fs_sqlite_declared: false,
3851 llm_inference: OnceLock::new(),
3852 extension_launch_provider: Arc::new(
3853 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(None),
3854 ),
3855 installation_confirmation: Arc::new(
3856 installation_confirmation::InstallationConfirmationDispatcher::new(),
3857 ),
3858 on_github_telemetry: None,
3859 on_get_trace_context: None,
3860 effective_connection_token: None,
3861 mode: ClientMode::default(),
3862 client_info: None,
3863 startup_timings: OnceLock::new(),
3864 }),
3865 }
3866 }
3867
3868 async fn wait_for_pending_session_registration(client: &Client) {
3869 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3870 while client.inner.router.session_ids().is_empty() {
3871 assert!(
3872 tokio::time::Instant::now() < deadline,
3873 "session was not registered"
3874 );
3875 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3876 }
3877 }
3878}