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;
36pub mod provider_token;
38mod provider_token_dispatch;
39pub(crate) mod resolve;
41mod router;
42pub mod session;
44pub mod session_fs;
46mod session_fs_dispatch;
47pub mod startup_timings;
49pub mod subscription;
51pub mod tool;
53pub mod trace_context;
55pub mod transforms;
57pub mod types;
59mod wire;
60
61pub mod session_events;
63
64pub mod rpc;
67
68pub(crate) mod generated;
73
74pub mod mode;
77
78use std::ffi::OsString;
79use std::path::{Path, PathBuf};
80use std::process::Stdio;
81use std::sync::{Arc, OnceLock};
82use std::time::{Duration, Instant};
83
84use async_trait::async_trait;
85pub use github_token::{
86 GitHubToken, GitHubTokenProvider, GitHubTokenProviderArgs, GitHubTokenProviderResult,
87 GitHubTokenRequestReason,
88};
89pub use indexmap::IndexMap;
93pub(crate) use jsonrpc::{
96 JsonRpcClient, JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, error_codes,
97};
98pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet};
99pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs};
100
101#[cfg(feature = "test-support")]
103pub mod test_support {
104 pub use crate::jsonrpc::{
105 JsonRpcClient, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
106 error_codes,
107 };
108}
109use serde::{Deserialize, Serialize};
110use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader};
111use tokio::net::TcpStream;
112use tokio::process::{Child, Command};
113use tokio::sync::{broadcast, mpsc, oneshot};
114use tracing::{Instrument, debug, error, info, warn};
115pub use types::*;
116
117mod sdk_protocol_version;
118pub use sdk_protocol_version::{SDK_PROTOCOL_VERSION, get_sdk_protocol_version};
119pub use startup_timings::StartupTimings;
120pub use subscription::{EventSubscription, LifecycleSubscription};
121
122const MIN_PROTOCOL_VERSION: u32 = 3;
124const RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
125
126fn record_optional_millis(span: &tracing::Span, field: &'static str, value: Option<u64>) {
127 match value {
128 Some(value) => {
129 span.record(field, value);
130 }
131 None => {
132 span.record(field, "None");
133 }
134 }
135}
136
137#[derive(Debug, Default)]
139#[non_exhaustive]
140pub enum Transport {
141 #[default]
144 Default,
145 Stdio,
147 InProcess,
160 Tcp {
162 port: u16,
164 connection_token: Option<String>,
168 },
169 External {
171 host: String,
173 port: u16,
175 connection_token: Option<String>,
178 },
179}
180
181#[derive(Debug, Clone, Default)]
183pub enum CliProgram {
184 #[default]
189 Resolve,
190 Path(PathBuf),
192}
193
194impl From<PathBuf> for CliProgram {
195 fn from(path: PathBuf) -> Self {
196 Self::Path(path)
197 }
198}
199
200pub const HAS_BUNDLED_CLI: bool = cfg!(has_bundled_cli);
207
208pub fn install_bundled_cli() -> Option<PathBuf> {
229 #[cfg(feature = "bundled-cli")]
230 {
231 embeddedcli::path()
232 }
233 #[cfg(not(feature = "bundled-cli"))]
234 {
235 None
236 }
237}
238
239pub fn install_bundled_runtime() -> Option<PathBuf> {
250 #[cfg(feature = "bundled-cli")]
251 {
252 embeddedcli::runtime_path()
253 }
254 #[cfg(not(feature = "bundled-cli"))]
255 {
256 None
257 }
258}
259
260#[non_exhaustive]
272pub struct ClientOptions {
273 pub program: CliProgram,
275 pub prefix_args: Vec<OsString>,
277 pub working_directory: PathBuf,
281 pub env: Vec<(OsString, OsString)>,
283 pub env_remove: Vec<OsString>,
285 pub extra_args: Vec<String>,
287 pub builtin_plugin_directories: Vec<PathBuf>,
292 pub transport: Transport,
294 pub github_token: Option<String>,
299 pub use_logged_in_user: Option<bool>,
303 pub log_level: Option<LogLevel>,
307 pub session_idle_timeout_seconds: Option<u64>,
313 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
321 pub session_fs: Option<SessionFsConfig>,
329 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
338 pub extension_launch_provider:
345 Option<Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>>,
346 #[doc(hidden)]
354 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
355 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
365 pub telemetry: Option<TelemetryConfig>,
369 pub base_directory: Option<PathBuf>,
374 pub enable_remote_sessions: bool,
380 pub bundled_cli_extract_dir: Option<PathBuf>,
399 pub mode: ClientMode,
403}
404
405impl std::fmt::Debug for ClientOptions {
406 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
407 f.debug_struct("ClientOptions")
408 .field("program", &self.program)
409 .field("prefix_args", &self.prefix_args)
410 .field("working_directory", &self.working_directory)
411 .field("env", &self.env)
412 .field("env_remove", &self.env_remove)
413 .field("extra_args", &self.extra_args)
414 .field(
415 "builtin_plugin_directories",
416 &self.builtin_plugin_directories,
417 )
418 .field("transport", &self.transport)
419 .field(
420 "github_token",
421 &self.github_token.as_ref().map(|_| "<redacted>"),
422 )
423 .field("use_logged_in_user", &self.use_logged_in_user)
424 .field("log_level", &self.log_level)
425 .field(
426 "session_idle_timeout_seconds",
427 &self.session_idle_timeout_seconds,
428 )
429 .field(
430 "on_list_models",
431 &self.on_list_models.as_ref().map(|_| "<set>"),
432 )
433 .field("session_fs", &self.session_fs)
434 .field(
435 "request_handler",
436 &self.request_handler.as_ref().map(|_| "<set>"),
437 )
438 .field(
439 "extension_launch_provider",
440 &self.extension_launch_provider.as_ref().map(|_| "<set>"),
441 )
442 .field(
443 "on_github_telemetry",
444 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
445 )
446 .field(
447 "on_get_trace_context",
448 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
449 )
450 .field("telemetry", &self.telemetry)
451 .field("base_directory", &self.base_directory)
452 .field("enable_remote_sessions", &self.enable_remote_sessions)
453 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
454 .finish()
455 }
456}
457
458#[async_trait]
467pub trait ListModelsHandler: Send + Sync + 'static {
468 async fn list_models(&self) -> Result<Vec<Model>>;
470}
471
472#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
474#[serde(rename_all = "lowercase")]
475pub enum LogLevel {
476 None,
478 Error,
480 Warning,
482 Info,
484 Debug,
486 All,
488}
489
490impl LogLevel {
491 pub fn as_str(self) -> &'static str {
493 match self {
494 Self::None => "none",
495 Self::Error => "error",
496 Self::Warning => "warning",
497 Self::Info => "info",
498 Self::Debug => "debug",
499 Self::All => "all",
500 }
501 }
502}
503
504impl std::fmt::Display for LogLevel {
505 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
506 f.write_str(self.as_str())
507 }
508}
509
510#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
515#[serde(rename_all = "kebab-case")]
516#[non_exhaustive]
517pub enum OtelExporterType {
518 OtlpHttp,
521 File,
524}
525
526impl OtelExporterType {
527 pub fn as_str(self) -> &'static str {
529 match self {
530 Self::OtlpHttp => "otlp-http",
531 Self::File => "file",
532 }
533 }
534}
535
536#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
542#[non_exhaustive]
543pub enum OtlpHttpProtocol {
544 #[serde(rename = "http/json")]
546 HttpJson,
547 #[serde(rename = "http/protobuf")]
549 HttpProtobuf,
550}
551
552impl OtlpHttpProtocol {
553 pub fn as_str(self) -> &'static str {
555 match self {
556 Self::HttpJson => "http/json",
557 Self::HttpProtobuf => "http/protobuf",
558 }
559 }
560}
561
562#[derive(Debug, Clone, Default)]
597#[non_exhaustive]
598pub struct TelemetryConfig {
599 pub otlp_endpoint: Option<String>,
601 pub otlp_protocol: Option<OtlpHttpProtocol>,
603 pub file_path: Option<PathBuf>,
605 pub exporter_type: Option<OtelExporterType>,
608 pub source_name: Option<String>,
612 pub capture_content: Option<bool>,
616}
617
618impl TelemetryConfig {
619 pub fn new() -> Self {
622 Self::default()
623 }
624
625 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
627 self.otlp_endpoint = Some(endpoint.into());
628 self
629 }
630
631 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
633 self.otlp_protocol = Some(protocol);
634 self
635 }
636
637 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
639 self.file_path = Some(path.into());
640 self
641 }
642
643 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
645 self.exporter_type = Some(exporter_type);
646 self
647 }
648
649 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
653 self.source_name = Some(source_name.into());
654 self
655 }
656
657 pub fn with_capture_content(mut self, capture: bool) -> Self {
661 self.capture_content = Some(capture);
662 self
663 }
664
665 pub fn is_empty(&self) -> bool {
668 self.otlp_endpoint.is_none()
669 && self.otlp_protocol.is_none()
670 && self.file_path.is_none()
671 && self.exporter_type.is_none()
672 && self.source_name.is_none()
673 && self.capture_content.is_none()
674 }
675}
676
677impl Default for ClientOptions {
678 fn default() -> Self {
679 Self {
680 program: CliProgram::Resolve,
681 prefix_args: Vec::new(),
682 working_directory: PathBuf::new(),
683 env: Vec::new(),
684 env_remove: Vec::new(),
685 extra_args: Vec::new(),
686 builtin_plugin_directories: Vec::new(),
687 transport: Transport::default(),
688 github_token: None,
689 use_logged_in_user: None,
690 log_level: None,
691 session_idle_timeout_seconds: None,
692 on_list_models: None,
693 session_fs: None,
694 request_handler: None,
695 extension_launch_provider: None,
696 on_github_telemetry: None,
697 on_get_trace_context: None,
698 telemetry: None,
699 base_directory: None,
700 enable_remote_sessions: false,
701 bundled_cli_extract_dir: None,
702 mode: ClientMode::default(),
703 }
704 }
705}
706
707impl ClientOptions {
708 pub fn new() -> Self {
724 Self::default()
725 }
726
727 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
729 self.program = program.into();
730 self
731 }
732
733 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
735 where
736 I: IntoIterator<Item = S>,
737 S: Into<OsString>,
738 {
739 self.prefix_args = args.into_iter().map(Into::into).collect();
740 self
741 }
742
743 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
745 self.working_directory = cwd.into();
746 self
747 }
748
749 pub fn with_env<I, K, V>(mut self, env: I) -> Self
751 where
752 I: IntoIterator<Item = (K, V)>,
753 K: Into<OsString>,
754 V: Into<OsString>,
755 {
756 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
757 self
758 }
759
760 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
762 where
763 I: IntoIterator<Item = S>,
764 S: Into<OsString>,
765 {
766 self.env_remove = names.into_iter().map(Into::into).collect();
767 self
768 }
769
770 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
772 where
773 I: IntoIterator<Item = S>,
774 S: Into<String>,
775 {
776 self.extra_args = args.into_iter().map(Into::into).collect();
777 self
778 }
779
780 pub fn with_builtin_plugin_directories<I, P>(mut self, paths: I) -> Self
785 where
786 I: IntoIterator<Item = P>,
787 P: Into<PathBuf>,
788 {
789 self.builtin_plugin_directories = paths.into_iter().map(Into::into).collect();
790 self
791 }
792
793 pub fn with_transport(mut self, transport: Transport) -> Self {
795 self.transport = transport;
796 self
797 }
798
799 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
802 self.github_token = Some(token.into());
803 self
804 }
805
806 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
809 self.use_logged_in_user = Some(use_logged_in);
810 self
811 }
812
813 pub fn with_log_level(mut self, level: LogLevel) -> Self {
815 self.log_level = Some(level);
816 self
817 }
818
819 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
822 self.session_idle_timeout_seconds = Some(seconds);
823 self
824 }
825
826 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
829 where
830 H: ListModelsHandler + 'static,
831 {
832 self.on_list_models = Some(Arc::new(handler));
833 self
834 }
835
836 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
838 self.session_fs = Some(config);
839 self
840 }
841
842 pub fn with_request_handler<H>(mut self, handler: H) -> Self
847 where
848 H: crate::copilot_request_handler::CopilotRequestHandler,
849 {
850 self.request_handler = Some(Arc::new(handler));
851 self
852 }
853
854 pub fn with_extension_launch_provider<P>(mut self, provider: P) -> Self
859 where
860 P: crate::extension_launch_provider::ExtensionLaunchProvider,
861 {
862 self.extension_launch_provider = Some(Arc::new(provider));
863 self
864 }
865
866 #[doc(hidden)]
872 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
873 where
874 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
875 {
876 self.on_github_telemetry = Some(Arc::new(callback));
877 self
878 }
879
880 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
884 where
885 P: TraceContextProvider + 'static,
886 {
887 self.on_get_trace_context = Some(Arc::new(provider));
888 self
889 }
890
891 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
893 self.telemetry = Some(config);
894 self
895 }
896
897 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
900 self.base_directory = Some(dir.into());
901 self
902 }
903
904 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
907 self.enable_remote_sessions = enabled;
908 self
909 }
910
911 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
921 self.bundled_cli_extract_dir = Some(dir.into());
922 self
923 }
924
925 pub fn with_mode(mut self, mode: ClientMode) -> Self {
930 self.mode = mode;
931 self
932 }
933}
934
935fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
937 if cfg.initial_cwd.trim().is_empty() {
938 return Err(Error::with_message(
939 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
940 "invalid SessionFsConfig: initial_cwd must not be empty",
941 ));
942 }
943 if cfg.session_state_path.trim().is_empty() {
944 return Err(Error::with_message(
945 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
946 "invalid SessionFsConfig: session_state_path must not be empty",
947 ));
948 }
949 Ok(())
950}
951
952fn generate_connection_token() -> String {
959 let mut bytes = [0u8; 16];
960 getrandom::getrandom(&mut bytes)
961 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
962 let mut hex = String::with_capacity(32);
963 for byte in bytes {
964 use std::fmt::Write;
965 let _ = write!(hex, "{byte:02x}");
966 }
967 hex
968}
969
970const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
975
976fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
978 let configured = options
979 .env
980 .iter()
981 .find(|(key, _)| {
982 key.to_string_lossy()
983 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
984 })
985 .map(|(_, value)| value.to_string_lossy().into_owned());
986 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
987 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
988}
989
990fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
991 match value {
992 None => Ok(Transport::Stdio),
993 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
994 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
995 Some(v) => Err(Error::with_message(
996 ErrorKind::InvalidConfig,
997 format!(
998 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
999 Expected 'inprocess', 'stdio', or unset."
1000 ),
1001 )),
1002 }
1003}
1004
1005#[cfg(any(feature = "bundled-in-process", test))]
1006fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
1007 if !matches!(&options.program, CliProgram::Resolve) {
1008 return Err(Error::with_message(
1009 ErrorKind::InvalidConfig,
1010 "ClientOptions::program is not supported with Transport::InProcess; \
1011 set COPILOT_CLI_PATH only when using an externally provisioned runtime package",
1012 ));
1013 }
1014 if !options.extra_args.is_empty() {
1015 return Err(Error::with_message(
1016 ErrorKind::InvalidConfig,
1017 "ClientOptions::extra_args is not supported with Transport::InProcess; \
1018 use typed client options instead",
1019 ));
1020 }
1021
1022 let unsupported = if !options.working_directory.as_os_str().is_empty() {
1023 Some("working_directory")
1024 } else if !options.env.is_empty() {
1025 Some("env")
1026 } else if !options.env_remove.is_empty() {
1027 Some("env_remove")
1028 } else if options.telemetry.is_some() {
1029 Some("telemetry")
1030 } else if !options.prefix_args.is_empty() {
1031 Some("prefix_args")
1032 } else {
1033 None
1034 };
1035
1036 if let Some(option) = unsupported {
1037 return Err(Error::with_message(
1038 ErrorKind::InvalidConfig,
1039 format!(
1040 "ClientOptions::{option} is not supported with Transport::InProcess; \
1041 configure process-global settings on the host process instead"
1042 ),
1043 ));
1044 }
1045
1046 Ok(())
1047}
1048
1049#[derive(Clone)]
1054pub struct Client {
1055 inner: Arc<ClientInner>,
1056}
1057
1058impl std::fmt::Debug for Client {
1059 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1060 f.debug_struct("Client")
1061 .field("working_directory", &self.inner.cwd)
1062 .field("pid", &self.pid())
1063 .finish()
1064 }
1065}
1066
1067struct ClientInner {
1068 child: parking_lot::Mutex<Option<Child>>,
1069 #[cfg(feature = "bundled-in-process")]
1070 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
1073 rpc: JsonRpcClient,
1074 cwd: PathBuf,
1075 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
1076 notification_tx: broadcast::Sender<JsonRpcNotification>,
1077 router: router::SessionRouter,
1078 github_token_registry: Arc<github_token::GitHubTokenRegistry>,
1079 negotiated_protocol_version: OnceLock<u32>,
1080 state: parking_lot::Mutex<ConnectionState>,
1081 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
1082 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1083 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
1084 session_fs_configured: bool,
1085 session_fs_sqlite_declared: bool,
1086 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
1089 extension_launch_provider: Arc<extension_launch_provider::ExtensionLaunchProviderDispatcher>,
1090 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1095 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1096 effective_connection_token: Option<String>,
1101 pub(crate) mode: ClientMode,
1104 startup_timings: OnceLock<StartupTimings>,
1108}
1109
1110impl Client {
1111 pub async fn start(options: ClientOptions) -> Result<Self> {
1124 let start_time = Instant::now();
1125 let mut timings = StartupTimings::default();
1126 let mut options = options;
1127 if matches!(options.transport, Transport::Default) {
1128 options.transport = resolve_default_transport(&options)?;
1129 }
1130 if matches!(options.transport, Transport::InProcess) {
1131 #[cfg(not(feature = "bundled-in-process"))]
1132 {
1133 return Err(Error::with_message(
1134 ErrorKind::InvalidConfig,
1135 "Transport::InProcess requires the `bundled-in-process` Cargo feature",
1136 ));
1137 }
1138 #[cfg(feature = "bundled-in-process")]
1139 validate_inprocess_options(&options)?;
1140 }
1141 if options.mode == ClientMode::Empty
1142 && options.base_directory.is_none()
1143 && options.session_fs.is_none()
1144 {
1145 return Err(Error::with_message(
1146 ErrorKind::InvalidConfig,
1147 "ClientMode::Empty requires either `base_directory` or \
1148 `session_fs` to be set (no implicit ~/.copilot fallback).",
1149 ));
1150 }
1151 if let Some(cfg) = &options.session_fs {
1152 validate_session_fs_config(cfg)?;
1153 }
1154 let builtin_plugin_directories = options
1155 .builtin_plugin_directories
1156 .iter()
1157 .map(|path| {
1158 if !path.is_absolute() {
1159 return Err(Error::with_message(
1160 ErrorKind::InvalidConfig,
1161 format!(
1162 "builtin_plugin_directories must contain only absolute paths: {}",
1163 path.display()
1164 ),
1165 ));
1166 }
1167 path.to_str().map(str::to_owned).ok_or_else(|| {
1168 Error::with_message(
1169 ErrorKind::InvalidConfig,
1170 format!(
1171 "builtin_plugin_directories must contain valid UTF-8 paths: {}",
1172 path.display()
1173 ),
1174 )
1175 })
1176 })
1177 .collect::<Result<Vec<_>>>()?;
1178 if matches!(options.transport, Transport::External { .. }) {
1181 if options.github_token.is_some() {
1182 return Err(Error::with_message(
1183 ErrorKind::InvalidConfig,
1184 "invalid client configuration: github_token cannot be used with \
1185 Transport::External (external server manages its own auth)",
1186 ));
1187 }
1188 if options.use_logged_in_user == Some(true) {
1189 return Err(Error::with_message(
1190 ErrorKind::InvalidConfig,
1191 "invalid client configuration: use_logged_in_user cannot be used with \
1192 Transport::External (external server manages its own auth)",
1193 ));
1194 }
1195 }
1196 match &options.transport {
1200 Transport::Tcp {
1201 connection_token: Some(t),
1202 ..
1203 }
1204 | Transport::External {
1205 connection_token: Some(t),
1206 ..
1207 } if t.is_empty() => {
1208 return Err(Error::with_message(
1209 ErrorKind::InvalidConfig,
1210 "invalid client configuration: connection_token must be a non-empty string",
1211 ));
1212 }
1213 _ => {}
1214 }
1215 let effective_connection_token: Option<String> = match &mut options.transport {
1220 Transport::Default => unreachable!("default transport resolved above"),
1221 Transport::Stdio | Transport::InProcess => None,
1222 Transport::Tcp {
1223 connection_token, ..
1224 } => Some(
1225 connection_token
1226 .get_or_insert_with(generate_connection_token)
1227 .clone(),
1228 ),
1229 Transport::External {
1230 connection_token, ..
1231 } => connection_token.clone(),
1232 };
1233 let session_fs_config = options.session_fs.clone();
1234 let request_handler = options.request_handler.clone();
1235 let extension_launch_provider = options.extension_launch_provider.clone();
1236 let session_fs_sqlite_declared = session_fs_config
1237 .as_ref()
1238 .and_then(|c| c.capabilities.as_ref())
1239 .is_some_and(|caps| caps.sqlite);
1240 let program = match &options.program {
1241 CliProgram::Path(path) => {
1242 info!(path = %path.display(), "using explicit copilot CLI path");
1243 path.clone()
1244 }
1245 CliProgram::Resolve => {
1246 let resolve_start = Instant::now();
1247 let resolved = resolve::copilot_binary_with_extract_dir(
1248 options.bundled_cli_extract_dir.as_deref(),
1249 true,
1250 )?;
1251 let resolve_elapsed = resolve_start.elapsed();
1252 timings.program_resolve_ms = Some(StartupTimings::millis(resolve_elapsed));
1253 debug!(
1254 elapsed_ms = resolve_elapsed.as_millis(),
1255 "Client::start CLI program resolution complete"
1256 );
1257 info!(path = %resolved.display(), "resolved copilot runtime");
1258 #[cfg(windows)]
1259 {
1260 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1261 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1262 }) {
1263 warn!(
1264 path = %resolved.display(),
1265 ext = %ext,
1266 "resolved copilot CLI is a .cmd/.bat wrapper; \
1267 this may cause console window flashes on Windows"
1268 );
1269 }
1270 }
1271 resolved
1272 }
1273 };
1274 let working_directory = {
1275 let cwd = options.working_directory.clone();
1276 if cwd.as_os_str().is_empty() {
1277 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1278 } else {
1279 cwd
1280 }
1281 };
1282
1283 let transport_setup_start = Instant::now();
1284 let client = match options.transport {
1285 Transport::Default => unreachable!("default transport resolved above"),
1286 Transport::External {
1287 ref host,
1288 port,
1289 connection_token: _,
1290 } => {
1291 info!(host = %host, port = %port, "connecting to external CLI server");
1292 let connect_start = Instant::now();
1293 let stream = TcpStream::connect((host.as_str(), port)).await?;
1294 debug!(
1295 elapsed_ms = connect_start.elapsed().as_millis(),
1296 host = %host,
1297 port,
1298 "Client::start TCP connect complete"
1299 );
1300 let (reader, writer) = tokio::io::split(stream);
1301 Self::from_transport(
1302 reader,
1303 writer,
1304 None,
1305 working_directory,
1306 options.on_list_models,
1307 extension_launch_provider.clone(),
1308 session_fs_config.is_some(),
1309 session_fs_sqlite_declared,
1310 options.on_get_trace_context,
1311 options.on_github_telemetry,
1312 effective_connection_token.clone(),
1313 options.mode,
1314 )?
1315 }
1316 Transport::Tcp {
1317 port,
1318 connection_token: _,
1319 } => {
1320 let (mut child, actual_port, spawn_elapsed, port_wait_elapsed) =
1321 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1322 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1323 timings.port_wait_ms = Some(StartupTimings::millis(port_wait_elapsed));
1324 let connect_start = Instant::now();
1325 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1326 debug!(
1327 elapsed_ms = connect_start.elapsed().as_millis(),
1328 port = actual_port,
1329 "Client::start TCP connect complete"
1330 );
1331 let (reader, writer) = tokio::io::split(stream);
1332 Self::drain_stderr(&mut child);
1333 Self::from_transport(
1334 reader,
1335 writer,
1336 Some(child),
1337 working_directory,
1338 options.on_list_models,
1339 extension_launch_provider.clone(),
1340 session_fs_config.is_some(),
1341 session_fs_sqlite_declared,
1342 options.on_get_trace_context,
1343 options.on_github_telemetry,
1344 effective_connection_token.clone(),
1345 options.mode,
1346 )?
1347 }
1348 Transport::Stdio => {
1349 let (mut child, spawn_elapsed) =
1350 Self::spawn_stdio(&program, &options, &working_directory)?;
1351 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1352 let stdin = child.stdin.take().expect("stdin is piped");
1353 let stdout = child.stdout.take().expect("stdout is piped");
1354 Self::drain_stderr(&mut child);
1355 Self::from_transport(
1356 stdout,
1357 stdin,
1358 Some(child),
1359 working_directory,
1360 options.on_list_models,
1361 extension_launch_provider.clone(),
1362 session_fs_config.is_some(),
1363 session_fs_sqlite_declared,
1364 options.on_get_trace_context,
1365 options.on_github_telemetry,
1366 effective_connection_token.clone(),
1367 options.mode,
1368 )?
1369 }
1370 Transport::InProcess => {
1371 #[cfg(feature = "bundled-in-process")]
1372 {
1373 info!(runtime_path = %program.display(), "hosting copilot runtime in-process (FFI)");
1374 let mut environment = Vec::new();
1375 if let Some(base_directory) = &options.base_directory {
1376 let value = base_directory.to_str().ok_or_else(|| {
1377 Error::with_message(
1378 ErrorKind::InvalidConfig,
1379 "base_directory must be valid UTF-8 for Transport::InProcess",
1380 )
1381 })?;
1382 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1383 }
1384 if options.mode == ClientMode::Empty {
1385 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1386 }
1387 if let Some(github_token) = &options.github_token {
1388 environment
1389 .push(("COPILOT_SDK_AUTH_TOKEN".to_string(), github_token.clone()));
1390 }
1391 let mut args = Vec::new();
1392 args.extend(
1393 Self::log_level_args(&options)
1394 .into_iter()
1395 .map(str::to_string),
1396 );
1397 args.extend(Self::session_idle_timeout_args(&options));
1398 args.extend(Self::remote_args(&options));
1399 if options.github_token.is_some() {
1400 args.extend([
1401 "--auth-token-env".to_string(),
1402 "COPILOT_SDK_AUTH_TOKEN".to_string(),
1403 ]);
1404 }
1405 let use_logged_in_user = options
1406 .use_logged_in_user
1407 .unwrap_or(options.github_token.is_none());
1408 if !use_logged_in_user {
1409 args.push("--no-auto-login".to_string());
1410 }
1411 let explicit_cli = std::env::var_os("COPILOT_CLI_PATH")
1412 .map(PathBuf::from)
1413 .filter(|path| path.is_file());
1414 let host = crate::ffi::FfiHost::create(
1415 &program,
1416 explicit_cli.as_deref(),
1417 environment,
1418 args,
1419 )?;
1420 let (reader, writer, shared) = host.start().await?;
1421 let client = Self::from_transport(
1422 reader,
1423 writer,
1424 None,
1425 working_directory,
1426 options.on_list_models,
1427 extension_launch_provider.clone(),
1428 session_fs_config.is_some(),
1429 session_fs_sqlite_declared,
1430 options.on_get_trace_context,
1431 options.on_github_telemetry,
1432 effective_connection_token.clone(),
1433 options.mode,
1434 )?;
1435 *client.inner.ffi_host.lock() = Some(shared);
1436 client
1437 }
1438 #[cfg(not(feature = "bundled-in-process"))]
1439 unreachable!("in-process feature validation returned above")
1440 }
1441 };
1442 timings.transport_setup_ms = StartupTimings::millis(transport_setup_start.elapsed());
1443 debug!(
1444 elapsed_ms = start_time.elapsed().as_millis(),
1445 "Client::start transport setup complete"
1446 );
1447 let handshake_start = Instant::now();
1448 client.verify_protocol_version().await?;
1449 timings.handshake_ms = StartupTimings::millis(handshake_start.elapsed());
1450 debug!(
1451 elapsed_ms = start_time.elapsed().as_millis(),
1452 "Client::start protocol verification complete"
1453 );
1454 let request_dispatcher = request_handler.map(|handler| {
1455 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1456 handler,
1457 ));
1458 dispatcher.set_client(Arc::downgrade(&client.inner));
1459 let _ = client.inner.llm_inference.set(dispatcher.clone());
1460 dispatcher
1461 });
1462 if client.inner.extension_launch_provider.is_configured() {
1463 client.inner.router.ensure_started(
1464 &client.inner.notification_tx,
1465 &client.inner.request_rx,
1466 client.inner.extension_launch_provider.clone(),
1467 request_dispatcher.clone(),
1468 client.inner.on_github_telemetry.clone(),
1469 client.inner.github_token_registry.clone(),
1470 );
1471 client.rpc().register_extension_launch_provider().await?;
1472 }
1473 if !builtin_plugin_directories.is_empty() {
1474 client
1475 .call(
1476 "plugins.builtin.set",
1477 Some(serde_json::json!({ "paths": builtin_plugin_directories })),
1478 )
1479 .await?;
1480 }
1481 if let Some(cfg) = session_fs_config {
1482 let session_fs_start = Instant::now();
1483 let capabilities = cfg.capabilities.as_ref().map(|c| {
1484 crate::generated::api_types::SessionFsSetProviderCapabilities {
1485 sqlite: Some(c.sqlite),
1486 }
1487 });
1488 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1489 capabilities,
1490 conventions: cfg.conventions.into_wire(),
1491 initial_cwd: cfg.initial_cwd,
1492 session_state_path: cfg.session_state_path,
1493 };
1494 client.rpc().session_fs().set_provider(request).await?;
1495 let session_fs_elapsed = session_fs_start.elapsed();
1496 timings.session_fs_ms = Some(StartupTimings::millis(session_fs_elapsed));
1497 debug!(
1498 elapsed_ms = session_fs_elapsed.as_millis(),
1499 "Client::start session filesystem setup complete"
1500 );
1501 }
1502 if let Some(dispatcher) = request_dispatcher {
1503 let llm_inference_start = Instant::now();
1504 client.inner.router.ensure_started(
1507 &client.inner.notification_tx,
1508 &client.inner.request_rx,
1509 client.inner.extension_launch_provider.clone(),
1510 Some(dispatcher.clone()),
1511 client.inner.on_github_telemetry.clone(),
1512 client.inner.github_token_registry.clone(),
1513 );
1514 client.rpc().llm_inference().set_provider().await?;
1515 let llm_inference_elapsed = llm_inference_start.elapsed();
1516 timings.llm_handler_ms = Some(StartupTimings::millis(llm_inference_elapsed));
1517 debug!(
1518 elapsed_ms = llm_inference_elapsed.as_millis(),
1519 "Client::start Copilot request handler registration complete"
1520 );
1521 }
1522 timings.total_ms = StartupTimings::millis(start_time.elapsed());
1523 let timings_span = tracing::debug_span!(
1526 "Client::start timings",
1527 program_resolve_ms = tracing::field::Empty,
1528 process_spawn_ms = tracing::field::Empty,
1529 port_wait_ms = tracing::field::Empty,
1530 transport_setup_ms = timings.transport_setup_ms,
1531 handshake_ms = timings.handshake_ms,
1532 session_fs_ms = tracing::field::Empty,
1533 llm_handler_ms = tracing::field::Empty,
1534 total_ms = timings.total_ms,
1535 );
1536 record_optional_millis(
1537 &timings_span,
1538 "program_resolve_ms",
1539 timings.program_resolve_ms,
1540 );
1541 record_optional_millis(&timings_span, "process_spawn_ms", timings.process_spawn_ms);
1542 record_optional_millis(&timings_span, "port_wait_ms", timings.port_wait_ms);
1543 record_optional_millis(&timings_span, "session_fs_ms", timings.session_fs_ms);
1544 record_optional_millis(&timings_span, "llm_handler_ms", timings.llm_handler_ms);
1545 timings_span.in_scope(|| debug!("Client::start timings"));
1546 let _ = client.inner.startup_timings.set(timings);
1547 debug!(
1548 elapsed_ms = start_time.elapsed().as_millis(),
1549 "Client::start complete"
1550 );
1551 Ok(client)
1552 }
1553
1554 pub fn from_streams(
1558 reader: impl AsyncRead + Unpin + Send + 'static,
1559 writer: impl AsyncWrite + Unpin + Send + 'static,
1560 cwd: PathBuf,
1561 ) -> Result<Self> {
1562 Self::from_transport(
1563 reader,
1564 writer,
1565 None,
1566 cwd,
1567 None,
1568 None,
1569 false,
1570 false,
1571 None,
1572 None,
1573 None,
1574 ClientMode::default(),
1575 )
1576 }
1577
1578 #[doc(hidden)]
1581 #[cfg(any(test, feature = "test-support"))]
1582 pub fn from_streams_with_extension_launch_provider(
1583 reader: impl AsyncRead + Unpin + Send + 'static,
1584 writer: impl AsyncWrite + Unpin + Send + 'static,
1585 cwd: PathBuf,
1586 provider: Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1587 ) -> Result<Self> {
1588 Self::from_transport(
1589 reader,
1590 writer,
1591 None,
1592 cwd,
1593 None,
1594 Some(provider),
1595 false,
1596 false,
1597 None,
1598 None,
1599 None,
1600 ClientMode::default(),
1601 )
1602 }
1603
1604 #[cfg(any(test, feature = "test-support"))]
1612 pub fn from_streams_with_trace_provider(
1613 reader: impl AsyncRead + Unpin + Send + 'static,
1614 writer: impl AsyncWrite + Unpin + Send + 'static,
1615 cwd: PathBuf,
1616 provider: Arc<dyn TraceContextProvider>,
1617 ) -> Result<Self> {
1618 Self::from_transport(
1619 reader,
1620 writer,
1621 None,
1622 cwd,
1623 None,
1624 None,
1625 false,
1626 false,
1627 Some(provider),
1628 None,
1629 None,
1630 ClientMode::default(),
1631 )
1632 }
1633
1634 #[cfg(any(test, feature = "test-support"))]
1638 pub fn from_streams_with_connection_token(
1639 reader: impl AsyncRead + Unpin + Send + 'static,
1640 writer: impl AsyncWrite + Unpin + Send + 'static,
1641 cwd: PathBuf,
1642 token: Option<String>,
1643 ) -> Result<Self> {
1644 Self::from_transport(
1645 reader,
1646 writer,
1647 None,
1648 cwd,
1649 None,
1650 None,
1651 false,
1652 false,
1653 None,
1654 None,
1655 token,
1656 ClientMode::default(),
1657 )
1658 }
1659
1660 #[doc(hidden)]
1663 #[cfg(any(test, feature = "test-support"))]
1664 pub fn from_streams_with_github_telemetry(
1665 reader: impl AsyncRead + Unpin + Send + 'static,
1666 writer: impl AsyncWrite + Unpin + Send + 'static,
1667 cwd: PathBuf,
1668 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1669 ) -> Result<Self> {
1670 Self::from_transport(
1671 reader,
1672 writer,
1673 None,
1674 cwd,
1675 None,
1676 None,
1677 false,
1678 false,
1679 None,
1680 Some(on_github_telemetry),
1681 None,
1682 ClientMode::default(),
1683 )
1684 }
1685
1686 #[cfg(any(test, feature = "test-support"))]
1692 pub fn generate_connection_token_for_test() -> String {
1693 generate_connection_token()
1694 }
1695
1696 #[allow(clippy::too_many_arguments)]
1697 fn from_transport(
1698 reader: impl AsyncRead + Unpin + Send + 'static,
1699 writer: impl AsyncWrite + Unpin + Send + 'static,
1700 child: Option<Child>,
1701 cwd: PathBuf,
1702 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1703 extension_launch_provider: Option<
1704 Arc<dyn crate::extension_launch_provider::ExtensionLaunchProvider>,
1705 >,
1706 session_fs_configured: bool,
1707 session_fs_sqlite_declared: bool,
1708 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1709 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1710 effective_connection_token: Option<String>,
1711 mode: ClientMode,
1712 ) -> Result<Self> {
1713 let setup_start = Instant::now();
1714 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1715 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1716 let rpc = JsonRpcClient::new(
1717 writer,
1718 reader,
1719 notification_broadcast_tx.clone(),
1720 request_tx,
1721 );
1722
1723 let pid = child.as_ref().and_then(|c| c.id());
1724 info!(pid = ?pid, "copilot CLI client ready");
1725
1726 let github_token_registry = Arc::new(github_token::GitHubTokenRegistry::new());
1727 let extension_launch_provider = Arc::new(
1728 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(
1729 extension_launch_provider,
1730 ),
1731 );
1732 let client = Self {
1733 inner: Arc::new(ClientInner {
1734 child: parking_lot::Mutex::new(child),
1735 #[cfg(feature = "bundled-in-process")]
1736 ffi_host: parking_lot::Mutex::new(None),
1737 rpc,
1738 cwd,
1739 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1740 notification_tx: notification_broadcast_tx,
1741 router: router::SessionRouter::new(),
1742 github_token_registry: github_token_registry.clone(),
1743 negotiated_protocol_version: OnceLock::new(),
1744 state: parking_lot::Mutex::new(ConnectionState::Connected),
1745 lifecycle_tx: broadcast::channel(256).0,
1746 on_list_models,
1747 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1748 session_fs_configured,
1749 session_fs_sqlite_declared,
1750 llm_inference: OnceLock::new(),
1751 extension_launch_provider: extension_launch_provider.clone(),
1752 on_github_telemetry,
1753 on_get_trace_context,
1754 effective_connection_token,
1755 mode,
1756 startup_timings: OnceLock::new(),
1757 }),
1758 };
1759 github_token_registry.set_client(Arc::downgrade(&client.inner));
1760 extension_launch_provider.set_client(Arc::downgrade(&client.inner));
1761 client.spawn_lifecycle_dispatcher();
1762 debug!(
1763 elapsed_ms = setup_start.elapsed().as_millis(),
1764 pid = ?pid,
1765 "Client::from_transport setup complete"
1766 );
1767 Ok(client)
1768 }
1769
1770 fn spawn_lifecycle_dispatcher(&self) {
1774 let mut notif_rx = self.inner.notification_tx.subscribe();
1775 let lifecycle_tx = self.inner.lifecycle_tx.clone();
1776 tokio::spawn(async move {
1777 loop {
1778 match notif_rx.recv().await {
1779 Ok(notification) => {
1780 if notification.method != "session.lifecycle" {
1781 continue;
1782 }
1783 let Some(params) = notification.params.as_ref() else {
1784 continue;
1785 };
1786 let event: SessionLifecycleEvent =
1787 match serde_json::from_value(params.clone()) {
1788 Ok(e) => e,
1789 Err(e) => {
1790 warn!(
1791 error = %e,
1792 "failed to deserialize session.lifecycle notification"
1793 );
1794 continue;
1795 }
1796 };
1797 let _ = lifecycle_tx.send(event);
1800 }
1801 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
1802 warn!(missed = n, "lifecycle dispatcher lagged");
1803 }
1804 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
1805 }
1806 }
1807 });
1808 }
1809
1810 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
1811 let mut command = Command::new(program);
1812 command.kill_on_drop(true);
1813 for arg in &options.prefix_args {
1814 command.arg(arg);
1815 }
1816 if let Some(token) = &options.github_token {
1819 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1820 }
1821 if let Some(telemetry) = &options.telemetry {
1824 command.env("COPILOT_OTEL_ENABLED", "true");
1825 if let Some(endpoint) = &telemetry.otlp_endpoint {
1826 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1827 }
1828 if let Some(protocol) = telemetry.otlp_protocol {
1829 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
1830 }
1831 if let Some(path) = &telemetry.file_path {
1832 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
1833 }
1834 if let Some(exporter) = telemetry.exporter_type {
1835 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
1836 }
1837 if let Some(source) = &telemetry.source_name {
1838 command.env("COPILOT_OTEL_SOURCE_NAME", source);
1839 }
1840 if let Some(capture) = telemetry.capture_content {
1841 command.env(
1842 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
1843 if capture { "true" } else { "false" },
1844 );
1845 }
1846 }
1847 if let Some(dir) = &options.base_directory {
1848 command.env("COPILOT_HOME", dir);
1849 }
1850 if options.mode == ClientMode::Empty {
1853 command.env("COPILOT_DISABLE_KEYTAR", "1");
1854 }
1855 if let Transport::Tcp {
1856 connection_token: Some(token),
1857 ..
1858 } = &options.transport
1859 {
1860 command.env("COPILOT_CONNECTION_TOKEN", token);
1861 }
1862 for (key, value) in &options.env {
1863 command.env(key, value);
1864 }
1865 for key in &options.env_remove {
1866 command.env_remove(key);
1867 }
1868 command
1869 .current_dir(working_directory)
1870 .stdout(Stdio::piped())
1871 .stderr(Stdio::piped());
1872
1873 #[cfg(windows)]
1874 {
1875 use std::os::windows::process::CommandExt;
1876 const CREATE_NO_WINDOW: u32 = 0x08000000;
1877 command.as_std_mut().creation_flags(CREATE_NO_WINDOW);
1878 }
1879
1880 command
1881 }
1882
1883 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
1891 let mut args: Vec<&'static str> = Vec::new();
1892 if options.github_token.is_some() {
1893 args.push("--auth-token-env");
1894 args.push("COPILOT_SDK_AUTH_TOKEN");
1895 }
1896 let use_logged_in = options
1897 .use_logged_in_user
1898 .unwrap_or(options.github_token.is_none());
1899 if !use_logged_in {
1900 args.push("--no-auto-login");
1901 }
1902 args
1903 }
1904
1905 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
1909 match options.session_idle_timeout_seconds {
1910 Some(secs) if secs > 0 => {
1911 vec!["--session-idle-timeout".to_string(), secs.to_string()]
1912 }
1913 _ => Vec::new(),
1914 }
1915 }
1916
1917 fn remote_args(options: &ClientOptions) -> Vec<String> {
1918 if options.enable_remote_sessions {
1919 vec!["--remote".to_string()]
1920 } else {
1921 Vec::new()
1922 }
1923 }
1924
1925 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
1926 match options.log_level {
1927 Some(level) => vec!["--log-level", level.as_str()],
1928 None => Vec::new(),
1929 }
1930 }
1931
1932 fn spawn_stdio(
1933 program: &Path,
1934 options: &ClientOptions,
1935 working_directory: &Path,
1936 ) -> Result<(Child, Duration)> {
1937 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
1938 let mut command = Self::build_command(program, options, working_directory);
1939 command
1940 .args(["--server", "--stdio", "--no-auto-update"])
1941 .args(Self::log_level_args(options))
1942 .args(Self::auth_args(options))
1943 .args(Self::session_idle_timeout_args(options))
1944 .args(Self::remote_args(options))
1945 .args(&options.extra_args)
1946 .stdin(Stdio::piped());
1947 let spawn_start = Instant::now();
1948 let child = command.spawn()?;
1949 let spawn_elapsed = spawn_start.elapsed();
1950 debug!(
1951 elapsed_ms = spawn_elapsed.as_millis(),
1952 "Client::spawn_stdio subprocess spawned"
1953 );
1954 Ok((child, spawn_elapsed))
1955 }
1956
1957 async fn spawn_tcp(
1958 program: &Path,
1959 options: &ClientOptions,
1960 working_directory: &Path,
1961 port: u16,
1962 ) -> Result<(Child, u16, Duration, Duration)> {
1963 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
1964 let mut command = Self::build_command(program, options, working_directory);
1965 command
1966 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
1967 .args(Self::log_level_args(options))
1968 .args(Self::auth_args(options))
1969 .args(Self::session_idle_timeout_args(options))
1970 .args(Self::remote_args(options))
1971 .args(&options.extra_args)
1972 .stdin(Stdio::null());
1973 let spawn_start = Instant::now();
1974 let mut child = command.spawn()?;
1975 let spawn_elapsed = spawn_start.elapsed();
1976 debug!(
1977 elapsed_ms = spawn_elapsed.as_millis(),
1978 "Client::spawn_tcp subprocess spawned"
1979 );
1980 let stdout = child.stdout.take().expect("stdout is piped");
1981
1982 let (port_tx, port_rx) = oneshot::channel::<u16>();
1983 let span = tracing::error_span!("copilot_cli_port_scan");
1984 tokio::spawn(
1985 async move {
1986 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
1988 let mut lines = BufReader::new(stdout).lines();
1989 let mut port_tx = Some(port_tx);
1990 while let Ok(Some(line)) = lines.next_line().await {
1991 debug!(line = %line, "CLI stdout");
1992 if let Some(tx) = port_tx.take() {
1993 if let Some(caps) = port_re.captures(&line)
1994 && let Some(p) =
1995 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
1996 {
1997 let _ = tx.send(p);
1998 continue;
1999 }
2000 port_tx = Some(tx);
2002 }
2003 }
2004 }
2005 .instrument(span),
2006 );
2007
2008 let port_wait_start = Instant::now();
2009 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
2010 .await
2011 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
2012 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
2013
2014 let port_wait_elapsed = port_wait_start.elapsed();
2015 debug!(
2016 elapsed_ms = port_wait_elapsed.as_millis(),
2017 port = actual_port,
2018 "Client::spawn_tcp TCP port wait complete"
2019 );
2020 info!(port = %actual_port, "CLI server listening");
2021 Ok((child, actual_port, spawn_elapsed, port_wait_elapsed))
2022 }
2023
2024 fn drain_stderr(child: &mut Child) {
2025 if let Some(stderr) = child.stderr.take() {
2026 let span = tracing::error_span!("copilot_cli");
2027 tokio::spawn(
2028 async move {
2029 let mut reader = BufReader::new(stderr).lines();
2030 while let Ok(Some(line)) = reader.next_line().await {
2031 warn!(line = %line, "CLI stderr");
2032 }
2033 }
2034 .instrument(span),
2035 );
2036 }
2037 }
2038
2039 pub fn cwd(&self) -> &PathBuf {
2041 &self.inner.cwd
2042 }
2043
2044 pub fn mode(&self) -> ClientMode {
2046 self.inner.mode
2047 }
2048
2049 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
2060 crate::generated::rpc::ClientRpc { client: self }
2061 }
2062
2063 #[allow(dead_code, reason = "convenience for future internal use")]
2065 pub(crate) async fn send_request(
2066 &self,
2067 method: &str,
2068 params: Option<serde_json::Value>,
2069 ) -> Result<JsonRpcResponse> {
2070 self.inner.rpc.send_request(method, params).await
2071 }
2072
2073 pub async fn call(
2093 &self,
2094 method: &str,
2095 params: Option<serde_json::Value>,
2096 ) -> Result<serde_json::Value> {
2097 self.call_with_inline_callback(method, params, None).await
2098 }
2099
2100 pub(crate) async fn call_with_inline_callback(
2115 &self,
2116 method: &str,
2117 params: Option<serde_json::Value>,
2118 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
2119 ) -> Result<serde_json::Value> {
2120 let session_id: Option<SessionId> = params
2121 .as_ref()
2122 .and_then(|p| p.get("sessionId"))
2123 .and_then(|v| v.as_str())
2124 .map(SessionId::from);
2125 let response = self
2126 .inner
2127 .rpc
2128 .send_request_with_inline_callback(method, params, inline_callback)
2129 .await?;
2130 if let Some(err) = response.error {
2131 if err.message.contains("Session not found") {
2132 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
2133 session_id.unwrap_or_else(|| "unknown".into()),
2134 ))
2135 .into());
2136 }
2137 return Err(Error::with_message(
2138 ErrorKind::Rpc { code: err.code },
2139 err.message,
2140 ));
2141 }
2142 Ok(response.result.unwrap_or(serde_json::Value::Null))
2143 }
2144
2145 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
2147 self.inner.rpc.write(response).await
2148 }
2149
2150 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
2152 Self { inner }
2153 }
2154
2155 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
2159 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
2160 self.inner.request_rx.lock().take()
2161 }
2162
2163 pub(crate) fn register_session(
2171 &self,
2172 session_id: &SessionId,
2173 ) -> crate::router::SessionChannels {
2174 self.inner.router.ensure_started(
2175 &self.inner.notification_tx,
2176 &self.inner.request_rx,
2177 self.inner.extension_launch_provider.clone(),
2178 self.inner.llm_inference.get().cloned(),
2179 self.inner.on_github_telemetry.clone(),
2180 self.inner.github_token_registry.clone(),
2181 );
2182 self.inner.router.register(session_id)
2183 }
2184
2185 pub(crate) fn unregister_session(&self, session_id: &SessionId) {
2187 self.inner.router.unregister(session_id);
2188 }
2189
2190 pub(crate) fn register_github_token_provider(
2191 &self,
2192 provider: Arc<dyn GitHubTokenProvider>,
2193 ) -> github_token::GitHubTokenRegistration {
2194 self.inner.router.ensure_started(
2195 &self.inner.notification_tx,
2196 &self.inner.request_rx,
2197 self.inner.extension_launch_provider.clone(),
2198 self.inner.llm_inference.get().cloned(),
2199 self.inner.on_github_telemetry.clone(),
2200 self.inner.github_token_registry.clone(),
2201 );
2202 let id = self.inner.github_token_registry.register(provider);
2203 github_token::GitHubTokenRegistration::new(self.inner.github_token_registry.clone(), id)
2204 }
2205
2206 pub(crate) fn retire_github_token_provider(&self, session_id: &SessionId) {
2207 self.inner.github_token_registry.retire_session(session_id);
2208 }
2209
2210 pub fn protocol_version(&self) -> Option<u32> {
2217 self.inner.negotiated_protocol_version.get().copied()
2218 }
2219
2220 pub fn startup_timings(&self) -> Option<StartupTimings> {
2227 self.inner.startup_timings.get().cloned()
2228 }
2229
2230 pub async fn verify_protocol_version(&self) -> Result<()> {
2254 let handshake_start = Instant::now();
2255 let mut used_fallback_ping = false;
2256 let server_version = match self.connect_handshake().await {
2260 Ok(v) => v,
2261 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2262 used_fallback_ping = true;
2263 self.ping(None).await?.protocol_version
2264 }
2265 Err(e) => return Err(e),
2266 };
2267
2268 match server_version {
2269 None => {
2270 warn!("CLI server did not report protocolVersion; skipping version check");
2271 }
2272 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2273 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2274 server: v,
2275 min: MIN_PROTOCOL_VERSION,
2276 max: SDK_PROTOCOL_VERSION,
2277 })
2278 .into());
2279 }
2280 Some(v) => {
2281 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2282 if existing != v {
2283 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2284 previous: existing,
2285 current: v,
2286 })
2287 .into());
2288 }
2289 } else {
2290 let _ = self.inner.negotiated_protocol_version.set(v);
2291 }
2292 }
2293 }
2294
2295 debug!(
2296 elapsed_ms = handshake_start.elapsed().as_millis(),
2297 protocol_version = ?server_version,
2298 used_fallback_ping,
2299 "Client::verify_protocol_version protocol handshake complete"
2300 );
2301 Ok(())
2302 }
2303
2304 async fn connect_handshake(&self) -> Result<Option<u32>> {
2311 let params = crate::generated::api_types::ConnectRequest {
2312 token: self.inner.effective_connection_token.clone(),
2313 enable_git_hub_telemetry_forwarding: self
2314 .inner
2315 .on_github_telemetry
2316 .is_some()
2317 .then_some(true),
2318 ..Default::default()
2319 };
2320 let value = self
2321 .call(
2322 crate::generated::api_types::rpc_methods::CONNECT,
2323 Some(serde_json::to_value(params)?),
2324 )
2325 .await?;
2326 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2327 Ok(Some(u32::try_from(result.protocol_version).map_err(
2328 |_| ProtocolErrorKind::InvalidProtocolVersion {
2329 server: result.protocol_version,
2330 },
2331 )?))
2332 }
2333
2334 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2342 let params = match message {
2343 Some(m) => serde_json::json!({ "message": m }),
2344 None => serde_json::json!({}),
2345 };
2346 let value = self
2347 .call(generated::api_types::rpc_methods::PING, Some(params))
2348 .await?;
2349 Ok(serde_json::from_value(value)?)
2350 }
2351
2352 pub async fn list_sessions(
2355 &self,
2356 filter: Option<SessionListFilter>,
2357 ) -> Result<Vec<SessionMetadata>> {
2358 let params = match filter {
2359 Some(f) => serde_json::json!({ "filter": f }),
2360 None => serde_json::json!({}),
2361 };
2362 let result = self.call("session.list", Some(params)).await?;
2363 let response: ListSessionsResponse = serde_json::from_value(result)?;
2364 Ok(response.sessions)
2365 }
2366
2367 pub async fn get_session_metadata(
2385 &self,
2386 session_id: &SessionId,
2387 ) -> Result<Option<SessionMetadata>> {
2388 let result = self
2389 .call(
2390 "session.getMetadata",
2391 Some(serde_json::json!({ "sessionId": session_id })),
2392 )
2393 .await?;
2394 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2395 Ok(response.session)
2396 }
2397
2398 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2400 self.call(
2401 "session.delete",
2402 Some(serde_json::json!({ "sessionId": session_id })),
2403 )
2404 .await?;
2405 self.retire_github_token_provider(session_id);
2406 Ok(())
2407 }
2408
2409 #[cfg(feature = "test-support")]
2412 #[doc(hidden)]
2413 pub fn start_router_for_test(&self) {
2414 self.inner.router.ensure_started(
2415 &self.inner.notification_tx,
2416 &self.inner.request_rx,
2417 self.inner.extension_launch_provider.clone(),
2418 self.inner.llm_inference.get().cloned(),
2419 self.inner.on_github_telemetry.clone(),
2420 self.inner.github_token_registry.clone(),
2421 );
2422 }
2423
2424 #[cfg(feature = "test-support")]
2425 #[doc(hidden)]
2426 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2429 let mut first_error = None;
2430
2431 for session_id in self.inner.router.session_ids() {
2432 if let Err(error) = self
2433 .call(
2434 "session.destroy",
2435 Some(serde_json::json!({ "sessionId": session_id })),
2436 )
2437 .await
2438 && first_error.is_none()
2439 {
2440 first_error = Some(error);
2441 }
2442 self.inner.router.unregister(&session_id);
2443 }
2444 self.inner.github_token_registry.clear();
2445
2446 match self.list_sessions(None).await {
2447 Ok(sessions) => {
2448 for session in sessions {
2449 if let Err(error) = self.delete_session(&session.session_id).await
2450 && first_error.is_none()
2451 {
2452 first_error = Some(error);
2453 }
2454 }
2455 }
2456 Err(error) if first_error.is_none() => first_error = Some(error),
2457 Err(_) => {}
2458 }
2459
2460 match first_error {
2461 Some(error) => Err(error),
2462 None => Ok(()),
2463 }
2464 }
2465
2466 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2482 let result = self
2483 .call("session.getLastId", Some(serde_json::json!({})))
2484 .await?;
2485 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2486 Ok(response.session_id)
2487 }
2488
2489 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2494 let result = self
2495 .call("session.getForeground", Some(serde_json::json!({})))
2496 .await?;
2497 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2498 Ok(response.session_id)
2499 }
2500
2501 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2506 self.call(
2507 "session.setForeground",
2508 Some(serde_json::json!({ "sessionId": session_id })),
2509 )
2510 .await?;
2511 Ok(())
2512 }
2513
2514 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2516 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2517 Ok(serde_json::from_value(result)?)
2518 }
2519
2520 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2522 let result = self
2523 .call("auth.getStatus", Some(serde_json::json!({})))
2524 .await?;
2525 Ok(serde_json::from_value(result)?)
2526 }
2527
2528 pub async fn list_models(&self) -> Result<Vec<Model>> {
2533 let cache = self.inner.models_cache.lock().clone();
2534 let models = cache
2535 .get_or_try_init(|| async {
2536 if let Some(handler) = &self.inner.on_list_models {
2537 handler.list_models().await
2538 } else {
2539 Ok(self.rpc().models().list().await?.models)
2540 }
2541 })
2542 .await?;
2543 Ok(models.clone())
2544 }
2545
2546 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2549 if let Some(provider) = &self.inner.on_get_trace_context {
2550 provider.get_trace_context().await
2551 } else {
2552 TraceContext::default()
2553 }
2554 }
2555
2556 pub fn pid(&self) -> Option<u32> {
2558 self.inner.child.lock().as_ref().and_then(|c| c.id())
2559 }
2560
2561 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2588 let pid = self.pid();
2589 info!(pid = ?pid, "stopping CLI process");
2590 let mut errors: Vec<Error> = Vec::new();
2591 self.inner.extension_launch_provider.clear();
2592
2593 for session_id in self.inner.router.session_ids() {
2596 match self
2597 .call(
2598 "session.destroy",
2599 Some(serde_json::json!({ "sessionId": session_id })),
2600 )
2601 .await
2602 {
2603 Ok(_) => {}
2604 Err(e) => {
2605 warn!(
2606 session_id = %session_id,
2607 error = %e,
2608 "session.destroy failed during Client::stop",
2609 );
2610 errors.push(e);
2611 }
2612 }
2613 self.inner.router.unregister(&session_id);
2614 }
2615 self.inner.github_token_registry.clear();
2616
2617 let should_shutdown_runtime = self.inner.child.lock().is_some();
2618 #[cfg(feature = "bundled-in-process")]
2619 let should_shutdown_runtime =
2620 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2621 if should_shutdown_runtime {
2622 let runtime_shutdown_start = Instant::now();
2623 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2624 .await
2625 {
2626 Ok(Ok(())) => {
2627 debug!(
2628 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2629 "Client::stop runtime shutdown complete"
2630 );
2631 }
2632 Ok(Err(e)) => {
2633 warn!(
2634 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2635 error = %e,
2636 "runtime.shutdown failed during Client::stop",
2637 );
2638 errors.push(e);
2639 }
2640 Err(_) => {
2641 let e = std::io::Error::new(
2642 std::io::ErrorKind::TimedOut,
2643 "runtime.shutdown timed out during Client::stop",
2644 );
2645 warn!(
2646 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2647 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2648 error = %e,
2649 "runtime.shutdown timed out during Client::stop",
2650 );
2651 errors.push(e.into());
2652 }
2653 }
2654 }
2655
2656 let child = self.inner.child.lock().take();
2657 *self.inner.state.lock() = ConnectionState::Disconnected;
2658 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2659 if let Some(mut child) = child {
2660 match child.try_wait() {
2661 Ok(Some(_status)) => {}
2662 Ok(None) => {
2663 if let Err(e) = child.kill().await {
2670 errors.push(e.into());
2671 }
2672 }
2673 Err(e) => errors.push(e.into()),
2674 }
2675 }
2676
2677 self.inner.rpc.force_close();
2680 #[cfg(feature = "bundled-in-process")]
2681 {
2682 if let Some(host) = self.inner.ffi_host.lock().take() {
2683 host.close();
2684 }
2685 }
2686
2687 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2688 if errors.is_empty() {
2689 Ok(())
2690 } else {
2691 Err(StopErrors(errors))
2692 }
2693 }
2694
2695 pub fn force_stop(&self) {
2725 let pid = self.pid();
2726 info!(pid = ?pid, "force-stopping CLI process");
2727 self.inner.extension_launch_provider.clear();
2728 if let Some(mut child) = self.inner.child.lock().take()
2729 && let Err(e) = child.start_kill()
2730 {
2731 error!(pid = ?pid, error = %e, "failed to send kill signal");
2732 }
2733 self.inner.rpc.force_close();
2734 #[cfg(feature = "bundled-in-process")]
2735 {
2736 if let Some(host) = self.inner.ffi_host.lock().take() {
2737 host.close();
2738 }
2739 }
2740 self.inner.router.clear();
2743 self.inner.github_token_registry.clear();
2744 *self.inner.state.lock() = ConnectionState::Disconnected;
2745 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2746 }
2747
2748 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2783 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2784 }
2785}
2786
2787impl Drop for ClientInner {
2788 fn drop(&mut self) {
2789 if let Some(ref mut child) = *self.child.lock() {
2790 let pid = child.id();
2791 if let Err(e) = child.start_kill() {
2792 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2793 } else {
2794 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2795 }
2796 }
2797 #[cfg(feature = "bundled-in-process")]
2798 {
2799 if let Some(host) = self.ffi_host.lock().take() {
2800 self.rpc.force_close();
2801 host.close();
2802 }
2803 }
2804 }
2805}
2806
2807#[cfg(test)]
2808mod tests {
2809 use super::*;
2810
2811 #[test]
2812 fn is_transport_failure_matches_request_cancelled() {
2813 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2814 assert!(err.is_transport_failure());
2815 }
2816
2817 #[test]
2818 fn is_transport_failure_matches_io_error() {
2819 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2820 assert!(err.is_transport_failure());
2821 }
2822
2823 #[test]
2824 fn is_transport_failure_rejects_rpc_error() {
2825 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2826 assert!(!err.is_transport_failure());
2827 }
2828
2829 #[test]
2830 fn is_transport_failure_rejects_session_error() {
2831 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2832 assert!(!err.is_transport_failure());
2833 }
2834
2835 #[test]
2836 fn client_options_builder_composes() {
2837 let opts = ClientOptions::new()
2838 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2839 .with_prefix_args(["node"])
2840 .with_cwd(PathBuf::from("/tmp"))
2841 .with_env([("KEY", "value")])
2842 .with_env_remove(["UNWANTED"])
2843 .with_extra_args(["--quiet"])
2844 .with_github_token("ghp_test")
2845 .with_use_logged_in_user(false)
2846 .with_log_level(LogLevel::Debug)
2847 .with_session_idle_timeout_seconds(120)
2848 .with_enable_remote_sessions(true);
2849 assert!(matches!(opts.program, CliProgram::Path(_)));
2850 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2851 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2852 assert_eq!(
2853 opts.env,
2854 vec![(
2855 std::ffi::OsString::from("KEY"),
2856 std::ffi::OsString::from("value")
2857 )]
2858 );
2859 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2860 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2861 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2862 assert_eq!(opts.use_logged_in_user, Some(false));
2863 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2864 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2865 assert!(opts.enable_remote_sessions);
2866 }
2867
2868 #[test]
2869 fn default_transport_values_resolve_without_process_state() {
2870 assert!(matches!(
2871 resolve_default_transport_value(None).unwrap(),
2872 Transport::Stdio
2873 ));
2874 assert!(matches!(
2875 resolve_default_transport_value(Some("stdio")).unwrap(),
2876 Transport::Stdio
2877 ));
2878 assert!(matches!(
2879 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2880 Transport::InProcess
2881 ));
2882 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2883 }
2884
2885 #[test]
2886 fn inprocess_rejects_process_scoped_options() {
2887 let invalid = [
2888 ClientOptions::new().with_cwd("."),
2889 ClientOptions::new().with_env([("KEY", "value")]),
2890 ClientOptions::new().with_env_remove(["KEY"]),
2891 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2892 ClientOptions::new().with_prefix_args(["index.js"]),
2893 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
2894 ClientOptions::new().with_extra_args(["--verbose"]),
2895 ];
2896
2897 for options in invalid {
2898 assert!(validate_inprocess_options(&options).is_err());
2899 }
2900 }
2901
2902 #[test]
2903 fn inprocess_allows_typed_runtime_options() {
2904 let options = ClientOptions::new()
2905 .with_base_directory("state")
2906 .with_log_level(LogLevel::Debug)
2907 .with_session_idle_timeout_seconds(10)
2908 .with_github_token("token")
2909 .with_use_logged_in_user(false)
2910 .with_enable_remote_sessions(true);
2911
2912 assert!(validate_inprocess_options(&options).is_ok());
2913 }
2914
2915 #[cfg(not(feature = "bundled-in-process"))]
2916 #[tokio::test]
2917 async fn inprocess_requires_cargo_feature() {
2918 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
2919 .await
2920 .unwrap_err();
2921
2922 assert!(error.to_string().contains("bundled-in-process"));
2923 }
2924
2925 #[test]
2926 fn is_transport_failure_rejects_other_protocol_errors() {
2927 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2928 assert!(!err.is_transport_failure());
2929 }
2930
2931 #[test]
2932 fn build_command_lets_env_remove_strip_injected_token() {
2933 let opts = ClientOptions {
2934 github_token: Some("secret".to_string()),
2935 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2936 ..Default::default()
2937 };
2938 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2939 let action = cmd
2941 .as_std()
2942 .get_envs()
2943 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2944 .map(|(_, v)| v);
2945 assert_eq!(
2946 action,
2947 Some(None),
2948 "env_remove should win over github_token"
2949 );
2950 }
2951
2952 #[test]
2953 fn build_command_lets_env_override_injected_token() {
2954 let opts = ClientOptions {
2955 github_token: Some("from-options".to_string()),
2956 env: vec![(
2957 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2958 std::ffi::OsString::from("from-env"),
2959 )],
2960 ..Default::default()
2961 };
2962 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2963 let value = cmd
2964 .as_std()
2965 .get_envs()
2966 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2967 .and_then(|(_, v)| v);
2968 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2969 }
2970
2971 #[test]
2972 fn build_command_injects_github_token_by_default() {
2973 let opts = ClientOptions {
2974 github_token: Some("just-the-token".to_string()),
2975 ..Default::default()
2976 };
2977 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2978 let value = cmd
2979 .as_std()
2980 .get_envs()
2981 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2982 .and_then(|(_, v)| v);
2983 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2984 }
2985
2986 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2987 cmd.as_std()
2988 .get_envs()
2989 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2990 .and_then(|(_, v)| v)
2991 }
2992
2993 #[test]
2994 fn telemetry_config_builder_composes() {
2995 let cfg = TelemetryConfig::new()
2996 .with_otlp_endpoint("http://collector:4318")
2997 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2998 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2999 .with_exporter_type(OtelExporterType::OtlpHttp)
3000 .with_source_name("my-app")
3001 .with_capture_content(true);
3002
3003 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
3004 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
3005 assert_eq!(
3006 cfg.file_path.as_deref(),
3007 Some(Path::new("/var/log/copilot.jsonl")),
3008 );
3009 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
3010 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
3011 assert_eq!(cfg.capture_content, Some(true));
3012 assert!(!cfg.is_empty());
3013 assert!(TelemetryConfig::new().is_empty());
3014 }
3015
3016 #[test]
3017 fn otlp_http_protocol_serde_matches_env_value() {
3018 for (protocol, wire) in [
3019 (OtlpHttpProtocol::HttpJson, "http/json"),
3020 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
3021 ] {
3022 assert_eq!(protocol.as_str(), wire);
3023
3024 let serialized = serde_json::to_string(&protocol).unwrap();
3025 assert_eq!(serialized, format!("\"{wire}\""));
3026
3027 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
3028 assert_eq!(deserialized, protocol);
3029 }
3030 }
3031
3032 #[test]
3033 fn build_command_sets_otel_env_when_telemetry_enabled() {
3034 let opts = ClientOptions {
3035 telemetry: Some(TelemetryConfig {
3036 otlp_endpoint: Some("http://collector:4318".to_string()),
3037 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
3038 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
3039 exporter_type: Some(OtelExporterType::OtlpHttp),
3040 source_name: Some("my-app".to_string()),
3041 capture_content: Some(true),
3042 }),
3043 ..Default::default()
3044 };
3045 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3046 assert_eq!(
3047 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
3048 Some(std::ffi::OsStr::new("true")),
3049 );
3050 assert_eq!(
3051 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3052 Some(std::ffi::OsStr::new("http://collector:4318")),
3053 );
3054 assert_eq!(
3055 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
3056 Some(std::ffi::OsStr::new("http/protobuf")),
3057 );
3058 assert_eq!(
3059 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
3060 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
3061 );
3062 assert_eq!(
3063 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
3064 Some(std::ffi::OsStr::new("otlp-http")),
3065 );
3066 assert_eq!(
3067 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
3068 Some(std::ffi::OsStr::new("my-app")),
3069 );
3070 assert_eq!(
3071 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
3072 Some(std::ffi::OsStr::new("true")),
3073 );
3074 }
3075
3076 #[test]
3077 fn build_command_omits_otel_env_when_telemetry_none() {
3078 let opts = ClientOptions::default();
3079 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3080 for key in [
3081 "COPILOT_OTEL_ENABLED",
3082 "OTEL_EXPORTER_OTLP_ENDPOINT",
3083 "OTEL_EXPORTER_OTLP_PROTOCOL",
3084 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3085 "COPILOT_OTEL_EXPORTER_TYPE",
3086 "COPILOT_OTEL_SOURCE_NAME",
3087 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3088 ] {
3089 assert!(
3090 env_value(&cmd, key).is_none(),
3091 "expected {key} to be unset when telemetry is None",
3092 );
3093 }
3094 }
3095
3096 #[test]
3097 fn build_command_omits_unset_telemetry_fields() {
3098 let opts = ClientOptions {
3099 telemetry: Some(TelemetryConfig {
3100 otlp_endpoint: Some("http://collector:4318".to_string()),
3101 ..Default::default()
3102 }),
3103 ..Default::default()
3104 };
3105 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3106 assert_eq!(
3108 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
3109 Some(std::ffi::OsStr::new("true")),
3110 );
3111 assert_eq!(
3112 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3113 Some(std::ffi::OsStr::new("http://collector:4318")),
3114 );
3115 for key in [
3117 "OTEL_EXPORTER_OTLP_PROTOCOL",
3118 "COPILOT_OTEL_FILE_EXPORTER_PATH",
3119 "COPILOT_OTEL_EXPORTER_TYPE",
3120 "COPILOT_OTEL_SOURCE_NAME",
3121 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
3122 ] {
3123 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
3124 }
3125 }
3126
3127 #[test]
3128 fn build_command_lets_user_env_override_telemetry() {
3129 let opts = ClientOptions {
3130 telemetry: Some(TelemetryConfig {
3131 otlp_endpoint: Some("http://from-config:4318".to_string()),
3132 ..Default::default()
3133 }),
3134 env: vec![(
3135 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
3136 std::ffi::OsString::from("http://from-user-env:4318"),
3137 )],
3138 ..Default::default()
3139 };
3140 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3141 assert_eq!(
3142 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3143 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
3144 "user-supplied options.env should override telemetry config",
3145 );
3146 }
3147
3148 #[test]
3149 fn build_command_sets_copilot_home_env_when_configured() {
3150 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
3151 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3152 assert_eq!(
3153 env_value(&cmd, "COPILOT_HOME"),
3154 Some(std::ffi::OsStr::new("/custom/copilot")),
3155 );
3156
3157 let opts = ClientOptions::default();
3158 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3159 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
3160 }
3161
3162 #[test]
3163 fn build_command_sets_connection_token_env_when_configured() {
3164 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3165 port: 0,
3166 connection_token: Some("secret-token".to_string()),
3167 });
3168 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3169 assert_eq!(
3170 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3171 Some(std::ffi::OsStr::new("secret-token")),
3172 );
3173
3174 let opts = ClientOptions::default();
3175 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3176 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3177 }
3178
3179 #[tokio::test]
3180 async fn start_rejects_empty_connection_token() {
3181 let opts = ClientOptions::new()
3182 .with_transport(Transport::Tcp {
3183 port: 0,
3184 connection_token: Some(String::new()),
3185 })
3186 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3187 let err = Client::start(opts).await.unwrap_err();
3188 assert!(
3189 matches!(err.kind(), ErrorKind::InvalidConfig),
3190 "got {err:?}"
3191 );
3192 }
3193
3194 #[tokio::test]
3195 async fn start_rejects_empty_external_connection_token() {
3196 let opts = ClientOptions::new()
3197 .with_transport(Transport::External {
3198 host: "127.0.0.1".to_string(),
3199 port: 1,
3200 connection_token: Some(String::new()),
3201 })
3202 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3203 let err = Client::start(opts).await.unwrap_err();
3204 assert!(
3205 matches!(err.kind(), ErrorKind::InvalidConfig),
3206 "got {err:?}"
3207 );
3208 }
3209
3210 #[test]
3211 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3212 let opts_true = ClientOptions {
3213 telemetry: Some(TelemetryConfig {
3214 capture_content: Some(true),
3215 ..Default::default()
3216 }),
3217 ..Default::default()
3218 };
3219 let opts_false = ClientOptions {
3220 telemetry: Some(TelemetryConfig {
3221 capture_content: Some(false),
3222 ..Default::default()
3223 }),
3224 ..Default::default()
3225 };
3226 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3227 let cmd_false =
3228 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3229 assert_eq!(
3230 env_value(
3231 &cmd_true,
3232 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3233 ),
3234 Some(std::ffi::OsStr::new("true")),
3235 );
3236 assert_eq!(
3237 env_value(
3238 &cmd_false,
3239 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3240 ),
3241 Some(std::ffi::OsStr::new("false")),
3242 );
3243 }
3244
3245 #[test]
3246 fn session_idle_timeout_args_are_omitted_by_default() {
3247 let opts = ClientOptions::default();
3248 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3249 }
3250
3251 #[test]
3252 fn session_idle_timeout_args_omitted_for_zero() {
3253 let opts = ClientOptions {
3254 session_idle_timeout_seconds: Some(0),
3255 ..Default::default()
3256 };
3257 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3258 }
3259
3260 #[test]
3261 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3262 let opts = ClientOptions {
3263 session_idle_timeout_seconds: Some(300),
3264 ..Default::default()
3265 };
3266 assert_eq!(
3267 Client::session_idle_timeout_args(&opts),
3268 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3269 );
3270 }
3271
3272 #[test]
3273 fn remote_args_omitted_by_default() {
3274 let opts = ClientOptions::default();
3275 assert!(Client::remote_args(&opts).is_empty());
3276 }
3277
3278 #[test]
3279 fn remote_args_emit_flag_when_enabled() {
3280 let opts = ClientOptions {
3281 enable_remote_sessions: true,
3282 ..Default::default()
3283 };
3284 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3285 }
3286
3287 #[test]
3288 fn log_level_args_omitted_when_unset() {
3289 let opts = ClientOptions::default();
3290 assert!(opts.log_level.is_none());
3291 assert!(
3292 Client::log_level_args(&opts).is_empty(),
3293 "with no caller-supplied log_level the SDK must not pass --log-level"
3294 );
3295 }
3296
3297 #[test]
3298 fn log_level_args_emit_flag_when_set() {
3299 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3300 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3301 }
3302
3303 #[test]
3304 fn log_level_str_round_trips() {
3305 for level in [
3306 LogLevel::None,
3307 LogLevel::Error,
3308 LogLevel::Warning,
3309 LogLevel::Info,
3310 LogLevel::Debug,
3311 LogLevel::All,
3312 ] {
3313 let s = level.as_str();
3314 let json = serde_json::to_string(&level).unwrap();
3315 assert_eq!(json, format!("\"{s}\""));
3316 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3317 assert_eq!(parsed, level);
3318 }
3319 }
3320
3321 #[test]
3322 fn client_options_debug_redacts_handler() {
3323 struct StubHandler;
3324 #[async_trait]
3325 impl ListModelsHandler for StubHandler {
3326 async fn list_models(&self) -> Result<Vec<Model>> {
3327 Ok(vec![])
3328 }
3329 }
3330 let opts = ClientOptions {
3331 on_list_models: Some(Arc::new(StubHandler)),
3332 github_token: Some("secret-token".into()),
3333 ..Default::default()
3334 };
3335 let debug = format!("{opts:?}");
3336 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3337 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3338 assert!(!debug.contains("secret-token"));
3339 }
3340
3341 #[tokio::test]
3342 async fn list_models_uses_on_list_models_handler_when_set() {
3343 use std::sync::atomic::{AtomicUsize, Ordering};
3344
3345 struct CountingHandler {
3346 calls: Arc<AtomicUsize>,
3347 models: Vec<Model>,
3348 }
3349 #[async_trait]
3350 impl ListModelsHandler for CountingHandler {
3351 async fn list_models(&self) -> Result<Vec<Model>> {
3352 self.calls.fetch_add(1, Ordering::SeqCst);
3353 Ok(self.models.clone())
3354 }
3355 }
3356
3357 let calls = Arc::new(AtomicUsize::new(0));
3358 let model = Model {
3359 id: "byok-gpt-4".into(),
3360 name: "BYOK GPT-4".into(),
3361 ..Default::default()
3362 };
3363 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3364 calls: Arc::clone(&calls),
3365 models: vec![model.clone()],
3366 });
3367
3368 let client = client_with_list_models_handler(handler);
3369
3370 let result = client.list_models().await.unwrap();
3371 assert_eq!(result.len(), 1);
3372 assert_eq!(result[0].id, "byok-gpt-4");
3373 assert_eq!(calls.load(Ordering::SeqCst), 1);
3374 }
3375
3376 #[tokio::test]
3377 async fn list_models_serializes_concurrent_cache_misses() {
3378 use std::sync::atomic::{AtomicUsize, Ordering};
3379
3380 struct SlowCountingHandler {
3381 calls: Arc<AtomicUsize>,
3382 models: Vec<Model>,
3383 }
3384 #[async_trait]
3385 impl ListModelsHandler for SlowCountingHandler {
3386 async fn list_models(&self) -> Result<Vec<Model>> {
3387 self.calls.fetch_add(1, Ordering::SeqCst);
3388 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3389 Ok(self.models.clone())
3390 }
3391 }
3392
3393 let calls = Arc::new(AtomicUsize::new(0));
3394 let model = Model {
3395 id: "single-flight-model".into(),
3396 name: "Single Flight Model".into(),
3397 ..Default::default()
3398 };
3399 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3400 calls: Arc::clone(&calls),
3401 models: vec![model],
3402 });
3403 let client = client_with_list_models_handler(handler);
3404
3405 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3406 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3407 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3408 assert_eq!(calls.load(Ordering::SeqCst), 1);
3409 }
3410
3411 #[tokio::test]
3412 async fn cancelled_resume_session_unregisters_pending_session() {
3413 let (client_write, _server_read) = tokio::io::duplex(8192);
3414 let (_server_write, client_read) = tokio::io::duplex(8192);
3415 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3416 assert!(client.startup_timings().is_none());
3417 let session_id = SessionId::new("resume-cancel-test");
3418 let handle = tokio::spawn({
3419 let client = client.clone();
3420 async move {
3421 client
3422 .resume_session(ResumeSessionConfig::new(session_id))
3423 .await
3424 }
3425 });
3426
3427 wait_for_pending_session_registration(&client).await;
3428 handle.abort();
3429 let _ = handle.await;
3430
3431 assert!(client.inner.router.session_ids().is_empty());
3432 client.force_stop();
3433 }
3434
3435 #[cfg(any(unix, windows))]
3436 #[tokio::test]
3437 async fn dropping_last_client_kills_spawned_cli() {
3438 let temp = tempfile::tempdir().unwrap();
3439 let ready = temp.path().join("ready");
3440 let survived = temp.path().join("survived");
3441 let child = test_child_command(temp.path(), &ready, &survived)
3442 .spawn()
3443 .unwrap();
3444 let (client_write, _server_read) = tokio::io::duplex(64);
3445 let (_server_write, client_read) = tokio::io::duplex(64);
3446 let client = Client::from_transport(
3447 client_read,
3448 client_write,
3449 Some(child),
3450 temp.path().to_path_buf(),
3451 None,
3452 None,
3453 false,
3454 false,
3455 None,
3456 None,
3457 None,
3458 ClientMode::default(),
3459 )
3460 .unwrap();
3461
3462 wait_for_test_child(&ready).await;
3463 drop(client);
3464
3465 assert_test_child_killed(&survived).await;
3466 }
3467
3468 #[cfg(any(unix, windows))]
3469 #[tokio::test]
3470 async fn spawned_child_is_killed_when_dropped() {
3471 let temp = tempfile::tempdir().unwrap();
3472 let ready = temp.path().join("ready");
3473 let survived = temp.path().join("survived");
3474 let child = test_child_command(temp.path(), &ready, &survived)
3475 .spawn()
3476 .unwrap();
3477
3478 wait_for_test_child(&ready).await;
3479 drop(child);
3480
3481 assert_test_child_killed(&survived).await;
3482 }
3483
3484 #[cfg(any(unix, windows))]
3485 fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command {
3486 #[cfg(unix)]
3487 let mut command = {
3488 let mut command =
3489 Client::build_command(Path::new("sh"), &ClientOptions::default(), temp);
3490 command.args([
3491 "-c",
3492 "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"",
3493 ]);
3494 command
3495 };
3496 #[cfg(windows)]
3497 let mut command = {
3498 let mut command =
3499 Client::build_command(Path::new("powershell.exe"), &ClientOptions::default(), temp);
3500 command.args([
3501 "-NoLogo",
3502 "-NoProfile",
3503 "-NonInteractive",
3504 "-Command",
3505 "Set-Content -LiteralPath $env:READY ready; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:SURVIVED survived",
3506 ]);
3507 command
3508 };
3509 command.env("READY", ready).env("SURVIVED", survived);
3510 command
3511 }
3512
3513 #[cfg(any(unix, windows))]
3514 async fn wait_for_test_child(ready: &Path) {
3515 let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
3516 while !ready.exists() {
3517 assert!(
3518 tokio::time::Instant::now() < deadline,
3519 "child did not report readiness"
3520 );
3521 tokio::time::sleep(Duration::from_millis(10)).await;
3522 }
3523 }
3524
3525 #[cfg(any(unix, windows))]
3526 async fn assert_test_child_killed(survived: &Path) {
3527 tokio::time::sleep(Duration::from_millis(1500)).await;
3528
3529 assert!(
3530 !survived.exists(),
3531 "child survived after its owner was dropped"
3532 );
3533 }
3534
3535 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3536 Client {
3537 inner: Arc::new(ClientInner {
3538 child: parking_lot::Mutex::new(None),
3539 #[cfg(feature = "bundled-in-process")]
3540 ffi_host: parking_lot::Mutex::new(None),
3541 rpc: {
3542 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3543 let (notif_tx, _notif_rx) = broadcast::channel(16);
3544 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3545 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3546 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3547 },
3548 cwd: PathBuf::from("."),
3549 request_rx: parking_lot::Mutex::new(None),
3550 notification_tx: broadcast::channel(16).0,
3551 router: router::SessionRouter::new(),
3552 github_token_registry: Arc::new(github_token::GitHubTokenRegistry::new()),
3553 negotiated_protocol_version: OnceLock::new(),
3554 state: parking_lot::Mutex::new(ConnectionState::Connected),
3555 lifecycle_tx: broadcast::channel(16).0,
3556 on_list_models: Some(handler),
3557 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3558 session_fs_configured: false,
3559 session_fs_sqlite_declared: false,
3560 llm_inference: OnceLock::new(),
3561 extension_launch_provider: Arc::new(
3562 extension_launch_provider::ExtensionLaunchProviderDispatcher::new(None),
3563 ),
3564 on_github_telemetry: None,
3565 on_get_trace_context: None,
3566 effective_connection_token: None,
3567 mode: ClientMode::default(),
3568 startup_timings: OnceLock::new(),
3569 }),
3570 }
3571 }
3572
3573 async fn wait_for_pending_session_registration(client: &Client) {
3574 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3575 while client.inner.router.session_ids().is_empty() {
3576 assert!(
3577 tokio::time::Instant::now() < deadline,
3578 "session was not registered"
3579 );
3580 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3581 }
3582 }
3583}