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