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;
13#[cfg(feature = "bundled-in-process")]
15pub(crate) mod ffi;
16pub use errors::*;
17pub mod copilot_request_handler;
21#[doc(hidden)]
24pub mod github_telemetry;
25pub mod handler;
27pub mod hooks;
29mod jsonrpc;
30pub mod permission;
32pub mod provider_token;
34mod provider_token_dispatch;
35pub(crate) mod resolve;
37mod router;
38pub mod session;
40pub mod session_fs;
42mod session_fs_dispatch;
43pub mod startup_timings;
45pub mod subscription;
47pub mod tool;
49pub mod trace_context;
51pub mod transforms;
53pub mod types;
55mod wire;
56
57pub mod session_events;
59
60pub mod rpc;
63
64pub(crate) mod generated;
69
70pub mod mode;
73
74use std::ffi::OsString;
75use std::path::{Path, PathBuf};
76use std::process::Stdio;
77use std::sync::{Arc, OnceLock};
78use std::time::{Duration, Instant};
79
80use async_trait::async_trait;
81pub use indexmap::IndexMap;
85pub(crate) use jsonrpc::{
88 JsonRpcClient, JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, error_codes,
89};
90pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet};
91pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs};
92
93#[cfg(feature = "test-support")]
95pub mod test_support {
96 pub use crate::jsonrpc::{
97 JsonRpcClient, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
98 error_codes,
99 };
100}
101use serde::{Deserialize, Serialize};
102use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader};
103use tokio::net::TcpStream;
104use tokio::process::{Child, Command};
105use tokio::sync::{broadcast, mpsc, oneshot};
106use tracing::{Instrument, debug, error, info, warn};
107pub use types::*;
108
109mod sdk_protocol_version;
110pub use sdk_protocol_version::{SDK_PROTOCOL_VERSION, get_sdk_protocol_version};
111pub use startup_timings::StartupTimings;
112pub use subscription::{EventSubscription, LifecycleSubscription};
113
114const MIN_PROTOCOL_VERSION: u32 = 3;
116const RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
117
118fn record_optional_millis(span: &tracing::Span, field: &'static str, value: Option<u64>) {
119 match value {
120 Some(value) => {
121 span.record(field, value);
122 }
123 None => {
124 span.record(field, "None");
125 }
126 }
127}
128
129#[derive(Debug, Default)]
131#[non_exhaustive]
132pub enum Transport {
133 #[default]
136 Default,
137 Stdio,
139 InProcess,
152 Tcp {
154 port: u16,
156 connection_token: Option<String>,
160 },
161 External {
163 host: String,
165 port: u16,
167 connection_token: Option<String>,
170 },
171}
172
173#[derive(Debug, Clone, Default)]
175pub enum CliProgram {
176 #[default]
179 Resolve,
180 Path(PathBuf),
182}
183
184impl From<PathBuf> for CliProgram {
185 fn from(path: PathBuf) -> Self {
186 Self::Path(path)
187 }
188}
189
190pub const HAS_BUNDLED_CLI: bool = cfg!(has_bundled_cli);
197
198pub fn install_bundled_cli() -> Option<PathBuf> {
222 #[cfg(feature = "bundled-cli")]
223 {
224 embeddedcli::path()
225 }
226 #[cfg(not(feature = "bundled-cli"))]
227 {
228 None
229 }
230}
231
232#[non_exhaustive]
242pub struct ClientOptions {
243 pub program: CliProgram,
245 pub prefix_args: Vec<OsString>,
247 pub working_directory: PathBuf,
251 pub env: Vec<(OsString, OsString)>,
253 pub env_remove: Vec<OsString>,
255 pub extra_args: Vec<String>,
257 pub builtin_plugin_directories: Vec<PathBuf>,
262 pub transport: Transport,
264 pub github_token: Option<String>,
269 pub use_logged_in_user: Option<bool>,
273 pub log_level: Option<LogLevel>,
277 pub session_idle_timeout_seconds: Option<u64>,
283 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
291 pub session_fs: Option<SessionFsConfig>,
299 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
308 #[doc(hidden)]
316 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
317 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
327 pub telemetry: Option<TelemetryConfig>,
331 pub base_directory: Option<PathBuf>,
336 pub enable_remote_sessions: bool,
342 pub bundled_cli_extract_dir: Option<PathBuf>,
361 pub mode: ClientMode,
365}
366
367impl std::fmt::Debug for ClientOptions {
368 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
369 f.debug_struct("ClientOptions")
370 .field("program", &self.program)
371 .field("prefix_args", &self.prefix_args)
372 .field("working_directory", &self.working_directory)
373 .field("env", &self.env)
374 .field("env_remove", &self.env_remove)
375 .field("extra_args", &self.extra_args)
376 .field(
377 "builtin_plugin_directories",
378 &self.builtin_plugin_directories,
379 )
380 .field("transport", &self.transport)
381 .field(
382 "github_token",
383 &self.github_token.as_ref().map(|_| "<redacted>"),
384 )
385 .field("use_logged_in_user", &self.use_logged_in_user)
386 .field("log_level", &self.log_level)
387 .field(
388 "session_idle_timeout_seconds",
389 &self.session_idle_timeout_seconds,
390 )
391 .field(
392 "on_list_models",
393 &self.on_list_models.as_ref().map(|_| "<set>"),
394 )
395 .field("session_fs", &self.session_fs)
396 .field(
397 "request_handler",
398 &self.request_handler.as_ref().map(|_| "<set>"),
399 )
400 .field(
401 "on_github_telemetry",
402 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
403 )
404 .field(
405 "on_get_trace_context",
406 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
407 )
408 .field("telemetry", &self.telemetry)
409 .field("base_directory", &self.base_directory)
410 .field("enable_remote_sessions", &self.enable_remote_sessions)
411 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
412 .finish()
413 }
414}
415
416#[async_trait]
425pub trait ListModelsHandler: Send + Sync + 'static {
426 async fn list_models(&self) -> Result<Vec<Model>>;
428}
429
430#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
432#[serde(rename_all = "lowercase")]
433pub enum LogLevel {
434 None,
436 Error,
438 Warning,
440 Info,
442 Debug,
444 All,
446}
447
448impl LogLevel {
449 pub fn as_str(self) -> &'static str {
451 match self {
452 Self::None => "none",
453 Self::Error => "error",
454 Self::Warning => "warning",
455 Self::Info => "info",
456 Self::Debug => "debug",
457 Self::All => "all",
458 }
459 }
460}
461
462impl std::fmt::Display for LogLevel {
463 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
464 f.write_str(self.as_str())
465 }
466}
467
468#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
473#[serde(rename_all = "kebab-case")]
474#[non_exhaustive]
475pub enum OtelExporterType {
476 OtlpHttp,
479 File,
482}
483
484impl OtelExporterType {
485 pub fn as_str(self) -> &'static str {
487 match self {
488 Self::OtlpHttp => "otlp-http",
489 Self::File => "file",
490 }
491 }
492}
493
494#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
500#[non_exhaustive]
501pub enum OtlpHttpProtocol {
502 #[serde(rename = "http/json")]
504 HttpJson,
505 #[serde(rename = "http/protobuf")]
507 HttpProtobuf,
508}
509
510impl OtlpHttpProtocol {
511 pub fn as_str(self) -> &'static str {
513 match self {
514 Self::HttpJson => "http/json",
515 Self::HttpProtobuf => "http/protobuf",
516 }
517 }
518}
519
520#[derive(Debug, Clone, Default)]
555#[non_exhaustive]
556pub struct TelemetryConfig {
557 pub otlp_endpoint: Option<String>,
559 pub otlp_protocol: Option<OtlpHttpProtocol>,
561 pub file_path: Option<PathBuf>,
563 pub exporter_type: Option<OtelExporterType>,
566 pub source_name: Option<String>,
570 pub capture_content: Option<bool>,
574}
575
576impl TelemetryConfig {
577 pub fn new() -> Self {
580 Self::default()
581 }
582
583 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
585 self.otlp_endpoint = Some(endpoint.into());
586 self
587 }
588
589 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
591 self.otlp_protocol = Some(protocol);
592 self
593 }
594
595 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
597 self.file_path = Some(path.into());
598 self
599 }
600
601 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
603 self.exporter_type = Some(exporter_type);
604 self
605 }
606
607 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
611 self.source_name = Some(source_name.into());
612 self
613 }
614
615 pub fn with_capture_content(mut self, capture: bool) -> Self {
619 self.capture_content = Some(capture);
620 self
621 }
622
623 pub fn is_empty(&self) -> bool {
626 self.otlp_endpoint.is_none()
627 && self.otlp_protocol.is_none()
628 && self.file_path.is_none()
629 && self.exporter_type.is_none()
630 && self.source_name.is_none()
631 && self.capture_content.is_none()
632 }
633}
634
635impl Default for ClientOptions {
636 fn default() -> Self {
637 Self {
638 program: CliProgram::Resolve,
639 prefix_args: Vec::new(),
640 working_directory: PathBuf::new(),
641 env: Vec::new(),
642 env_remove: Vec::new(),
643 extra_args: Vec::new(),
644 builtin_plugin_directories: Vec::new(),
645 transport: Transport::default(),
646 github_token: None,
647 use_logged_in_user: None,
648 log_level: None,
649 session_idle_timeout_seconds: None,
650 on_list_models: None,
651 session_fs: None,
652 request_handler: None,
653 on_github_telemetry: None,
654 on_get_trace_context: None,
655 telemetry: None,
656 base_directory: None,
657 enable_remote_sessions: false,
658 bundled_cli_extract_dir: None,
659 mode: ClientMode::default(),
660 }
661 }
662}
663
664impl ClientOptions {
665 pub fn new() -> Self {
681 Self::default()
682 }
683
684 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
686 self.program = program.into();
687 self
688 }
689
690 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
692 where
693 I: IntoIterator<Item = S>,
694 S: Into<OsString>,
695 {
696 self.prefix_args = args.into_iter().map(Into::into).collect();
697 self
698 }
699
700 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
702 self.working_directory = cwd.into();
703 self
704 }
705
706 pub fn with_env<I, K, V>(mut self, env: I) -> Self
708 where
709 I: IntoIterator<Item = (K, V)>,
710 K: Into<OsString>,
711 V: Into<OsString>,
712 {
713 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
714 self
715 }
716
717 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
719 where
720 I: IntoIterator<Item = S>,
721 S: Into<OsString>,
722 {
723 self.env_remove = names.into_iter().map(Into::into).collect();
724 self
725 }
726
727 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
729 where
730 I: IntoIterator<Item = S>,
731 S: Into<String>,
732 {
733 self.extra_args = args.into_iter().map(Into::into).collect();
734 self
735 }
736
737 pub fn with_builtin_plugin_directories<I, P>(mut self, paths: I) -> Self
742 where
743 I: IntoIterator<Item = P>,
744 P: Into<PathBuf>,
745 {
746 self.builtin_plugin_directories = paths.into_iter().map(Into::into).collect();
747 self
748 }
749
750 pub fn with_transport(mut self, transport: Transport) -> Self {
752 self.transport = transport;
753 self
754 }
755
756 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
759 self.github_token = Some(token.into());
760 self
761 }
762
763 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
766 self.use_logged_in_user = Some(use_logged_in);
767 self
768 }
769
770 pub fn with_log_level(mut self, level: LogLevel) -> Self {
772 self.log_level = Some(level);
773 self
774 }
775
776 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
779 self.session_idle_timeout_seconds = Some(seconds);
780 self
781 }
782
783 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
786 where
787 H: ListModelsHandler + 'static,
788 {
789 self.on_list_models = Some(Arc::new(handler));
790 self
791 }
792
793 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
795 self.session_fs = Some(config);
796 self
797 }
798
799 pub fn with_request_handler<H>(mut self, handler: H) -> Self
804 where
805 H: crate::copilot_request_handler::CopilotRequestHandler,
806 {
807 self.request_handler = Some(Arc::new(handler));
808 self
809 }
810
811 #[doc(hidden)]
817 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
818 where
819 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
820 {
821 self.on_github_telemetry = Some(Arc::new(callback));
822 self
823 }
824
825 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
829 where
830 P: TraceContextProvider + 'static,
831 {
832 self.on_get_trace_context = Some(Arc::new(provider));
833 self
834 }
835
836 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
838 self.telemetry = Some(config);
839 self
840 }
841
842 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
845 self.base_directory = Some(dir.into());
846 self
847 }
848
849 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
852 self.enable_remote_sessions = enabled;
853 self
854 }
855
856 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
866 self.bundled_cli_extract_dir = Some(dir.into());
867 self
868 }
869
870 pub fn with_mode(mut self, mode: ClientMode) -> Self {
875 self.mode = mode;
876 self
877 }
878}
879
880fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
882 if cfg.initial_cwd.trim().is_empty() {
883 return Err(Error::with_message(
884 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
885 "invalid SessionFsConfig: initial_cwd must not be empty",
886 ));
887 }
888 if cfg.session_state_path.trim().is_empty() {
889 return Err(Error::with_message(
890 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
891 "invalid SessionFsConfig: session_state_path must not be empty",
892 ));
893 }
894 Ok(())
895}
896
897fn generate_connection_token() -> String {
904 let mut bytes = [0u8; 16];
905 getrandom::getrandom(&mut bytes)
906 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
907 let mut hex = String::with_capacity(32);
908 for byte in bytes {
909 use std::fmt::Write;
910 let _ = write!(hex, "{byte:02x}");
911 }
912 hex
913}
914
915const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
920
921fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
923 let configured = options
924 .env
925 .iter()
926 .find(|(key, _)| {
927 key.to_string_lossy()
928 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
929 })
930 .map(|(_, value)| value.to_string_lossy().into_owned());
931 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
932 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
933}
934
935fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
936 match value {
937 None => Ok(Transport::Stdio),
938 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
939 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
940 Some(v) => Err(Error::with_message(
941 ErrorKind::InvalidConfig,
942 format!(
943 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
944 Expected 'inprocess', 'stdio', or unset."
945 ),
946 )),
947 }
948}
949
950#[cfg(any(feature = "bundled-in-process", test))]
951fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
952 if !matches!(&options.program, CliProgram::Resolve) {
953 return Err(Error::with_message(
954 ErrorKind::InvalidConfig,
955 "ClientOptions::program is not supported with Transport::InProcess; \
956 set COPILOT_CLI_PATH only when using an externally provisioned runtime package",
957 ));
958 }
959 if !options.extra_args.is_empty() {
960 return Err(Error::with_message(
961 ErrorKind::InvalidConfig,
962 "ClientOptions::extra_args is not supported with Transport::InProcess; \
963 use typed client options instead",
964 ));
965 }
966
967 let unsupported = if !options.working_directory.as_os_str().is_empty() {
968 Some("working_directory")
969 } else if !options.env.is_empty() {
970 Some("env")
971 } else if !options.env_remove.is_empty() {
972 Some("env_remove")
973 } else if options.telemetry.is_some() {
974 Some("telemetry")
975 } else if !options.prefix_args.is_empty() {
976 Some("prefix_args")
977 } else {
978 None
979 };
980
981 if let Some(option) = unsupported {
982 return Err(Error::with_message(
983 ErrorKind::InvalidConfig,
984 format!(
985 "ClientOptions::{option} is not supported with Transport::InProcess; \
986 configure process-global settings on the host process instead"
987 ),
988 ));
989 }
990
991 Ok(())
992}
993
994#[derive(Clone)]
999pub struct Client {
1000 inner: Arc<ClientInner>,
1001}
1002
1003impl std::fmt::Debug for Client {
1004 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1005 f.debug_struct("Client")
1006 .field("working_directory", &self.inner.cwd)
1007 .field("pid", &self.pid())
1008 .finish()
1009 }
1010}
1011
1012struct ClientInner {
1013 child: parking_lot::Mutex<Option<Child>>,
1014 #[cfg(feature = "bundled-in-process")]
1015 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
1018 rpc: JsonRpcClient,
1019 cwd: PathBuf,
1020 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
1021 notification_tx: broadcast::Sender<JsonRpcNotification>,
1022 router: router::SessionRouter,
1023 negotiated_protocol_version: OnceLock<u32>,
1024 state: parking_lot::Mutex<ConnectionState>,
1025 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
1026 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1027 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
1028 session_fs_configured: bool,
1029 session_fs_sqlite_declared: bool,
1030 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
1033 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1038 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1039 effective_connection_token: Option<String>,
1044 pub(crate) mode: ClientMode,
1047 startup_timings: OnceLock<StartupTimings>,
1051}
1052
1053impl Client {
1054 pub async fn start(options: ClientOptions) -> Result<Self> {
1067 let start_time = Instant::now();
1068 let mut timings = StartupTimings::default();
1069 let mut options = options;
1070 if matches!(options.transport, Transport::Default) {
1071 options.transport = resolve_default_transport(&options)?;
1072 }
1073 if matches!(options.transport, Transport::InProcess) {
1074 #[cfg(not(feature = "bundled-in-process"))]
1075 {
1076 return Err(Error::with_message(
1077 ErrorKind::InvalidConfig,
1078 "Transport::InProcess requires the `bundled-in-process` Cargo feature",
1079 ));
1080 }
1081 #[cfg(feature = "bundled-in-process")]
1082 validate_inprocess_options(&options)?;
1083 }
1084 if options.mode == ClientMode::Empty
1085 && options.base_directory.is_none()
1086 && options.session_fs.is_none()
1087 {
1088 return Err(Error::with_message(
1089 ErrorKind::InvalidConfig,
1090 "ClientMode::Empty requires either `base_directory` or \
1091 `session_fs` to be set (no implicit ~/.copilot fallback).",
1092 ));
1093 }
1094 if let Some(cfg) = &options.session_fs {
1095 validate_session_fs_config(cfg)?;
1096 }
1097 let builtin_plugin_directories = options
1098 .builtin_plugin_directories
1099 .iter()
1100 .map(|path| {
1101 if !path.is_absolute() {
1102 return Err(Error::with_message(
1103 ErrorKind::InvalidConfig,
1104 format!(
1105 "builtin_plugin_directories must contain only absolute paths: {}",
1106 path.display()
1107 ),
1108 ));
1109 }
1110 path.to_str().map(str::to_owned).ok_or_else(|| {
1111 Error::with_message(
1112 ErrorKind::InvalidConfig,
1113 format!(
1114 "builtin_plugin_directories must contain valid UTF-8 paths: {}",
1115 path.display()
1116 ),
1117 )
1118 })
1119 })
1120 .collect::<Result<Vec<_>>>()?;
1121 if matches!(options.transport, Transport::External { .. }) {
1124 if options.github_token.is_some() {
1125 return Err(Error::with_message(
1126 ErrorKind::InvalidConfig,
1127 "invalid client configuration: github_token cannot be used with \
1128 Transport::External (external server manages its own auth)",
1129 ));
1130 }
1131 if options.use_logged_in_user == Some(true) {
1132 return Err(Error::with_message(
1133 ErrorKind::InvalidConfig,
1134 "invalid client configuration: use_logged_in_user cannot be used with \
1135 Transport::External (external server manages its own auth)",
1136 ));
1137 }
1138 }
1139 match &options.transport {
1143 Transport::Tcp {
1144 connection_token: Some(t),
1145 ..
1146 }
1147 | Transport::External {
1148 connection_token: Some(t),
1149 ..
1150 } if t.is_empty() => {
1151 return Err(Error::with_message(
1152 ErrorKind::InvalidConfig,
1153 "invalid client configuration: connection_token must be a non-empty string",
1154 ));
1155 }
1156 _ => {}
1157 }
1158 let effective_connection_token: Option<String> = match &mut options.transport {
1163 Transport::Default => unreachable!("default transport resolved above"),
1164 Transport::Stdio | Transport::InProcess => None,
1165 Transport::Tcp {
1166 connection_token, ..
1167 } => Some(
1168 connection_token
1169 .get_or_insert_with(generate_connection_token)
1170 .clone(),
1171 ),
1172 Transport::External {
1173 connection_token, ..
1174 } => connection_token.clone(),
1175 };
1176 let session_fs_config = options.session_fs.clone();
1177 let request_handler = options.request_handler.clone();
1178 let session_fs_sqlite_declared = session_fs_config
1179 .as_ref()
1180 .and_then(|c| c.capabilities.as_ref())
1181 .is_some_and(|caps| caps.sqlite);
1182 let program = match &options.program {
1183 CliProgram::Path(path) => {
1184 info!(path = %path.display(), "using explicit copilot CLI path");
1185 path.clone()
1186 }
1187 CliProgram::Resolve => {
1188 let resolve_start = Instant::now();
1189 let resolved = resolve::copilot_binary_with_extract_dir(
1190 options.bundled_cli_extract_dir.as_deref(),
1191 )?;
1192 let resolve_elapsed = resolve_start.elapsed();
1193 timings.program_resolve_ms = Some(StartupTimings::millis(resolve_elapsed));
1194 debug!(
1195 elapsed_ms = resolve_elapsed.as_millis(),
1196 "Client::start CLI program resolution complete"
1197 );
1198 info!(path = %resolved.display(), "resolved copilot CLI");
1199 #[cfg(windows)]
1200 {
1201 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1202 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1203 }) {
1204 warn!(
1205 path = %resolved.display(),
1206 ext = %ext,
1207 "resolved copilot CLI is a .cmd/.bat wrapper; \
1208 this may cause console window flashes on Windows"
1209 );
1210 }
1211 }
1212 resolved
1213 }
1214 };
1215 let working_directory = {
1216 let cwd = options.working_directory.clone();
1217 if cwd.as_os_str().is_empty() {
1218 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1219 } else {
1220 cwd
1221 }
1222 };
1223
1224 let transport_setup_start = Instant::now();
1225 let client = match options.transport {
1226 Transport::Default => unreachable!("default transport resolved above"),
1227 Transport::External {
1228 ref host,
1229 port,
1230 connection_token: _,
1231 } => {
1232 info!(host = %host, port = %port, "connecting to external CLI server");
1233 let connect_start = Instant::now();
1234 let stream = TcpStream::connect((host.as_str(), port)).await?;
1235 debug!(
1236 elapsed_ms = connect_start.elapsed().as_millis(),
1237 host = %host,
1238 port,
1239 "Client::start TCP connect complete"
1240 );
1241 let (reader, writer) = tokio::io::split(stream);
1242 Self::from_transport(
1243 reader,
1244 writer,
1245 None,
1246 working_directory,
1247 options.on_list_models,
1248 session_fs_config.is_some(),
1249 session_fs_sqlite_declared,
1250 options.on_get_trace_context,
1251 options.on_github_telemetry,
1252 effective_connection_token.clone(),
1253 options.mode,
1254 )?
1255 }
1256 Transport::Tcp {
1257 port,
1258 connection_token: _,
1259 } => {
1260 let (mut child, actual_port, spawn_elapsed, port_wait_elapsed) =
1261 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1262 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1263 timings.port_wait_ms = Some(StartupTimings::millis(port_wait_elapsed));
1264 let connect_start = Instant::now();
1265 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1266 debug!(
1267 elapsed_ms = connect_start.elapsed().as_millis(),
1268 port = actual_port,
1269 "Client::start TCP connect complete"
1270 );
1271 let (reader, writer) = tokio::io::split(stream);
1272 Self::drain_stderr(&mut child);
1273 Self::from_transport(
1274 reader,
1275 writer,
1276 Some(child),
1277 working_directory,
1278 options.on_list_models,
1279 session_fs_config.is_some(),
1280 session_fs_sqlite_declared,
1281 options.on_get_trace_context,
1282 options.on_github_telemetry,
1283 effective_connection_token.clone(),
1284 options.mode,
1285 )?
1286 }
1287 Transport::Stdio => {
1288 let (mut child, spawn_elapsed) =
1289 Self::spawn_stdio(&program, &options, &working_directory)?;
1290 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1291 let stdin = child.stdin.take().expect("stdin is piped");
1292 let stdout = child.stdout.take().expect("stdout is piped");
1293 Self::drain_stderr(&mut child);
1294 Self::from_transport(
1295 stdout,
1296 stdin,
1297 Some(child),
1298 working_directory,
1299 options.on_list_models,
1300 session_fs_config.is_some(),
1301 session_fs_sqlite_declared,
1302 options.on_get_trace_context,
1303 options.on_github_telemetry,
1304 effective_connection_token.clone(),
1305 options.mode,
1306 )?
1307 }
1308 Transport::InProcess => {
1309 #[cfg(feature = "bundled-in-process")]
1310 {
1311 info!(runtime_path = %program.display(), "hosting copilot runtime in-process (FFI)");
1312 let mut environment = Vec::new();
1313 if let Some(base_directory) = &options.base_directory {
1314 let value = base_directory.to_str().ok_or_else(|| {
1315 Error::with_message(
1316 ErrorKind::InvalidConfig,
1317 "base_directory must be valid UTF-8 for Transport::InProcess",
1318 )
1319 })?;
1320 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1321 }
1322 if options.mode == ClientMode::Empty {
1323 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1324 }
1325 if let Some(github_token) = &options.github_token {
1326 environment
1327 .push(("COPILOT_SDK_AUTH_TOKEN".to_string(), github_token.clone()));
1328 }
1329 let mut args = Vec::new();
1330 args.extend(
1331 Self::log_level_args(&options)
1332 .into_iter()
1333 .map(str::to_string),
1334 );
1335 args.extend(Self::session_idle_timeout_args(&options));
1336 args.extend(Self::remote_args(&options));
1337 if options.github_token.is_some() {
1338 args.extend([
1339 "--auth-token-env".to_string(),
1340 "COPILOT_SDK_AUTH_TOKEN".to_string(),
1341 ]);
1342 }
1343 let use_logged_in_user = options
1344 .use_logged_in_user
1345 .unwrap_or(options.github_token.is_none());
1346 if !use_logged_in_user {
1347 args.push("--no-auto-login".to_string());
1348 }
1349 let host = crate::ffi::FfiHost::create(&program, environment, args)?;
1350 let (reader, writer, shared) = host.start().await?;
1351 let client = Self::from_transport(
1352 reader,
1353 writer,
1354 None,
1355 working_directory,
1356 options.on_list_models,
1357 session_fs_config.is_some(),
1358 session_fs_sqlite_declared,
1359 options.on_get_trace_context,
1360 options.on_github_telemetry,
1361 effective_connection_token.clone(),
1362 options.mode,
1363 )?;
1364 *client.inner.ffi_host.lock() = Some(shared);
1365 client
1366 }
1367 #[cfg(not(feature = "bundled-in-process"))]
1368 unreachable!("in-process feature validation returned above")
1369 }
1370 };
1371 timings.transport_setup_ms = StartupTimings::millis(transport_setup_start.elapsed());
1372 debug!(
1373 elapsed_ms = start_time.elapsed().as_millis(),
1374 "Client::start transport setup complete"
1375 );
1376 let handshake_start = Instant::now();
1377 client.verify_protocol_version().await?;
1378 timings.handshake_ms = StartupTimings::millis(handshake_start.elapsed());
1379 debug!(
1380 elapsed_ms = start_time.elapsed().as_millis(),
1381 "Client::start protocol verification complete"
1382 );
1383 if !builtin_plugin_directories.is_empty() {
1384 client
1385 .call(
1386 "plugins.builtin.set",
1387 Some(serde_json::json!({ "paths": builtin_plugin_directories })),
1388 )
1389 .await?;
1390 }
1391 if let Some(cfg) = session_fs_config {
1392 let session_fs_start = Instant::now();
1393 let capabilities = cfg.capabilities.as_ref().map(|c| {
1394 crate::generated::api_types::SessionFsSetProviderCapabilities {
1395 sqlite: Some(c.sqlite),
1396 }
1397 });
1398 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1399 capabilities,
1400 conventions: cfg.conventions.into_wire(),
1401 initial_cwd: cfg.initial_cwd,
1402 session_state_path: cfg.session_state_path,
1403 };
1404 client.rpc().session_fs().set_provider(request).await?;
1405 let session_fs_elapsed = session_fs_start.elapsed();
1406 timings.session_fs_ms = Some(StartupTimings::millis(session_fs_elapsed));
1407 debug!(
1408 elapsed_ms = session_fs_elapsed.as_millis(),
1409 "Client::start session filesystem setup complete"
1410 );
1411 }
1412 if let Some(handler) = request_handler {
1413 let llm_inference_start = Instant::now();
1414 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1415 handler,
1416 ));
1417 dispatcher.set_client(Arc::downgrade(&client.inner));
1418 let _ = client.inner.llm_inference.set(dispatcher.clone());
1419 client.inner.router.ensure_started(
1422 &client.inner.notification_tx,
1423 &client.inner.request_rx,
1424 Some(dispatcher.clone()),
1425 client.inner.on_github_telemetry.clone(),
1426 );
1427 client.rpc().llm_inference().set_provider().await?;
1428 let llm_inference_elapsed = llm_inference_start.elapsed();
1429 timings.llm_handler_ms = Some(StartupTimings::millis(llm_inference_elapsed));
1430 debug!(
1431 elapsed_ms = llm_inference_elapsed.as_millis(),
1432 "Client::start Copilot request handler registration complete"
1433 );
1434 }
1435 timings.total_ms = StartupTimings::millis(start_time.elapsed());
1436 let timings_span = tracing::debug_span!(
1439 "Client::start timings",
1440 program_resolve_ms = tracing::field::Empty,
1441 process_spawn_ms = tracing::field::Empty,
1442 port_wait_ms = tracing::field::Empty,
1443 transport_setup_ms = timings.transport_setup_ms,
1444 handshake_ms = timings.handshake_ms,
1445 session_fs_ms = tracing::field::Empty,
1446 llm_handler_ms = tracing::field::Empty,
1447 total_ms = timings.total_ms,
1448 );
1449 record_optional_millis(
1450 &timings_span,
1451 "program_resolve_ms",
1452 timings.program_resolve_ms,
1453 );
1454 record_optional_millis(&timings_span, "process_spawn_ms", timings.process_spawn_ms);
1455 record_optional_millis(&timings_span, "port_wait_ms", timings.port_wait_ms);
1456 record_optional_millis(&timings_span, "session_fs_ms", timings.session_fs_ms);
1457 record_optional_millis(&timings_span, "llm_handler_ms", timings.llm_handler_ms);
1458 timings_span.in_scope(|| debug!("Client::start timings"));
1459 let _ = client.inner.startup_timings.set(timings);
1460 debug!(
1461 elapsed_ms = start_time.elapsed().as_millis(),
1462 "Client::start complete"
1463 );
1464 Ok(client)
1465 }
1466
1467 pub fn from_streams(
1471 reader: impl AsyncRead + Unpin + Send + 'static,
1472 writer: impl AsyncWrite + Unpin + Send + 'static,
1473 cwd: PathBuf,
1474 ) -> Result<Self> {
1475 Self::from_transport(
1476 reader,
1477 writer,
1478 None,
1479 cwd,
1480 None,
1481 false,
1482 false,
1483 None,
1484 None,
1485 None,
1486 ClientMode::default(),
1487 )
1488 }
1489
1490 #[cfg(any(test, feature = "test-support"))]
1498 pub fn from_streams_with_trace_provider(
1499 reader: impl AsyncRead + Unpin + Send + 'static,
1500 writer: impl AsyncWrite + Unpin + Send + 'static,
1501 cwd: PathBuf,
1502 provider: Arc<dyn TraceContextProvider>,
1503 ) -> Result<Self> {
1504 Self::from_transport(
1505 reader,
1506 writer,
1507 None,
1508 cwd,
1509 None,
1510 false,
1511 false,
1512 Some(provider),
1513 None,
1514 None,
1515 ClientMode::default(),
1516 )
1517 }
1518
1519 #[cfg(any(test, feature = "test-support"))]
1523 pub fn from_streams_with_connection_token(
1524 reader: impl AsyncRead + Unpin + Send + 'static,
1525 writer: impl AsyncWrite + Unpin + Send + 'static,
1526 cwd: PathBuf,
1527 token: Option<String>,
1528 ) -> Result<Self> {
1529 Self::from_transport(
1530 reader,
1531 writer,
1532 None,
1533 cwd,
1534 None,
1535 false,
1536 false,
1537 None,
1538 None,
1539 token,
1540 ClientMode::default(),
1541 )
1542 }
1543
1544 #[doc(hidden)]
1547 #[cfg(any(test, feature = "test-support"))]
1548 pub fn from_streams_with_github_telemetry(
1549 reader: impl AsyncRead + Unpin + Send + 'static,
1550 writer: impl AsyncWrite + Unpin + Send + 'static,
1551 cwd: PathBuf,
1552 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1553 ) -> Result<Self> {
1554 Self::from_transport(
1555 reader,
1556 writer,
1557 None,
1558 cwd,
1559 None,
1560 false,
1561 false,
1562 None,
1563 Some(on_github_telemetry),
1564 None,
1565 ClientMode::default(),
1566 )
1567 }
1568
1569 #[cfg(any(test, feature = "test-support"))]
1575 pub fn generate_connection_token_for_test() -> String {
1576 generate_connection_token()
1577 }
1578
1579 #[allow(clippy::too_many_arguments)]
1580 fn from_transport(
1581 reader: impl AsyncRead + Unpin + Send + 'static,
1582 writer: impl AsyncWrite + Unpin + Send + 'static,
1583 child: Option<Child>,
1584 cwd: PathBuf,
1585 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1586 session_fs_configured: bool,
1587 session_fs_sqlite_declared: bool,
1588 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1589 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1590 effective_connection_token: Option<String>,
1591 mode: ClientMode,
1592 ) -> Result<Self> {
1593 let setup_start = Instant::now();
1594 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1595 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1596 let rpc = JsonRpcClient::new(
1597 writer,
1598 reader,
1599 notification_broadcast_tx.clone(),
1600 request_tx,
1601 );
1602
1603 let pid = child.as_ref().and_then(|c| c.id());
1604 info!(pid = ?pid, "copilot CLI client ready");
1605
1606 let client = Self {
1607 inner: Arc::new(ClientInner {
1608 child: parking_lot::Mutex::new(child),
1609 #[cfg(feature = "bundled-in-process")]
1610 ffi_host: parking_lot::Mutex::new(None),
1611 rpc,
1612 cwd,
1613 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1614 notification_tx: notification_broadcast_tx,
1615 router: router::SessionRouter::new(),
1616 negotiated_protocol_version: OnceLock::new(),
1617 state: parking_lot::Mutex::new(ConnectionState::Connected),
1618 lifecycle_tx: broadcast::channel(256).0,
1619 on_list_models,
1620 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1621 session_fs_configured,
1622 session_fs_sqlite_declared,
1623 llm_inference: OnceLock::new(),
1624 on_github_telemetry,
1625 on_get_trace_context,
1626 effective_connection_token,
1627 mode,
1628 startup_timings: OnceLock::new(),
1629 }),
1630 };
1631 client.spawn_lifecycle_dispatcher();
1632 debug!(
1633 elapsed_ms = setup_start.elapsed().as_millis(),
1634 pid = ?pid,
1635 "Client::from_transport setup complete"
1636 );
1637 Ok(client)
1638 }
1639
1640 fn spawn_lifecycle_dispatcher(&self) {
1644 let mut notif_rx = self.inner.notification_tx.subscribe();
1645 let lifecycle_tx = self.inner.lifecycle_tx.clone();
1646 tokio::spawn(async move {
1647 loop {
1648 match notif_rx.recv().await {
1649 Ok(notification) => {
1650 if notification.method != "session.lifecycle" {
1651 continue;
1652 }
1653 let Some(params) = notification.params.as_ref() else {
1654 continue;
1655 };
1656 let event: SessionLifecycleEvent =
1657 match serde_json::from_value(params.clone()) {
1658 Ok(e) => e,
1659 Err(e) => {
1660 warn!(
1661 error = %e,
1662 "failed to deserialize session.lifecycle notification"
1663 );
1664 continue;
1665 }
1666 };
1667 let _ = lifecycle_tx.send(event);
1670 }
1671 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
1672 warn!(missed = n, "lifecycle dispatcher lagged");
1673 }
1674 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
1675 }
1676 }
1677 });
1678 }
1679
1680 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
1681 let mut command = Command::new(program);
1682 command.kill_on_drop(true);
1683 for arg in &options.prefix_args {
1684 command.arg(arg);
1685 }
1686 if let Some(token) = &options.github_token {
1689 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1690 }
1691 if let Some(telemetry) = &options.telemetry {
1694 command.env("COPILOT_OTEL_ENABLED", "true");
1695 if let Some(endpoint) = &telemetry.otlp_endpoint {
1696 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1697 }
1698 if let Some(protocol) = telemetry.otlp_protocol {
1699 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
1700 }
1701 if let Some(path) = &telemetry.file_path {
1702 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
1703 }
1704 if let Some(exporter) = telemetry.exporter_type {
1705 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
1706 }
1707 if let Some(source) = &telemetry.source_name {
1708 command.env("COPILOT_OTEL_SOURCE_NAME", source);
1709 }
1710 if let Some(capture) = telemetry.capture_content {
1711 command.env(
1712 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
1713 if capture { "true" } else { "false" },
1714 );
1715 }
1716 }
1717 if let Some(dir) = &options.base_directory {
1718 command.env("COPILOT_HOME", dir);
1719 }
1720 if options.mode == ClientMode::Empty {
1723 command.env("COPILOT_DISABLE_KEYTAR", "1");
1724 }
1725 if let Transport::Tcp {
1726 connection_token: Some(token),
1727 ..
1728 } = &options.transport
1729 {
1730 command.env("COPILOT_CONNECTION_TOKEN", token);
1731 }
1732 for (key, value) in &options.env {
1733 command.env(key, value);
1734 }
1735 for key in &options.env_remove {
1736 command.env_remove(key);
1737 }
1738 command
1739 .current_dir(working_directory)
1740 .stdout(Stdio::piped())
1741 .stderr(Stdio::piped());
1742
1743 #[cfg(windows)]
1744 {
1745 use std::os::windows::process::CommandExt;
1746 const CREATE_NO_WINDOW: u32 = 0x08000000;
1747 command.as_std_mut().creation_flags(CREATE_NO_WINDOW);
1748 }
1749
1750 command
1751 }
1752
1753 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
1761 let mut args: Vec<&'static str> = Vec::new();
1762 if options.github_token.is_some() {
1763 args.push("--auth-token-env");
1764 args.push("COPILOT_SDK_AUTH_TOKEN");
1765 }
1766 let use_logged_in = options
1767 .use_logged_in_user
1768 .unwrap_or(options.github_token.is_none());
1769 if !use_logged_in {
1770 args.push("--no-auto-login");
1771 }
1772 args
1773 }
1774
1775 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
1779 match options.session_idle_timeout_seconds {
1780 Some(secs) if secs > 0 => {
1781 vec!["--session-idle-timeout".to_string(), secs.to_string()]
1782 }
1783 _ => Vec::new(),
1784 }
1785 }
1786
1787 fn remote_args(options: &ClientOptions) -> Vec<String> {
1788 if options.enable_remote_sessions {
1789 vec!["--remote".to_string()]
1790 } else {
1791 Vec::new()
1792 }
1793 }
1794
1795 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
1796 match options.log_level {
1797 Some(level) => vec!["--log-level", level.as_str()],
1798 None => Vec::new(),
1799 }
1800 }
1801
1802 fn spawn_stdio(
1803 program: &Path,
1804 options: &ClientOptions,
1805 working_directory: &Path,
1806 ) -> Result<(Child, Duration)> {
1807 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
1808 let mut command = Self::build_command(program, options, working_directory);
1809 command
1810 .args(["--server", "--stdio", "--no-auto-update"])
1811 .args(Self::log_level_args(options))
1812 .args(Self::auth_args(options))
1813 .args(Self::session_idle_timeout_args(options))
1814 .args(Self::remote_args(options))
1815 .args(&options.extra_args)
1816 .stdin(Stdio::piped());
1817 let spawn_start = Instant::now();
1818 let child = command.spawn()?;
1819 let spawn_elapsed = spawn_start.elapsed();
1820 debug!(
1821 elapsed_ms = spawn_elapsed.as_millis(),
1822 "Client::spawn_stdio subprocess spawned"
1823 );
1824 Ok((child, spawn_elapsed))
1825 }
1826
1827 async fn spawn_tcp(
1828 program: &Path,
1829 options: &ClientOptions,
1830 working_directory: &Path,
1831 port: u16,
1832 ) -> Result<(Child, u16, Duration, Duration)> {
1833 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
1834 let mut command = Self::build_command(program, options, working_directory);
1835 command
1836 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
1837 .args(Self::log_level_args(options))
1838 .args(Self::auth_args(options))
1839 .args(Self::session_idle_timeout_args(options))
1840 .args(Self::remote_args(options))
1841 .args(&options.extra_args)
1842 .stdin(Stdio::null());
1843 let spawn_start = Instant::now();
1844 let mut child = command.spawn()?;
1845 let spawn_elapsed = spawn_start.elapsed();
1846 debug!(
1847 elapsed_ms = spawn_elapsed.as_millis(),
1848 "Client::spawn_tcp subprocess spawned"
1849 );
1850 let stdout = child.stdout.take().expect("stdout is piped");
1851
1852 let (port_tx, port_rx) = oneshot::channel::<u16>();
1853 let span = tracing::error_span!("copilot_cli_port_scan");
1854 tokio::spawn(
1855 async move {
1856 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
1858 let mut lines = BufReader::new(stdout).lines();
1859 let mut port_tx = Some(port_tx);
1860 while let Ok(Some(line)) = lines.next_line().await {
1861 debug!(line = %line, "CLI stdout");
1862 if let Some(tx) = port_tx.take() {
1863 if let Some(caps) = port_re.captures(&line)
1864 && let Some(p) =
1865 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
1866 {
1867 let _ = tx.send(p);
1868 continue;
1869 }
1870 port_tx = Some(tx);
1872 }
1873 }
1874 }
1875 .instrument(span),
1876 );
1877
1878 let port_wait_start = Instant::now();
1879 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
1880 .await
1881 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
1882 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
1883
1884 let port_wait_elapsed = port_wait_start.elapsed();
1885 debug!(
1886 elapsed_ms = port_wait_elapsed.as_millis(),
1887 port = actual_port,
1888 "Client::spawn_tcp TCP port wait complete"
1889 );
1890 info!(port = %actual_port, "CLI server listening");
1891 Ok((child, actual_port, spawn_elapsed, port_wait_elapsed))
1892 }
1893
1894 fn drain_stderr(child: &mut Child) {
1895 if let Some(stderr) = child.stderr.take() {
1896 let span = tracing::error_span!("copilot_cli");
1897 tokio::spawn(
1898 async move {
1899 let mut reader = BufReader::new(stderr).lines();
1900 while let Ok(Some(line)) = reader.next_line().await {
1901 warn!(line = %line, "CLI stderr");
1902 }
1903 }
1904 .instrument(span),
1905 );
1906 }
1907 }
1908
1909 pub fn cwd(&self) -> &PathBuf {
1911 &self.inner.cwd
1912 }
1913
1914 pub fn mode(&self) -> ClientMode {
1916 self.inner.mode
1917 }
1918
1919 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
1930 crate::generated::rpc::ClientRpc { client: self }
1931 }
1932
1933 #[allow(dead_code, reason = "convenience for future internal use")]
1935 pub(crate) async fn send_request(
1936 &self,
1937 method: &str,
1938 params: Option<serde_json::Value>,
1939 ) -> Result<JsonRpcResponse> {
1940 self.inner.rpc.send_request(method, params).await
1941 }
1942
1943 pub async fn call(
1963 &self,
1964 method: &str,
1965 params: Option<serde_json::Value>,
1966 ) -> Result<serde_json::Value> {
1967 self.call_with_inline_callback(method, params, None).await
1968 }
1969
1970 pub(crate) async fn call_with_inline_callback(
1985 &self,
1986 method: &str,
1987 params: Option<serde_json::Value>,
1988 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
1989 ) -> Result<serde_json::Value> {
1990 let session_id: Option<SessionId> = params
1991 .as_ref()
1992 .and_then(|p| p.get("sessionId"))
1993 .and_then(|v| v.as_str())
1994 .map(SessionId::from);
1995 let response = self
1996 .inner
1997 .rpc
1998 .send_request_with_inline_callback(method, params, inline_callback)
1999 .await?;
2000 if let Some(err) = response.error {
2001 if err.message.contains("Session not found") {
2002 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
2003 session_id.unwrap_or_else(|| "unknown".into()),
2004 ))
2005 .into());
2006 }
2007 return Err(Error::with_message(
2008 ErrorKind::Rpc { code: err.code },
2009 err.message,
2010 ));
2011 }
2012 Ok(response.result.unwrap_or(serde_json::Value::Null))
2013 }
2014
2015 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
2017 self.inner.rpc.write(response).await
2018 }
2019
2020 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
2022 Self { inner }
2023 }
2024
2025 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
2029 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
2030 self.inner.request_rx.lock().take()
2031 }
2032
2033 pub(crate) fn register_session(
2041 &self,
2042 session_id: &SessionId,
2043 ) -> crate::router::SessionChannels {
2044 self.inner.router.ensure_started(
2045 &self.inner.notification_tx,
2046 &self.inner.request_rx,
2047 self.inner.llm_inference.get().cloned(),
2048 self.inner.on_github_telemetry.clone(),
2049 );
2050 self.inner.router.register(session_id)
2051 }
2052
2053 pub(crate) fn unregister_session(&self, session_id: &SessionId) {
2055 self.inner.router.unregister(session_id);
2056 }
2057
2058 pub fn protocol_version(&self) -> Option<u32> {
2065 self.inner.negotiated_protocol_version.get().copied()
2066 }
2067
2068 pub fn startup_timings(&self) -> Option<StartupTimings> {
2075 self.inner.startup_timings.get().cloned()
2076 }
2077
2078 pub async fn verify_protocol_version(&self) -> Result<()> {
2102 let handshake_start = Instant::now();
2103 let mut used_fallback_ping = false;
2104 let server_version = match self.connect_handshake().await {
2108 Ok(v) => v,
2109 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2110 used_fallback_ping = true;
2111 self.ping(None).await?.protocol_version
2112 }
2113 Err(e) => return Err(e),
2114 };
2115
2116 match server_version {
2117 None => {
2118 warn!("CLI server did not report protocolVersion; skipping version check");
2119 }
2120 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2121 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2122 server: v,
2123 min: MIN_PROTOCOL_VERSION,
2124 max: SDK_PROTOCOL_VERSION,
2125 })
2126 .into());
2127 }
2128 Some(v) => {
2129 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2130 if existing != v {
2131 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2132 previous: existing,
2133 current: v,
2134 })
2135 .into());
2136 }
2137 } else {
2138 let _ = self.inner.negotiated_protocol_version.set(v);
2139 }
2140 }
2141 }
2142
2143 debug!(
2144 elapsed_ms = handshake_start.elapsed().as_millis(),
2145 protocol_version = ?server_version,
2146 used_fallback_ping,
2147 "Client::verify_protocol_version protocol handshake complete"
2148 );
2149 Ok(())
2150 }
2151
2152 async fn connect_handshake(&self) -> Result<Option<u32>> {
2159 let params = crate::generated::api_types::ConnectRequest {
2160 token: self.inner.effective_connection_token.clone(),
2161 enable_git_hub_telemetry_forwarding: self
2162 .inner
2163 .on_github_telemetry
2164 .is_some()
2165 .then_some(true),
2166 ..Default::default()
2167 };
2168 let value = self
2169 .call(
2170 crate::generated::api_types::rpc_methods::CONNECT,
2171 Some(serde_json::to_value(params)?),
2172 )
2173 .await?;
2174 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2175 Ok(Some(u32::try_from(result.protocol_version).map_err(
2176 |_| ProtocolErrorKind::InvalidProtocolVersion {
2177 server: result.protocol_version,
2178 },
2179 )?))
2180 }
2181
2182 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2190 let params = match message {
2191 Some(m) => serde_json::json!({ "message": m }),
2192 None => serde_json::json!({}),
2193 };
2194 let value = self
2195 .call(generated::api_types::rpc_methods::PING, Some(params))
2196 .await?;
2197 Ok(serde_json::from_value(value)?)
2198 }
2199
2200 pub async fn list_sessions(
2203 &self,
2204 filter: Option<SessionListFilter>,
2205 ) -> Result<Vec<SessionMetadata>> {
2206 let params = match filter {
2207 Some(f) => serde_json::json!({ "filter": f }),
2208 None => serde_json::json!({}),
2209 };
2210 let result = self.call("session.list", Some(params)).await?;
2211 let response: ListSessionsResponse = serde_json::from_value(result)?;
2212 Ok(response.sessions)
2213 }
2214
2215 pub async fn get_session_metadata(
2233 &self,
2234 session_id: &SessionId,
2235 ) -> Result<Option<SessionMetadata>> {
2236 let result = self
2237 .call(
2238 "session.getMetadata",
2239 Some(serde_json::json!({ "sessionId": session_id })),
2240 )
2241 .await?;
2242 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2243 Ok(response.session)
2244 }
2245
2246 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2248 self.call(
2249 "session.delete",
2250 Some(serde_json::json!({ "sessionId": session_id })),
2251 )
2252 .await?;
2253 Ok(())
2254 }
2255
2256 #[cfg(feature = "test-support")]
2259 #[doc(hidden)]
2260 pub fn start_router_for_test(&self) {
2261 self.inner.router.ensure_started(
2262 &self.inner.notification_tx,
2263 &self.inner.request_rx,
2264 self.inner.llm_inference.get().cloned(),
2265 self.inner.on_github_telemetry.clone(),
2266 );
2267 }
2268
2269 #[cfg(feature = "test-support")]
2270 #[doc(hidden)]
2271 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2274 let mut first_error = None;
2275
2276 for session_id in self.inner.router.session_ids() {
2277 if let Err(error) = self
2278 .call(
2279 "session.destroy",
2280 Some(serde_json::json!({ "sessionId": session_id })),
2281 )
2282 .await
2283 && first_error.is_none()
2284 {
2285 first_error = Some(error);
2286 }
2287 self.inner.router.unregister(&session_id);
2288 }
2289
2290 match self.list_sessions(None).await {
2291 Ok(sessions) => {
2292 for session in sessions {
2293 if let Err(error) = self.delete_session(&session.session_id).await
2294 && first_error.is_none()
2295 {
2296 first_error = Some(error);
2297 }
2298 }
2299 }
2300 Err(error) if first_error.is_none() => first_error = Some(error),
2301 Err(_) => {}
2302 }
2303
2304 match first_error {
2305 Some(error) => Err(error),
2306 None => Ok(()),
2307 }
2308 }
2309
2310 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2326 let result = self
2327 .call("session.getLastId", Some(serde_json::json!({})))
2328 .await?;
2329 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2330 Ok(response.session_id)
2331 }
2332
2333 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2338 let result = self
2339 .call("session.getForeground", Some(serde_json::json!({})))
2340 .await?;
2341 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2342 Ok(response.session_id)
2343 }
2344
2345 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2350 self.call(
2351 "session.setForeground",
2352 Some(serde_json::json!({ "sessionId": session_id })),
2353 )
2354 .await?;
2355 Ok(())
2356 }
2357
2358 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2360 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2361 Ok(serde_json::from_value(result)?)
2362 }
2363
2364 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2366 let result = self
2367 .call("auth.getStatus", Some(serde_json::json!({})))
2368 .await?;
2369 Ok(serde_json::from_value(result)?)
2370 }
2371
2372 pub async fn list_models(&self) -> Result<Vec<Model>> {
2377 let cache = self.inner.models_cache.lock().clone();
2378 let models = cache
2379 .get_or_try_init(|| async {
2380 if let Some(handler) = &self.inner.on_list_models {
2381 handler.list_models().await
2382 } else {
2383 Ok(self.rpc().models().list().await?.models)
2384 }
2385 })
2386 .await?;
2387 Ok(models.clone())
2388 }
2389
2390 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2393 if let Some(provider) = &self.inner.on_get_trace_context {
2394 provider.get_trace_context().await
2395 } else {
2396 TraceContext::default()
2397 }
2398 }
2399
2400 pub fn pid(&self) -> Option<u32> {
2402 self.inner.child.lock().as_ref().and_then(|c| c.id())
2403 }
2404
2405 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2432 let pid = self.pid();
2433 info!(pid = ?pid, "stopping CLI process");
2434 let mut errors: Vec<Error> = Vec::new();
2435
2436 for session_id in self.inner.router.session_ids() {
2439 match self
2440 .call(
2441 "session.destroy",
2442 Some(serde_json::json!({ "sessionId": session_id })),
2443 )
2444 .await
2445 {
2446 Ok(_) => {}
2447 Err(e) => {
2448 warn!(
2449 session_id = %session_id,
2450 error = %e,
2451 "session.destroy failed during Client::stop",
2452 );
2453 errors.push(e);
2454 }
2455 }
2456 self.inner.router.unregister(&session_id);
2457 }
2458
2459 let should_shutdown_runtime = self.inner.child.lock().is_some();
2460 #[cfg(feature = "bundled-in-process")]
2461 let should_shutdown_runtime =
2462 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2463 if should_shutdown_runtime {
2464 let runtime_shutdown_start = Instant::now();
2465 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2466 .await
2467 {
2468 Ok(Ok(())) => {
2469 debug!(
2470 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2471 "Client::stop runtime shutdown complete"
2472 );
2473 }
2474 Ok(Err(e)) => {
2475 warn!(
2476 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2477 error = %e,
2478 "runtime.shutdown failed during Client::stop",
2479 );
2480 errors.push(e);
2481 }
2482 Err(_) => {
2483 let e = std::io::Error::new(
2484 std::io::ErrorKind::TimedOut,
2485 "runtime.shutdown timed out during Client::stop",
2486 );
2487 warn!(
2488 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2489 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2490 error = %e,
2491 "runtime.shutdown timed out during Client::stop",
2492 );
2493 errors.push(e.into());
2494 }
2495 }
2496 }
2497
2498 let child = self.inner.child.lock().take();
2499 *self.inner.state.lock() = ConnectionState::Disconnected;
2500 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2501 if let Some(mut child) = child {
2502 match child.try_wait() {
2503 Ok(Some(_status)) => {}
2504 Ok(None) => {
2505 if let Err(e) = child.kill().await {
2512 errors.push(e.into());
2513 }
2514 }
2515 Err(e) => errors.push(e.into()),
2516 }
2517 }
2518
2519 #[cfg(feature = "bundled-in-process")]
2522 {
2523 if let Some(host) = self.inner.ffi_host.lock().take() {
2524 self.inner.rpc.force_close();
2525 host.close();
2526 }
2527 }
2528
2529 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2530 if errors.is_empty() {
2531 Ok(())
2532 } else {
2533 Err(StopErrors(errors))
2534 }
2535 }
2536
2537 pub fn force_stop(&self) {
2567 let pid = self.pid();
2568 info!(pid = ?pid, "force-stopping CLI process");
2569 if let Some(mut child) = self.inner.child.lock().take()
2570 && let Err(e) = child.start_kill()
2571 {
2572 error!(pid = ?pid, error = %e, "failed to send kill signal");
2573 }
2574 self.inner.rpc.force_close();
2575 #[cfg(feature = "bundled-in-process")]
2576 {
2577 if let Some(host) = self.inner.ffi_host.lock().take() {
2578 host.close();
2579 }
2580 }
2581 self.inner.router.clear();
2584 *self.inner.state.lock() = ConnectionState::Disconnected;
2585 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2586 }
2587
2588 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2623 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2624 }
2625}
2626
2627impl Drop for ClientInner {
2628 fn drop(&mut self) {
2629 if let Some(ref mut child) = *self.child.lock() {
2630 let pid = child.id();
2631 if let Err(e) = child.start_kill() {
2632 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2633 } else {
2634 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2635 }
2636 }
2637 #[cfg(feature = "bundled-in-process")]
2638 {
2639 if let Some(host) = self.ffi_host.lock().take() {
2640 self.rpc.force_close();
2641 host.close();
2642 }
2643 }
2644 }
2645}
2646
2647#[cfg(test)]
2648mod tests {
2649 use super::*;
2650
2651 #[test]
2652 fn is_transport_failure_matches_request_cancelled() {
2653 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2654 assert!(err.is_transport_failure());
2655 }
2656
2657 #[test]
2658 fn is_transport_failure_matches_io_error() {
2659 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2660 assert!(err.is_transport_failure());
2661 }
2662
2663 #[test]
2664 fn is_transport_failure_rejects_rpc_error() {
2665 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2666 assert!(!err.is_transport_failure());
2667 }
2668
2669 #[test]
2670 fn is_transport_failure_rejects_session_error() {
2671 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2672 assert!(!err.is_transport_failure());
2673 }
2674
2675 #[test]
2676 fn client_options_builder_composes() {
2677 let opts = ClientOptions::new()
2678 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2679 .with_prefix_args(["node"])
2680 .with_cwd(PathBuf::from("/tmp"))
2681 .with_env([("KEY", "value")])
2682 .with_env_remove(["UNWANTED"])
2683 .with_extra_args(["--quiet"])
2684 .with_github_token("ghp_test")
2685 .with_use_logged_in_user(false)
2686 .with_log_level(LogLevel::Debug)
2687 .with_session_idle_timeout_seconds(120)
2688 .with_enable_remote_sessions(true);
2689 assert!(matches!(opts.program, CliProgram::Path(_)));
2690 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2691 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2692 assert_eq!(
2693 opts.env,
2694 vec![(
2695 std::ffi::OsString::from("KEY"),
2696 std::ffi::OsString::from("value")
2697 )]
2698 );
2699 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2700 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2701 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2702 assert_eq!(opts.use_logged_in_user, Some(false));
2703 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2704 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2705 assert!(opts.enable_remote_sessions);
2706 }
2707
2708 #[test]
2709 fn default_transport_values_resolve_without_process_state() {
2710 assert!(matches!(
2711 resolve_default_transport_value(None).unwrap(),
2712 Transport::Stdio
2713 ));
2714 assert!(matches!(
2715 resolve_default_transport_value(Some("stdio")).unwrap(),
2716 Transport::Stdio
2717 ));
2718 assert!(matches!(
2719 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2720 Transport::InProcess
2721 ));
2722 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2723 }
2724
2725 #[test]
2726 fn inprocess_rejects_process_scoped_options() {
2727 let invalid = [
2728 ClientOptions::new().with_cwd("."),
2729 ClientOptions::new().with_env([("KEY", "value")]),
2730 ClientOptions::new().with_env_remove(["KEY"]),
2731 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2732 ClientOptions::new().with_prefix_args(["index.js"]),
2733 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
2734 ClientOptions::new().with_extra_args(["--verbose"]),
2735 ];
2736
2737 for options in invalid {
2738 assert!(validate_inprocess_options(&options).is_err());
2739 }
2740 }
2741
2742 #[test]
2743 fn inprocess_allows_typed_runtime_options() {
2744 let options = ClientOptions::new()
2745 .with_base_directory("state")
2746 .with_log_level(LogLevel::Debug)
2747 .with_session_idle_timeout_seconds(10)
2748 .with_github_token("token")
2749 .with_use_logged_in_user(false)
2750 .with_enable_remote_sessions(true);
2751
2752 assert!(validate_inprocess_options(&options).is_ok());
2753 }
2754
2755 #[cfg(not(feature = "bundled-in-process"))]
2756 #[tokio::test]
2757 async fn inprocess_requires_cargo_feature() {
2758 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
2759 .await
2760 .unwrap_err();
2761
2762 assert!(error.to_string().contains("bundled-in-process"));
2763 }
2764
2765 #[test]
2766 fn is_transport_failure_rejects_other_protocol_errors() {
2767 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2768 assert!(!err.is_transport_failure());
2769 }
2770
2771 #[test]
2772 fn build_command_lets_env_remove_strip_injected_token() {
2773 let opts = ClientOptions {
2774 github_token: Some("secret".to_string()),
2775 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2776 ..Default::default()
2777 };
2778 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2779 let action = cmd
2781 .as_std()
2782 .get_envs()
2783 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2784 .map(|(_, v)| v);
2785 assert_eq!(
2786 action,
2787 Some(None),
2788 "env_remove should win over github_token"
2789 );
2790 }
2791
2792 #[test]
2793 fn build_command_lets_env_override_injected_token() {
2794 let opts = ClientOptions {
2795 github_token: Some("from-options".to_string()),
2796 env: vec![(
2797 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2798 std::ffi::OsString::from("from-env"),
2799 )],
2800 ..Default::default()
2801 };
2802 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2803 let value = cmd
2804 .as_std()
2805 .get_envs()
2806 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2807 .and_then(|(_, v)| v);
2808 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2809 }
2810
2811 #[test]
2812 fn build_command_injects_github_token_by_default() {
2813 let opts = ClientOptions {
2814 github_token: Some("just-the-token".to_string()),
2815 ..Default::default()
2816 };
2817 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2818 let value = cmd
2819 .as_std()
2820 .get_envs()
2821 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2822 .and_then(|(_, v)| v);
2823 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2824 }
2825
2826 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2827 cmd.as_std()
2828 .get_envs()
2829 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2830 .and_then(|(_, v)| v)
2831 }
2832
2833 #[test]
2834 fn telemetry_config_builder_composes() {
2835 let cfg = TelemetryConfig::new()
2836 .with_otlp_endpoint("http://collector:4318")
2837 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2838 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2839 .with_exporter_type(OtelExporterType::OtlpHttp)
2840 .with_source_name("my-app")
2841 .with_capture_content(true);
2842
2843 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
2844 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
2845 assert_eq!(
2846 cfg.file_path.as_deref(),
2847 Some(Path::new("/var/log/copilot.jsonl")),
2848 );
2849 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
2850 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
2851 assert_eq!(cfg.capture_content, Some(true));
2852 assert!(!cfg.is_empty());
2853 assert!(TelemetryConfig::new().is_empty());
2854 }
2855
2856 #[test]
2857 fn otlp_http_protocol_serde_matches_env_value() {
2858 for (protocol, wire) in [
2859 (OtlpHttpProtocol::HttpJson, "http/json"),
2860 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
2861 ] {
2862 assert_eq!(protocol.as_str(), wire);
2863
2864 let serialized = serde_json::to_string(&protocol).unwrap();
2865 assert_eq!(serialized, format!("\"{wire}\""));
2866
2867 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
2868 assert_eq!(deserialized, protocol);
2869 }
2870 }
2871
2872 #[test]
2873 fn build_command_sets_otel_env_when_telemetry_enabled() {
2874 let opts = ClientOptions {
2875 telemetry: Some(TelemetryConfig {
2876 otlp_endpoint: Some("http://collector:4318".to_string()),
2877 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
2878 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
2879 exporter_type: Some(OtelExporterType::OtlpHttp),
2880 source_name: Some("my-app".to_string()),
2881 capture_content: Some(true),
2882 }),
2883 ..Default::default()
2884 };
2885 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2886 assert_eq!(
2887 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2888 Some(std::ffi::OsStr::new("true")),
2889 );
2890 assert_eq!(
2891 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2892 Some(std::ffi::OsStr::new("http://collector:4318")),
2893 );
2894 assert_eq!(
2895 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
2896 Some(std::ffi::OsStr::new("http/protobuf")),
2897 );
2898 assert_eq!(
2899 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
2900 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
2901 );
2902 assert_eq!(
2903 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
2904 Some(std::ffi::OsStr::new("otlp-http")),
2905 );
2906 assert_eq!(
2907 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
2908 Some(std::ffi::OsStr::new("my-app")),
2909 );
2910 assert_eq!(
2911 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
2912 Some(std::ffi::OsStr::new("true")),
2913 );
2914 }
2915
2916 #[test]
2917 fn build_command_omits_otel_env_when_telemetry_none() {
2918 let opts = ClientOptions::default();
2919 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2920 for key in [
2921 "COPILOT_OTEL_ENABLED",
2922 "OTEL_EXPORTER_OTLP_ENDPOINT",
2923 "OTEL_EXPORTER_OTLP_PROTOCOL",
2924 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2925 "COPILOT_OTEL_EXPORTER_TYPE",
2926 "COPILOT_OTEL_SOURCE_NAME",
2927 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2928 ] {
2929 assert!(
2930 env_value(&cmd, key).is_none(),
2931 "expected {key} to be unset when telemetry is None",
2932 );
2933 }
2934 }
2935
2936 #[test]
2937 fn build_command_omits_unset_telemetry_fields() {
2938 let opts = ClientOptions {
2939 telemetry: Some(TelemetryConfig {
2940 otlp_endpoint: Some("http://collector:4318".to_string()),
2941 ..Default::default()
2942 }),
2943 ..Default::default()
2944 };
2945 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2946 assert_eq!(
2948 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2949 Some(std::ffi::OsStr::new("true")),
2950 );
2951 assert_eq!(
2952 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2953 Some(std::ffi::OsStr::new("http://collector:4318")),
2954 );
2955 for key in [
2957 "OTEL_EXPORTER_OTLP_PROTOCOL",
2958 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2959 "COPILOT_OTEL_EXPORTER_TYPE",
2960 "COPILOT_OTEL_SOURCE_NAME",
2961 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2962 ] {
2963 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
2964 }
2965 }
2966
2967 #[test]
2968 fn build_command_lets_user_env_override_telemetry() {
2969 let opts = ClientOptions {
2970 telemetry: Some(TelemetryConfig {
2971 otlp_endpoint: Some("http://from-config:4318".to_string()),
2972 ..Default::default()
2973 }),
2974 env: vec![(
2975 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
2976 std::ffi::OsString::from("http://from-user-env:4318"),
2977 )],
2978 ..Default::default()
2979 };
2980 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2981 assert_eq!(
2982 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2983 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
2984 "user-supplied options.env should override telemetry config",
2985 );
2986 }
2987
2988 #[test]
2989 fn build_command_sets_copilot_home_env_when_configured() {
2990 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
2991 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2992 assert_eq!(
2993 env_value(&cmd, "COPILOT_HOME"),
2994 Some(std::ffi::OsStr::new("/custom/copilot")),
2995 );
2996
2997 let opts = ClientOptions::default();
2998 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2999 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
3000 }
3001
3002 #[test]
3003 fn build_command_sets_connection_token_env_when_configured() {
3004 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3005 port: 0,
3006 connection_token: Some("secret-token".to_string()),
3007 });
3008 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3009 assert_eq!(
3010 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3011 Some(std::ffi::OsStr::new("secret-token")),
3012 );
3013
3014 let opts = ClientOptions::default();
3015 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3016 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3017 }
3018
3019 #[tokio::test]
3020 async fn start_rejects_empty_connection_token() {
3021 let opts = ClientOptions::new()
3022 .with_transport(Transport::Tcp {
3023 port: 0,
3024 connection_token: Some(String::new()),
3025 })
3026 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3027 let err = Client::start(opts).await.unwrap_err();
3028 assert!(
3029 matches!(err.kind(), ErrorKind::InvalidConfig),
3030 "got {err:?}"
3031 );
3032 }
3033
3034 #[tokio::test]
3035 async fn start_rejects_empty_external_connection_token() {
3036 let opts = ClientOptions::new()
3037 .with_transport(Transport::External {
3038 host: "127.0.0.1".to_string(),
3039 port: 1,
3040 connection_token: Some(String::new()),
3041 })
3042 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3043 let err = Client::start(opts).await.unwrap_err();
3044 assert!(
3045 matches!(err.kind(), ErrorKind::InvalidConfig),
3046 "got {err:?}"
3047 );
3048 }
3049
3050 #[test]
3051 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3052 let opts_true = ClientOptions {
3053 telemetry: Some(TelemetryConfig {
3054 capture_content: Some(true),
3055 ..Default::default()
3056 }),
3057 ..Default::default()
3058 };
3059 let opts_false = ClientOptions {
3060 telemetry: Some(TelemetryConfig {
3061 capture_content: Some(false),
3062 ..Default::default()
3063 }),
3064 ..Default::default()
3065 };
3066 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3067 let cmd_false =
3068 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3069 assert_eq!(
3070 env_value(
3071 &cmd_true,
3072 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3073 ),
3074 Some(std::ffi::OsStr::new("true")),
3075 );
3076 assert_eq!(
3077 env_value(
3078 &cmd_false,
3079 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3080 ),
3081 Some(std::ffi::OsStr::new("false")),
3082 );
3083 }
3084
3085 #[test]
3086 fn session_idle_timeout_args_are_omitted_by_default() {
3087 let opts = ClientOptions::default();
3088 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3089 }
3090
3091 #[test]
3092 fn session_idle_timeout_args_omitted_for_zero() {
3093 let opts = ClientOptions {
3094 session_idle_timeout_seconds: Some(0),
3095 ..Default::default()
3096 };
3097 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3098 }
3099
3100 #[test]
3101 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3102 let opts = ClientOptions {
3103 session_idle_timeout_seconds: Some(300),
3104 ..Default::default()
3105 };
3106 assert_eq!(
3107 Client::session_idle_timeout_args(&opts),
3108 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3109 );
3110 }
3111
3112 #[test]
3113 fn remote_args_omitted_by_default() {
3114 let opts = ClientOptions::default();
3115 assert!(Client::remote_args(&opts).is_empty());
3116 }
3117
3118 #[test]
3119 fn remote_args_emit_flag_when_enabled() {
3120 let opts = ClientOptions {
3121 enable_remote_sessions: true,
3122 ..Default::default()
3123 };
3124 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3125 }
3126
3127 #[test]
3128 fn log_level_args_omitted_when_unset() {
3129 let opts = ClientOptions::default();
3130 assert!(opts.log_level.is_none());
3131 assert!(
3132 Client::log_level_args(&opts).is_empty(),
3133 "with no caller-supplied log_level the SDK must not pass --log-level"
3134 );
3135 }
3136
3137 #[test]
3138 fn log_level_args_emit_flag_when_set() {
3139 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3140 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3141 }
3142
3143 #[test]
3144 fn log_level_str_round_trips() {
3145 for level in [
3146 LogLevel::None,
3147 LogLevel::Error,
3148 LogLevel::Warning,
3149 LogLevel::Info,
3150 LogLevel::Debug,
3151 LogLevel::All,
3152 ] {
3153 let s = level.as_str();
3154 let json = serde_json::to_string(&level).unwrap();
3155 assert_eq!(json, format!("\"{s}\""));
3156 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3157 assert_eq!(parsed, level);
3158 }
3159 }
3160
3161 #[test]
3162 fn client_options_debug_redacts_handler() {
3163 struct StubHandler;
3164 #[async_trait]
3165 impl ListModelsHandler for StubHandler {
3166 async fn list_models(&self) -> Result<Vec<Model>> {
3167 Ok(vec![])
3168 }
3169 }
3170 let opts = ClientOptions {
3171 on_list_models: Some(Arc::new(StubHandler)),
3172 github_token: Some("secret-token".into()),
3173 ..Default::default()
3174 };
3175 let debug = format!("{opts:?}");
3176 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3177 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3178 assert!(!debug.contains("secret-token"));
3179 }
3180
3181 #[tokio::test]
3182 async fn list_models_uses_on_list_models_handler_when_set() {
3183 use std::sync::atomic::{AtomicUsize, Ordering};
3184
3185 struct CountingHandler {
3186 calls: Arc<AtomicUsize>,
3187 models: Vec<Model>,
3188 }
3189 #[async_trait]
3190 impl ListModelsHandler for CountingHandler {
3191 async fn list_models(&self) -> Result<Vec<Model>> {
3192 self.calls.fetch_add(1, Ordering::SeqCst);
3193 Ok(self.models.clone())
3194 }
3195 }
3196
3197 let calls = Arc::new(AtomicUsize::new(0));
3198 let model = Model {
3199 id: "byok-gpt-4".into(),
3200 name: "BYOK GPT-4".into(),
3201 ..Default::default()
3202 };
3203 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3204 calls: Arc::clone(&calls),
3205 models: vec![model.clone()],
3206 });
3207
3208 let client = client_with_list_models_handler(handler);
3209
3210 let result = client.list_models().await.unwrap();
3211 assert_eq!(result.len(), 1);
3212 assert_eq!(result[0].id, "byok-gpt-4");
3213 assert_eq!(calls.load(Ordering::SeqCst), 1);
3214 }
3215
3216 #[tokio::test]
3217 async fn list_models_serializes_concurrent_cache_misses() {
3218 use std::sync::atomic::{AtomicUsize, Ordering};
3219
3220 struct SlowCountingHandler {
3221 calls: Arc<AtomicUsize>,
3222 models: Vec<Model>,
3223 }
3224 #[async_trait]
3225 impl ListModelsHandler for SlowCountingHandler {
3226 async fn list_models(&self) -> Result<Vec<Model>> {
3227 self.calls.fetch_add(1, Ordering::SeqCst);
3228 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3229 Ok(self.models.clone())
3230 }
3231 }
3232
3233 let calls = Arc::new(AtomicUsize::new(0));
3234 let model = Model {
3235 id: "single-flight-model".into(),
3236 name: "Single Flight Model".into(),
3237 ..Default::default()
3238 };
3239 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3240 calls: Arc::clone(&calls),
3241 models: vec![model],
3242 });
3243 let client = client_with_list_models_handler(handler);
3244
3245 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3246 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3247 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3248 assert_eq!(calls.load(Ordering::SeqCst), 1);
3249 }
3250
3251 #[tokio::test]
3252 async fn cancelled_resume_session_unregisters_pending_session() {
3253 let (client_write, _server_read) = tokio::io::duplex(8192);
3254 let (_server_write, client_read) = tokio::io::duplex(8192);
3255 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3256 assert!(client.startup_timings().is_none());
3257 let session_id = SessionId::new("resume-cancel-test");
3258 let handle = tokio::spawn({
3259 let client = client.clone();
3260 async move {
3261 client
3262 .resume_session(ResumeSessionConfig::new(session_id))
3263 .await
3264 }
3265 });
3266
3267 wait_for_pending_session_registration(&client).await;
3268 handle.abort();
3269 let _ = handle.await;
3270
3271 assert!(client.inner.router.session_ids().is_empty());
3272 client.force_stop();
3273 }
3274
3275 #[cfg(any(unix, windows))]
3276 #[tokio::test]
3277 async fn dropping_last_client_kills_spawned_cli() {
3278 let temp = tempfile::tempdir().unwrap();
3279 let ready = temp.path().join("ready");
3280 let survived = temp.path().join("survived");
3281 let child = test_child_command(temp.path(), &ready, &survived)
3282 .spawn()
3283 .unwrap();
3284 let (client_write, _server_read) = tokio::io::duplex(64);
3285 let (_server_write, client_read) = tokio::io::duplex(64);
3286 let client = Client::from_transport(
3287 client_read,
3288 client_write,
3289 Some(child),
3290 temp.path().to_path_buf(),
3291 None,
3292 false,
3293 false,
3294 None,
3295 None,
3296 None,
3297 ClientMode::default(),
3298 )
3299 .unwrap();
3300
3301 wait_for_test_child(&ready).await;
3302 drop(client);
3303
3304 assert_test_child_killed(&survived).await;
3305 }
3306
3307 #[cfg(any(unix, windows))]
3308 #[tokio::test]
3309 async fn spawned_child_is_killed_when_dropped() {
3310 let temp = tempfile::tempdir().unwrap();
3311 let ready = temp.path().join("ready");
3312 let survived = temp.path().join("survived");
3313 let child = test_child_command(temp.path(), &ready, &survived)
3314 .spawn()
3315 .unwrap();
3316
3317 wait_for_test_child(&ready).await;
3318 drop(child);
3319
3320 assert_test_child_killed(&survived).await;
3321 }
3322
3323 #[cfg(any(unix, windows))]
3324 fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command {
3325 #[cfg(unix)]
3326 let mut command = {
3327 let mut command =
3328 Client::build_command(Path::new("sh"), &ClientOptions::default(), temp);
3329 command.args([
3330 "-c",
3331 "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"",
3332 ]);
3333 command
3334 };
3335 #[cfg(windows)]
3336 let mut command = {
3337 let mut command =
3338 Client::build_command(Path::new("powershell.exe"), &ClientOptions::default(), temp);
3339 command.args([
3340 "-NoLogo",
3341 "-NoProfile",
3342 "-NonInteractive",
3343 "-Command",
3344 "Set-Content -LiteralPath $env:READY ready; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:SURVIVED survived",
3345 ]);
3346 command
3347 };
3348 command.env("READY", ready).env("SURVIVED", survived);
3349 command
3350 }
3351
3352 #[cfg(any(unix, windows))]
3353 async fn wait_for_test_child(ready: &Path) {
3354 let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
3355 while !ready.exists() {
3356 assert!(
3357 tokio::time::Instant::now() < deadline,
3358 "child did not report readiness"
3359 );
3360 tokio::time::sleep(Duration::from_millis(10)).await;
3361 }
3362 }
3363
3364 #[cfg(any(unix, windows))]
3365 async fn assert_test_child_killed(survived: &Path) {
3366 tokio::time::sleep(Duration::from_millis(1500)).await;
3367
3368 assert!(
3369 !survived.exists(),
3370 "child survived after its owner was dropped"
3371 );
3372 }
3373
3374 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3375 Client {
3376 inner: Arc::new(ClientInner {
3377 child: parking_lot::Mutex::new(None),
3378 #[cfg(feature = "bundled-in-process")]
3379 ffi_host: parking_lot::Mutex::new(None),
3380 rpc: {
3381 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3382 let (notif_tx, _notif_rx) = broadcast::channel(16);
3383 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3384 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3385 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3386 },
3387 cwd: PathBuf::from("."),
3388 request_rx: parking_lot::Mutex::new(None),
3389 notification_tx: broadcast::channel(16).0,
3390 router: router::SessionRouter::new(),
3391 negotiated_protocol_version: OnceLock::new(),
3392 state: parking_lot::Mutex::new(ConnectionState::Connected),
3393 lifecycle_tx: broadcast::channel(16).0,
3394 on_list_models: Some(handler),
3395 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3396 session_fs_configured: false,
3397 session_fs_sqlite_declared: false,
3398 llm_inference: OnceLock::new(),
3399 on_github_telemetry: None,
3400 on_get_trace_context: None,
3401 effective_connection_token: None,
3402 mode: ClientMode::default(),
3403 startup_timings: OnceLock::new(),
3404 }),
3405 }
3406 }
3407
3408 async fn wait_for_pending_session_registration(client: &Client) {
3409 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3410 while client.inner.router.session_ids().is_empty() {
3411 assert!(
3412 tokio::time::Instant::now() < deadline,
3413 "session was not registered"
3414 );
3415 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3416 }
3417 }
3418}