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 };
2167 let value = self
2168 .call(
2169 crate::generated::api_types::rpc_methods::CONNECT,
2170 Some(serde_json::to_value(params)?),
2171 )
2172 .await?;
2173 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2174 Ok(Some(u32::try_from(result.protocol_version).map_err(
2175 |_| ProtocolErrorKind::InvalidProtocolVersion {
2176 server: result.protocol_version,
2177 },
2178 )?))
2179 }
2180
2181 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2189 let params = match message {
2190 Some(m) => serde_json::json!({ "message": m }),
2191 None => serde_json::json!({}),
2192 };
2193 let value = self
2194 .call(generated::api_types::rpc_methods::PING, Some(params))
2195 .await?;
2196 Ok(serde_json::from_value(value)?)
2197 }
2198
2199 pub async fn list_sessions(
2202 &self,
2203 filter: Option<SessionListFilter>,
2204 ) -> Result<Vec<SessionMetadata>> {
2205 let params = match filter {
2206 Some(f) => serde_json::json!({ "filter": f }),
2207 None => serde_json::json!({}),
2208 };
2209 let result = self.call("session.list", Some(params)).await?;
2210 let response: ListSessionsResponse = serde_json::from_value(result)?;
2211 Ok(response.sessions)
2212 }
2213
2214 pub async fn get_session_metadata(
2232 &self,
2233 session_id: &SessionId,
2234 ) -> Result<Option<SessionMetadata>> {
2235 let result = self
2236 .call(
2237 "session.getMetadata",
2238 Some(serde_json::json!({ "sessionId": session_id })),
2239 )
2240 .await?;
2241 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2242 Ok(response.session)
2243 }
2244
2245 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2247 self.call(
2248 "session.delete",
2249 Some(serde_json::json!({ "sessionId": session_id })),
2250 )
2251 .await?;
2252 Ok(())
2253 }
2254
2255 #[cfg(feature = "test-support")]
2258 #[doc(hidden)]
2259 pub fn start_router_for_test(&self) {
2260 self.inner.router.ensure_started(
2261 &self.inner.notification_tx,
2262 &self.inner.request_rx,
2263 self.inner.llm_inference.get().cloned(),
2264 self.inner.on_github_telemetry.clone(),
2265 );
2266 }
2267
2268 #[cfg(feature = "test-support")]
2269 #[doc(hidden)]
2270 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2273 let mut first_error = None;
2274
2275 for session_id in self.inner.router.session_ids() {
2276 if let Err(error) = self
2277 .call(
2278 "session.destroy",
2279 Some(serde_json::json!({ "sessionId": session_id })),
2280 )
2281 .await
2282 && first_error.is_none()
2283 {
2284 first_error = Some(error);
2285 }
2286 self.inner.router.unregister(&session_id);
2287 }
2288
2289 match self.list_sessions(None).await {
2290 Ok(sessions) => {
2291 for session in sessions {
2292 if let Err(error) = self.delete_session(&session.session_id).await
2293 && first_error.is_none()
2294 {
2295 first_error = Some(error);
2296 }
2297 }
2298 }
2299 Err(error) if first_error.is_none() => first_error = Some(error),
2300 Err(_) => {}
2301 }
2302
2303 match first_error {
2304 Some(error) => Err(error),
2305 None => Ok(()),
2306 }
2307 }
2308
2309 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2325 let result = self
2326 .call("session.getLastId", Some(serde_json::json!({})))
2327 .await?;
2328 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2329 Ok(response.session_id)
2330 }
2331
2332 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2337 let result = self
2338 .call("session.getForeground", Some(serde_json::json!({})))
2339 .await?;
2340 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2341 Ok(response.session_id)
2342 }
2343
2344 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2349 self.call(
2350 "session.setForeground",
2351 Some(serde_json::json!({ "sessionId": session_id })),
2352 )
2353 .await?;
2354 Ok(())
2355 }
2356
2357 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2359 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2360 Ok(serde_json::from_value(result)?)
2361 }
2362
2363 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2365 let result = self
2366 .call("auth.getStatus", Some(serde_json::json!({})))
2367 .await?;
2368 Ok(serde_json::from_value(result)?)
2369 }
2370
2371 pub async fn list_models(&self) -> Result<Vec<Model>> {
2376 let cache = self.inner.models_cache.lock().clone();
2377 let models = cache
2378 .get_or_try_init(|| async {
2379 if let Some(handler) = &self.inner.on_list_models {
2380 handler.list_models().await
2381 } else {
2382 Ok(self.rpc().models().list().await?.models)
2383 }
2384 })
2385 .await?;
2386 Ok(models.clone())
2387 }
2388
2389 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2392 if let Some(provider) = &self.inner.on_get_trace_context {
2393 provider.get_trace_context().await
2394 } else {
2395 TraceContext::default()
2396 }
2397 }
2398
2399 pub fn pid(&self) -> Option<u32> {
2401 self.inner.child.lock().as_ref().and_then(|c| c.id())
2402 }
2403
2404 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2431 let pid = self.pid();
2432 info!(pid = ?pid, "stopping CLI process");
2433 let mut errors: Vec<Error> = Vec::new();
2434
2435 for session_id in self.inner.router.session_ids() {
2438 match self
2439 .call(
2440 "session.destroy",
2441 Some(serde_json::json!({ "sessionId": session_id })),
2442 )
2443 .await
2444 {
2445 Ok(_) => {}
2446 Err(e) => {
2447 warn!(
2448 session_id = %session_id,
2449 error = %e,
2450 "session.destroy failed during Client::stop",
2451 );
2452 errors.push(e);
2453 }
2454 }
2455 self.inner.router.unregister(&session_id);
2456 }
2457
2458 let should_shutdown_runtime = self.inner.child.lock().is_some();
2459 #[cfg(feature = "bundled-in-process")]
2460 let should_shutdown_runtime =
2461 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2462 if should_shutdown_runtime {
2463 let runtime_shutdown_start = Instant::now();
2464 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2465 .await
2466 {
2467 Ok(Ok(())) => {
2468 debug!(
2469 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2470 "Client::stop runtime shutdown complete"
2471 );
2472 }
2473 Ok(Err(e)) => {
2474 warn!(
2475 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2476 error = %e,
2477 "runtime.shutdown failed during Client::stop",
2478 );
2479 errors.push(e);
2480 }
2481 Err(_) => {
2482 let e = std::io::Error::new(
2483 std::io::ErrorKind::TimedOut,
2484 "runtime.shutdown timed out during Client::stop",
2485 );
2486 warn!(
2487 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2488 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2489 error = %e,
2490 "runtime.shutdown timed out during Client::stop",
2491 );
2492 errors.push(e.into());
2493 }
2494 }
2495 }
2496
2497 let child = self.inner.child.lock().take();
2498 *self.inner.state.lock() = ConnectionState::Disconnected;
2499 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2500 if let Some(mut child) = child {
2501 match child.try_wait() {
2502 Ok(Some(_status)) => {}
2503 Ok(None) => {
2504 if let Err(e) = child.kill().await {
2511 errors.push(e.into());
2512 }
2513 }
2514 Err(e) => errors.push(e.into()),
2515 }
2516 }
2517
2518 #[cfg(feature = "bundled-in-process")]
2521 {
2522 if let Some(host) = self.inner.ffi_host.lock().take() {
2523 self.inner.rpc.force_close();
2524 host.close();
2525 }
2526 }
2527
2528 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2529 if errors.is_empty() {
2530 Ok(())
2531 } else {
2532 Err(StopErrors(errors))
2533 }
2534 }
2535
2536 pub fn force_stop(&self) {
2566 let pid = self.pid();
2567 info!(pid = ?pid, "force-stopping CLI process");
2568 if let Some(mut child) = self.inner.child.lock().take()
2569 && let Err(e) = child.start_kill()
2570 {
2571 error!(pid = ?pid, error = %e, "failed to send kill signal");
2572 }
2573 self.inner.rpc.force_close();
2574 #[cfg(feature = "bundled-in-process")]
2575 {
2576 if let Some(host) = self.inner.ffi_host.lock().take() {
2577 host.close();
2578 }
2579 }
2580 self.inner.router.clear();
2583 *self.inner.state.lock() = ConnectionState::Disconnected;
2584 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2585 }
2586
2587 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2622 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2623 }
2624}
2625
2626impl Drop for ClientInner {
2627 fn drop(&mut self) {
2628 if let Some(ref mut child) = *self.child.lock() {
2629 let pid = child.id();
2630 if let Err(e) = child.start_kill() {
2631 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2632 } else {
2633 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2634 }
2635 }
2636 #[cfg(feature = "bundled-in-process")]
2637 {
2638 if let Some(host) = self.ffi_host.lock().take() {
2639 self.rpc.force_close();
2640 host.close();
2641 }
2642 }
2643 }
2644}
2645
2646#[cfg(test)]
2647mod tests {
2648 use super::*;
2649
2650 #[test]
2651 fn is_transport_failure_matches_request_cancelled() {
2652 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2653 assert!(err.is_transport_failure());
2654 }
2655
2656 #[test]
2657 fn is_transport_failure_matches_io_error() {
2658 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2659 assert!(err.is_transport_failure());
2660 }
2661
2662 #[test]
2663 fn is_transport_failure_rejects_rpc_error() {
2664 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2665 assert!(!err.is_transport_failure());
2666 }
2667
2668 #[test]
2669 fn is_transport_failure_rejects_session_error() {
2670 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2671 assert!(!err.is_transport_failure());
2672 }
2673
2674 #[test]
2675 fn client_options_builder_composes() {
2676 let opts = ClientOptions::new()
2677 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2678 .with_prefix_args(["node"])
2679 .with_cwd(PathBuf::from("/tmp"))
2680 .with_env([("KEY", "value")])
2681 .with_env_remove(["UNWANTED"])
2682 .with_extra_args(["--quiet"])
2683 .with_github_token("ghp_test")
2684 .with_use_logged_in_user(false)
2685 .with_log_level(LogLevel::Debug)
2686 .with_session_idle_timeout_seconds(120)
2687 .with_enable_remote_sessions(true);
2688 assert!(matches!(opts.program, CliProgram::Path(_)));
2689 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2690 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2691 assert_eq!(
2692 opts.env,
2693 vec![(
2694 std::ffi::OsString::from("KEY"),
2695 std::ffi::OsString::from("value")
2696 )]
2697 );
2698 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2699 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2700 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2701 assert_eq!(opts.use_logged_in_user, Some(false));
2702 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2703 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2704 assert!(opts.enable_remote_sessions);
2705 }
2706
2707 #[test]
2708 fn default_transport_values_resolve_without_process_state() {
2709 assert!(matches!(
2710 resolve_default_transport_value(None).unwrap(),
2711 Transport::Stdio
2712 ));
2713 assert!(matches!(
2714 resolve_default_transport_value(Some("stdio")).unwrap(),
2715 Transport::Stdio
2716 ));
2717 assert!(matches!(
2718 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2719 Transport::InProcess
2720 ));
2721 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2722 }
2723
2724 #[test]
2725 fn inprocess_rejects_process_scoped_options() {
2726 let invalid = [
2727 ClientOptions::new().with_cwd("."),
2728 ClientOptions::new().with_env([("KEY", "value")]),
2729 ClientOptions::new().with_env_remove(["KEY"]),
2730 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2731 ClientOptions::new().with_prefix_args(["index.js"]),
2732 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
2733 ClientOptions::new().with_extra_args(["--verbose"]),
2734 ];
2735
2736 for options in invalid {
2737 assert!(validate_inprocess_options(&options).is_err());
2738 }
2739 }
2740
2741 #[test]
2742 fn inprocess_allows_typed_runtime_options() {
2743 let options = ClientOptions::new()
2744 .with_base_directory("state")
2745 .with_log_level(LogLevel::Debug)
2746 .with_session_idle_timeout_seconds(10)
2747 .with_github_token("token")
2748 .with_use_logged_in_user(false)
2749 .with_enable_remote_sessions(true);
2750
2751 assert!(validate_inprocess_options(&options).is_ok());
2752 }
2753
2754 #[cfg(not(feature = "bundled-in-process"))]
2755 #[tokio::test]
2756 async fn inprocess_requires_cargo_feature() {
2757 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
2758 .await
2759 .unwrap_err();
2760
2761 assert!(error.to_string().contains("bundled-in-process"));
2762 }
2763
2764 #[test]
2765 fn is_transport_failure_rejects_other_protocol_errors() {
2766 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2767 assert!(!err.is_transport_failure());
2768 }
2769
2770 #[test]
2771 fn build_command_lets_env_remove_strip_injected_token() {
2772 let opts = ClientOptions {
2773 github_token: Some("secret".to_string()),
2774 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2775 ..Default::default()
2776 };
2777 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2778 let action = cmd
2780 .as_std()
2781 .get_envs()
2782 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2783 .map(|(_, v)| v);
2784 assert_eq!(
2785 action,
2786 Some(None),
2787 "env_remove should win over github_token"
2788 );
2789 }
2790
2791 #[test]
2792 fn build_command_lets_env_override_injected_token() {
2793 let opts = ClientOptions {
2794 github_token: Some("from-options".to_string()),
2795 env: vec![(
2796 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2797 std::ffi::OsString::from("from-env"),
2798 )],
2799 ..Default::default()
2800 };
2801 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2802 let value = cmd
2803 .as_std()
2804 .get_envs()
2805 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2806 .and_then(|(_, v)| v);
2807 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2808 }
2809
2810 #[test]
2811 fn build_command_injects_github_token_by_default() {
2812 let opts = ClientOptions {
2813 github_token: Some("just-the-token".to_string()),
2814 ..Default::default()
2815 };
2816 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2817 let value = cmd
2818 .as_std()
2819 .get_envs()
2820 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2821 .and_then(|(_, v)| v);
2822 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2823 }
2824
2825 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2826 cmd.as_std()
2827 .get_envs()
2828 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2829 .and_then(|(_, v)| v)
2830 }
2831
2832 #[test]
2833 fn telemetry_config_builder_composes() {
2834 let cfg = TelemetryConfig::new()
2835 .with_otlp_endpoint("http://collector:4318")
2836 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2837 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2838 .with_exporter_type(OtelExporterType::OtlpHttp)
2839 .with_source_name("my-app")
2840 .with_capture_content(true);
2841
2842 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
2843 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
2844 assert_eq!(
2845 cfg.file_path.as_deref(),
2846 Some(Path::new("/var/log/copilot.jsonl")),
2847 );
2848 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
2849 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
2850 assert_eq!(cfg.capture_content, Some(true));
2851 assert!(!cfg.is_empty());
2852 assert!(TelemetryConfig::new().is_empty());
2853 }
2854
2855 #[test]
2856 fn otlp_http_protocol_serde_matches_env_value() {
2857 for (protocol, wire) in [
2858 (OtlpHttpProtocol::HttpJson, "http/json"),
2859 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
2860 ] {
2861 assert_eq!(protocol.as_str(), wire);
2862
2863 let serialized = serde_json::to_string(&protocol).unwrap();
2864 assert_eq!(serialized, format!("\"{wire}\""));
2865
2866 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
2867 assert_eq!(deserialized, protocol);
2868 }
2869 }
2870
2871 #[test]
2872 fn build_command_sets_otel_env_when_telemetry_enabled() {
2873 let opts = ClientOptions {
2874 telemetry: Some(TelemetryConfig {
2875 otlp_endpoint: Some("http://collector:4318".to_string()),
2876 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
2877 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
2878 exporter_type: Some(OtelExporterType::OtlpHttp),
2879 source_name: Some("my-app".to_string()),
2880 capture_content: Some(true),
2881 }),
2882 ..Default::default()
2883 };
2884 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2885 assert_eq!(
2886 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2887 Some(std::ffi::OsStr::new("true")),
2888 );
2889 assert_eq!(
2890 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2891 Some(std::ffi::OsStr::new("http://collector:4318")),
2892 );
2893 assert_eq!(
2894 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
2895 Some(std::ffi::OsStr::new("http/protobuf")),
2896 );
2897 assert_eq!(
2898 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
2899 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
2900 );
2901 assert_eq!(
2902 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
2903 Some(std::ffi::OsStr::new("otlp-http")),
2904 );
2905 assert_eq!(
2906 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
2907 Some(std::ffi::OsStr::new("my-app")),
2908 );
2909 assert_eq!(
2910 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
2911 Some(std::ffi::OsStr::new("true")),
2912 );
2913 }
2914
2915 #[test]
2916 fn build_command_omits_otel_env_when_telemetry_none() {
2917 let opts = ClientOptions::default();
2918 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2919 for key in [
2920 "COPILOT_OTEL_ENABLED",
2921 "OTEL_EXPORTER_OTLP_ENDPOINT",
2922 "OTEL_EXPORTER_OTLP_PROTOCOL",
2923 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2924 "COPILOT_OTEL_EXPORTER_TYPE",
2925 "COPILOT_OTEL_SOURCE_NAME",
2926 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2927 ] {
2928 assert!(
2929 env_value(&cmd, key).is_none(),
2930 "expected {key} to be unset when telemetry is None",
2931 );
2932 }
2933 }
2934
2935 #[test]
2936 fn build_command_omits_unset_telemetry_fields() {
2937 let opts = ClientOptions {
2938 telemetry: Some(TelemetryConfig {
2939 otlp_endpoint: Some("http://collector:4318".to_string()),
2940 ..Default::default()
2941 }),
2942 ..Default::default()
2943 };
2944 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2945 assert_eq!(
2947 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2948 Some(std::ffi::OsStr::new("true")),
2949 );
2950 assert_eq!(
2951 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2952 Some(std::ffi::OsStr::new("http://collector:4318")),
2953 );
2954 for key in [
2956 "OTEL_EXPORTER_OTLP_PROTOCOL",
2957 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2958 "COPILOT_OTEL_EXPORTER_TYPE",
2959 "COPILOT_OTEL_SOURCE_NAME",
2960 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2961 ] {
2962 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
2963 }
2964 }
2965
2966 #[test]
2967 fn build_command_lets_user_env_override_telemetry() {
2968 let opts = ClientOptions {
2969 telemetry: Some(TelemetryConfig {
2970 otlp_endpoint: Some("http://from-config:4318".to_string()),
2971 ..Default::default()
2972 }),
2973 env: vec![(
2974 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
2975 std::ffi::OsString::from("http://from-user-env:4318"),
2976 )],
2977 ..Default::default()
2978 };
2979 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2980 assert_eq!(
2981 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2982 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
2983 "user-supplied options.env should override telemetry config",
2984 );
2985 }
2986
2987 #[test]
2988 fn build_command_sets_copilot_home_env_when_configured() {
2989 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
2990 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2991 assert_eq!(
2992 env_value(&cmd, "COPILOT_HOME"),
2993 Some(std::ffi::OsStr::new("/custom/copilot")),
2994 );
2995
2996 let opts = ClientOptions::default();
2997 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2998 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
2999 }
3000
3001 #[test]
3002 fn build_command_sets_connection_token_env_when_configured() {
3003 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3004 port: 0,
3005 connection_token: Some("secret-token".to_string()),
3006 });
3007 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3008 assert_eq!(
3009 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3010 Some(std::ffi::OsStr::new("secret-token")),
3011 );
3012
3013 let opts = ClientOptions::default();
3014 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3015 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3016 }
3017
3018 #[tokio::test]
3019 async fn start_rejects_empty_connection_token() {
3020 let opts = ClientOptions::new()
3021 .with_transport(Transport::Tcp {
3022 port: 0,
3023 connection_token: Some(String::new()),
3024 })
3025 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3026 let err = Client::start(opts).await.unwrap_err();
3027 assert!(
3028 matches!(err.kind(), ErrorKind::InvalidConfig),
3029 "got {err:?}"
3030 );
3031 }
3032
3033 #[tokio::test]
3034 async fn start_rejects_empty_external_connection_token() {
3035 let opts = ClientOptions::new()
3036 .with_transport(Transport::External {
3037 host: "127.0.0.1".to_string(),
3038 port: 1,
3039 connection_token: Some(String::new()),
3040 })
3041 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3042 let err = Client::start(opts).await.unwrap_err();
3043 assert!(
3044 matches!(err.kind(), ErrorKind::InvalidConfig),
3045 "got {err:?}"
3046 );
3047 }
3048
3049 #[test]
3050 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3051 let opts_true = ClientOptions {
3052 telemetry: Some(TelemetryConfig {
3053 capture_content: Some(true),
3054 ..Default::default()
3055 }),
3056 ..Default::default()
3057 };
3058 let opts_false = ClientOptions {
3059 telemetry: Some(TelemetryConfig {
3060 capture_content: Some(false),
3061 ..Default::default()
3062 }),
3063 ..Default::default()
3064 };
3065 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3066 let cmd_false =
3067 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3068 assert_eq!(
3069 env_value(
3070 &cmd_true,
3071 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3072 ),
3073 Some(std::ffi::OsStr::new("true")),
3074 );
3075 assert_eq!(
3076 env_value(
3077 &cmd_false,
3078 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3079 ),
3080 Some(std::ffi::OsStr::new("false")),
3081 );
3082 }
3083
3084 #[test]
3085 fn session_idle_timeout_args_are_omitted_by_default() {
3086 let opts = ClientOptions::default();
3087 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3088 }
3089
3090 #[test]
3091 fn session_idle_timeout_args_omitted_for_zero() {
3092 let opts = ClientOptions {
3093 session_idle_timeout_seconds: Some(0),
3094 ..Default::default()
3095 };
3096 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3097 }
3098
3099 #[test]
3100 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3101 let opts = ClientOptions {
3102 session_idle_timeout_seconds: Some(300),
3103 ..Default::default()
3104 };
3105 assert_eq!(
3106 Client::session_idle_timeout_args(&opts),
3107 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3108 );
3109 }
3110
3111 #[test]
3112 fn remote_args_omitted_by_default() {
3113 let opts = ClientOptions::default();
3114 assert!(Client::remote_args(&opts).is_empty());
3115 }
3116
3117 #[test]
3118 fn remote_args_emit_flag_when_enabled() {
3119 let opts = ClientOptions {
3120 enable_remote_sessions: true,
3121 ..Default::default()
3122 };
3123 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3124 }
3125
3126 #[test]
3127 fn log_level_args_omitted_when_unset() {
3128 let opts = ClientOptions::default();
3129 assert!(opts.log_level.is_none());
3130 assert!(
3131 Client::log_level_args(&opts).is_empty(),
3132 "with no caller-supplied log_level the SDK must not pass --log-level"
3133 );
3134 }
3135
3136 #[test]
3137 fn log_level_args_emit_flag_when_set() {
3138 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3139 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3140 }
3141
3142 #[test]
3143 fn log_level_str_round_trips() {
3144 for level in [
3145 LogLevel::None,
3146 LogLevel::Error,
3147 LogLevel::Warning,
3148 LogLevel::Info,
3149 LogLevel::Debug,
3150 LogLevel::All,
3151 ] {
3152 let s = level.as_str();
3153 let json = serde_json::to_string(&level).unwrap();
3154 assert_eq!(json, format!("\"{s}\""));
3155 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3156 assert_eq!(parsed, level);
3157 }
3158 }
3159
3160 #[test]
3161 fn client_options_debug_redacts_handler() {
3162 struct StubHandler;
3163 #[async_trait]
3164 impl ListModelsHandler for StubHandler {
3165 async fn list_models(&self) -> Result<Vec<Model>> {
3166 Ok(vec![])
3167 }
3168 }
3169 let opts = ClientOptions {
3170 on_list_models: Some(Arc::new(StubHandler)),
3171 github_token: Some("secret-token".into()),
3172 ..Default::default()
3173 };
3174 let debug = format!("{opts:?}");
3175 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3176 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3177 assert!(!debug.contains("secret-token"));
3178 }
3179
3180 #[tokio::test]
3181 async fn list_models_uses_on_list_models_handler_when_set() {
3182 use std::sync::atomic::{AtomicUsize, Ordering};
3183
3184 struct CountingHandler {
3185 calls: Arc<AtomicUsize>,
3186 models: Vec<Model>,
3187 }
3188 #[async_trait]
3189 impl ListModelsHandler for CountingHandler {
3190 async fn list_models(&self) -> Result<Vec<Model>> {
3191 self.calls.fetch_add(1, Ordering::SeqCst);
3192 Ok(self.models.clone())
3193 }
3194 }
3195
3196 let calls = Arc::new(AtomicUsize::new(0));
3197 let model = Model {
3198 id: "byok-gpt-4".into(),
3199 name: "BYOK GPT-4".into(),
3200 ..Default::default()
3201 };
3202 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3203 calls: Arc::clone(&calls),
3204 models: vec![model.clone()],
3205 });
3206
3207 let client = client_with_list_models_handler(handler);
3208
3209 let result = client.list_models().await.unwrap();
3210 assert_eq!(result.len(), 1);
3211 assert_eq!(result[0].id, "byok-gpt-4");
3212 assert_eq!(calls.load(Ordering::SeqCst), 1);
3213 }
3214
3215 #[tokio::test]
3216 async fn list_models_serializes_concurrent_cache_misses() {
3217 use std::sync::atomic::{AtomicUsize, Ordering};
3218
3219 struct SlowCountingHandler {
3220 calls: Arc<AtomicUsize>,
3221 models: Vec<Model>,
3222 }
3223 #[async_trait]
3224 impl ListModelsHandler for SlowCountingHandler {
3225 async fn list_models(&self) -> Result<Vec<Model>> {
3226 self.calls.fetch_add(1, Ordering::SeqCst);
3227 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3228 Ok(self.models.clone())
3229 }
3230 }
3231
3232 let calls = Arc::new(AtomicUsize::new(0));
3233 let model = Model {
3234 id: "single-flight-model".into(),
3235 name: "Single Flight Model".into(),
3236 ..Default::default()
3237 };
3238 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3239 calls: Arc::clone(&calls),
3240 models: vec![model],
3241 });
3242 let client = client_with_list_models_handler(handler);
3243
3244 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3245 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3246 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3247 assert_eq!(calls.load(Ordering::SeqCst), 1);
3248 }
3249
3250 #[tokio::test]
3251 async fn cancelled_resume_session_unregisters_pending_session() {
3252 let (client_write, _server_read) = tokio::io::duplex(8192);
3253 let (_server_write, client_read) = tokio::io::duplex(8192);
3254 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3255 assert!(client.startup_timings().is_none());
3256 let session_id = SessionId::new("resume-cancel-test");
3257 let handle = tokio::spawn({
3258 let client = client.clone();
3259 async move {
3260 client
3261 .resume_session(ResumeSessionConfig::new(session_id))
3262 .await
3263 }
3264 });
3265
3266 wait_for_pending_session_registration(&client).await;
3267 handle.abort();
3268 let _ = handle.await;
3269
3270 assert!(client.inner.router.session_ids().is_empty());
3271 client.force_stop();
3272 }
3273
3274 #[cfg(any(unix, windows))]
3275 #[tokio::test]
3276 async fn dropping_last_client_kills_spawned_cli() {
3277 let temp = tempfile::tempdir().unwrap();
3278 let ready = temp.path().join("ready");
3279 let survived = temp.path().join("survived");
3280 let child = test_child_command(temp.path(), &ready, &survived)
3281 .spawn()
3282 .unwrap();
3283 let (client_write, _server_read) = tokio::io::duplex(64);
3284 let (_server_write, client_read) = tokio::io::duplex(64);
3285 let client = Client::from_transport(
3286 client_read,
3287 client_write,
3288 Some(child),
3289 temp.path().to_path_buf(),
3290 None,
3291 false,
3292 false,
3293 None,
3294 None,
3295 None,
3296 ClientMode::default(),
3297 )
3298 .unwrap();
3299
3300 wait_for_test_child(&ready).await;
3301 drop(client);
3302
3303 assert_test_child_killed(&survived).await;
3304 }
3305
3306 #[cfg(any(unix, windows))]
3307 #[tokio::test]
3308 async fn spawned_child_is_killed_when_dropped() {
3309 let temp = tempfile::tempdir().unwrap();
3310 let ready = temp.path().join("ready");
3311 let survived = temp.path().join("survived");
3312 let child = test_child_command(temp.path(), &ready, &survived)
3313 .spawn()
3314 .unwrap();
3315
3316 wait_for_test_child(&ready).await;
3317 drop(child);
3318
3319 assert_test_child_killed(&survived).await;
3320 }
3321
3322 #[cfg(any(unix, windows))]
3323 fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command {
3324 #[cfg(unix)]
3325 let mut command = {
3326 let mut command =
3327 Client::build_command(Path::new("sh"), &ClientOptions::default(), temp);
3328 command.args([
3329 "-c",
3330 "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"",
3331 ]);
3332 command
3333 };
3334 #[cfg(windows)]
3335 let mut command = {
3336 let mut command =
3337 Client::build_command(Path::new("powershell.exe"), &ClientOptions::default(), temp);
3338 command.args([
3339 "-NoLogo",
3340 "-NoProfile",
3341 "-NonInteractive",
3342 "-Command",
3343 "Set-Content -LiteralPath $env:READY ready; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:SURVIVED survived",
3344 ]);
3345 command
3346 };
3347 command.env("READY", ready).env("SURVIVED", survived);
3348 command
3349 }
3350
3351 #[cfg(any(unix, windows))]
3352 async fn wait_for_test_child(ready: &Path) {
3353 let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
3354 while !ready.exists() {
3355 assert!(
3356 tokio::time::Instant::now() < deadline,
3357 "child did not report readiness"
3358 );
3359 tokio::time::sleep(Duration::from_millis(10)).await;
3360 }
3361 }
3362
3363 #[cfg(any(unix, windows))]
3364 async fn assert_test_child_killed(survived: &Path) {
3365 tokio::time::sleep(Duration::from_millis(1500)).await;
3366
3367 assert!(
3368 !survived.exists(),
3369 "child survived after its owner was dropped"
3370 );
3371 }
3372
3373 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3374 Client {
3375 inner: Arc::new(ClientInner {
3376 child: parking_lot::Mutex::new(None),
3377 #[cfg(feature = "bundled-in-process")]
3378 ffi_host: parking_lot::Mutex::new(None),
3379 rpc: {
3380 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3381 let (notif_tx, _notif_rx) = broadcast::channel(16);
3382 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3383 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3384 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3385 },
3386 cwd: PathBuf::from("."),
3387 request_rx: parking_lot::Mutex::new(None),
3388 notification_tx: broadcast::channel(16).0,
3389 router: router::SessionRouter::new(),
3390 negotiated_protocol_version: OnceLock::new(),
3391 state: parking_lot::Mutex::new(ConnectionState::Connected),
3392 lifecycle_tx: broadcast::channel(16).0,
3393 on_list_models: Some(handler),
3394 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3395 session_fs_configured: false,
3396 session_fs_sqlite_declared: false,
3397 llm_inference: OnceLock::new(),
3398 on_github_telemetry: None,
3399 on_get_trace_context: None,
3400 effective_connection_token: None,
3401 mode: ClientMode::default(),
3402 startup_timings: OnceLock::new(),
3403 }),
3404 }
3405 }
3406
3407 async fn wait_for_pending_session_registration(client: &Client) {
3408 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3409 while client.inner.router.session_ids().is_empty() {
3410 assert!(
3411 tokio::time::Instant::now() < deadline,
3412 "session was not registered"
3413 );
3414 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3415 }
3416 }
3417}