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 inner = Arc::clone(&self.inner);
1645 let mut notif_rx = inner.notification_tx.subscribe();
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 _ = inner.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 for arg in &options.prefix_args {
1683 command.arg(arg);
1684 }
1685 if let Some(token) = &options.github_token {
1688 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1689 }
1690 if let Some(telemetry) = &options.telemetry {
1693 command.env("COPILOT_OTEL_ENABLED", "true");
1694 if let Some(endpoint) = &telemetry.otlp_endpoint {
1695 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1696 }
1697 if let Some(protocol) = telemetry.otlp_protocol {
1698 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
1699 }
1700 if let Some(path) = &telemetry.file_path {
1701 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
1702 }
1703 if let Some(exporter) = telemetry.exporter_type {
1704 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
1705 }
1706 if let Some(source) = &telemetry.source_name {
1707 command.env("COPILOT_OTEL_SOURCE_NAME", source);
1708 }
1709 if let Some(capture) = telemetry.capture_content {
1710 command.env(
1711 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
1712 if capture { "true" } else { "false" },
1713 );
1714 }
1715 }
1716 if let Some(dir) = &options.base_directory {
1717 command.env("COPILOT_HOME", dir);
1718 }
1719 if options.mode == ClientMode::Empty {
1722 command.env("COPILOT_DISABLE_KEYTAR", "1");
1723 }
1724 if let Transport::Tcp {
1725 connection_token: Some(token),
1726 ..
1727 } = &options.transport
1728 {
1729 command.env("COPILOT_CONNECTION_TOKEN", token);
1730 }
1731 for (key, value) in &options.env {
1732 command.env(key, value);
1733 }
1734 for key in &options.env_remove {
1735 command.env_remove(key);
1736 }
1737 command
1738 .current_dir(working_directory)
1739 .stdout(Stdio::piped())
1740 .stderr(Stdio::piped());
1741
1742 #[cfg(windows)]
1743 {
1744 use std::os::windows::process::CommandExt;
1745 const CREATE_NO_WINDOW: u32 = 0x08000000;
1746 command.as_std_mut().creation_flags(CREATE_NO_WINDOW);
1747 }
1748
1749 command
1750 }
1751
1752 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
1760 let mut args: Vec<&'static str> = Vec::new();
1761 if options.github_token.is_some() {
1762 args.push("--auth-token-env");
1763 args.push("COPILOT_SDK_AUTH_TOKEN");
1764 }
1765 let use_logged_in = options
1766 .use_logged_in_user
1767 .unwrap_or(options.github_token.is_none());
1768 if !use_logged_in {
1769 args.push("--no-auto-login");
1770 }
1771 args
1772 }
1773
1774 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
1778 match options.session_idle_timeout_seconds {
1779 Some(secs) if secs > 0 => {
1780 vec!["--session-idle-timeout".to_string(), secs.to_string()]
1781 }
1782 _ => Vec::new(),
1783 }
1784 }
1785
1786 fn remote_args(options: &ClientOptions) -> Vec<String> {
1787 if options.enable_remote_sessions {
1788 vec!["--remote".to_string()]
1789 } else {
1790 Vec::new()
1791 }
1792 }
1793
1794 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
1795 match options.log_level {
1796 Some(level) => vec!["--log-level", level.as_str()],
1797 None => Vec::new(),
1798 }
1799 }
1800
1801 fn spawn_stdio(
1802 program: &Path,
1803 options: &ClientOptions,
1804 working_directory: &Path,
1805 ) -> Result<(Child, Duration)> {
1806 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
1807 let mut command = Self::build_command(program, options, working_directory);
1808 command
1809 .args(["--server", "--stdio", "--no-auto-update"])
1810 .args(Self::log_level_args(options))
1811 .args(Self::auth_args(options))
1812 .args(Self::session_idle_timeout_args(options))
1813 .args(Self::remote_args(options))
1814 .args(&options.extra_args)
1815 .stdin(Stdio::piped());
1816 let spawn_start = Instant::now();
1817 let child = command.spawn()?;
1818 let spawn_elapsed = spawn_start.elapsed();
1819 debug!(
1820 elapsed_ms = spawn_elapsed.as_millis(),
1821 "Client::spawn_stdio subprocess spawned"
1822 );
1823 Ok((child, spawn_elapsed))
1824 }
1825
1826 async fn spawn_tcp(
1827 program: &Path,
1828 options: &ClientOptions,
1829 working_directory: &Path,
1830 port: u16,
1831 ) -> Result<(Child, u16, Duration, Duration)> {
1832 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
1833 let mut command = Self::build_command(program, options, working_directory);
1834 command
1835 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
1836 .args(Self::log_level_args(options))
1837 .args(Self::auth_args(options))
1838 .args(Self::session_idle_timeout_args(options))
1839 .args(Self::remote_args(options))
1840 .args(&options.extra_args)
1841 .stdin(Stdio::null());
1842 let spawn_start = Instant::now();
1843 let mut child = command.spawn()?;
1844 let spawn_elapsed = spawn_start.elapsed();
1845 debug!(
1846 elapsed_ms = spawn_elapsed.as_millis(),
1847 "Client::spawn_tcp subprocess spawned"
1848 );
1849 let stdout = child.stdout.take().expect("stdout is piped");
1850
1851 let (port_tx, port_rx) = oneshot::channel::<u16>();
1852 let span = tracing::error_span!("copilot_cli_port_scan");
1853 tokio::spawn(
1854 async move {
1855 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
1857 let mut lines = BufReader::new(stdout).lines();
1858 let mut port_tx = Some(port_tx);
1859 while let Ok(Some(line)) = lines.next_line().await {
1860 debug!(line = %line, "CLI stdout");
1861 if let Some(tx) = port_tx.take() {
1862 if let Some(caps) = port_re.captures(&line)
1863 && let Some(p) =
1864 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
1865 {
1866 let _ = tx.send(p);
1867 continue;
1868 }
1869 port_tx = Some(tx);
1871 }
1872 }
1873 }
1874 .instrument(span),
1875 );
1876
1877 let port_wait_start = Instant::now();
1878 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
1879 .await
1880 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
1881 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
1882
1883 let port_wait_elapsed = port_wait_start.elapsed();
1884 debug!(
1885 elapsed_ms = port_wait_elapsed.as_millis(),
1886 port = actual_port,
1887 "Client::spawn_tcp TCP port wait complete"
1888 );
1889 info!(port = %actual_port, "CLI server listening");
1890 Ok((child, actual_port, spawn_elapsed, port_wait_elapsed))
1891 }
1892
1893 fn drain_stderr(child: &mut Child) {
1894 if let Some(stderr) = child.stderr.take() {
1895 let span = tracing::error_span!("copilot_cli");
1896 tokio::spawn(
1897 async move {
1898 let mut reader = BufReader::new(stderr).lines();
1899 while let Ok(Some(line)) = reader.next_line().await {
1900 warn!(line = %line, "CLI stderr");
1901 }
1902 }
1903 .instrument(span),
1904 );
1905 }
1906 }
1907
1908 pub fn cwd(&self) -> &PathBuf {
1910 &self.inner.cwd
1911 }
1912
1913 pub fn mode(&self) -> ClientMode {
1915 self.inner.mode
1916 }
1917
1918 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
1929 crate::generated::rpc::ClientRpc { client: self }
1930 }
1931
1932 #[allow(dead_code, reason = "convenience for future internal use")]
1934 pub(crate) async fn send_request(
1935 &self,
1936 method: &str,
1937 params: Option<serde_json::Value>,
1938 ) -> Result<JsonRpcResponse> {
1939 self.inner.rpc.send_request(method, params).await
1940 }
1941
1942 pub async fn call(
1962 &self,
1963 method: &str,
1964 params: Option<serde_json::Value>,
1965 ) -> Result<serde_json::Value> {
1966 self.call_with_inline_callback(method, params, None).await
1967 }
1968
1969 pub(crate) async fn call_with_inline_callback(
1984 &self,
1985 method: &str,
1986 params: Option<serde_json::Value>,
1987 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
1988 ) -> Result<serde_json::Value> {
1989 let session_id: Option<SessionId> = params
1990 .as_ref()
1991 .and_then(|p| p.get("sessionId"))
1992 .and_then(|v| v.as_str())
1993 .map(SessionId::from);
1994 let response = self
1995 .inner
1996 .rpc
1997 .send_request_with_inline_callback(method, params, inline_callback)
1998 .await?;
1999 if let Some(err) = response.error {
2000 if err.message.contains("Session not found") {
2001 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
2002 session_id.unwrap_or_else(|| "unknown".into()),
2003 ))
2004 .into());
2005 }
2006 return Err(Error::with_message(
2007 ErrorKind::Rpc { code: err.code },
2008 err.message,
2009 ));
2010 }
2011 Ok(response.result.unwrap_or(serde_json::Value::Null))
2012 }
2013
2014 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
2016 self.inner.rpc.write(response).await
2017 }
2018
2019 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
2021 Self { inner }
2022 }
2023
2024 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
2028 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
2029 self.inner.request_rx.lock().take()
2030 }
2031
2032 pub(crate) fn register_session(
2040 &self,
2041 session_id: &SessionId,
2042 ) -> crate::router::SessionChannels {
2043 self.inner.router.ensure_started(
2044 &self.inner.notification_tx,
2045 &self.inner.request_rx,
2046 self.inner.llm_inference.get().cloned(),
2047 self.inner.on_github_telemetry.clone(),
2048 );
2049 self.inner.router.register(session_id)
2050 }
2051
2052 pub(crate) fn unregister_session(&self, session_id: &SessionId) {
2054 self.inner.router.unregister(session_id);
2055 }
2056
2057 pub fn protocol_version(&self) -> Option<u32> {
2064 self.inner.negotiated_protocol_version.get().copied()
2065 }
2066
2067 pub fn startup_timings(&self) -> Option<StartupTimings> {
2074 self.inner.startup_timings.get().cloned()
2075 }
2076
2077 pub async fn verify_protocol_version(&self) -> Result<()> {
2101 let handshake_start = Instant::now();
2102 let mut used_fallback_ping = false;
2103 let server_version = match self.connect_handshake().await {
2107 Ok(v) => v,
2108 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2109 used_fallback_ping = true;
2110 self.ping(None).await?.protocol_version
2111 }
2112 Err(e) => return Err(e),
2113 };
2114
2115 match server_version {
2116 None => {
2117 warn!("CLI server did not report protocolVersion; skipping version check");
2118 }
2119 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2120 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2121 server: v,
2122 min: MIN_PROTOCOL_VERSION,
2123 max: SDK_PROTOCOL_VERSION,
2124 })
2125 .into());
2126 }
2127 Some(v) => {
2128 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2129 if existing != v {
2130 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2131 previous: existing,
2132 current: v,
2133 })
2134 .into());
2135 }
2136 } else {
2137 let _ = self.inner.negotiated_protocol_version.set(v);
2138 }
2139 }
2140 }
2141
2142 debug!(
2143 elapsed_ms = handshake_start.elapsed().as_millis(),
2144 protocol_version = ?server_version,
2145 used_fallback_ping,
2146 "Client::verify_protocol_version protocol handshake complete"
2147 );
2148 Ok(())
2149 }
2150
2151 async fn connect_handshake(&self) -> Result<Option<u32>> {
2158 let params = crate::generated::api_types::ConnectRequest {
2159 token: self.inner.effective_connection_token.clone(),
2160 enable_git_hub_telemetry_forwarding: self
2161 .inner
2162 .on_github_telemetry
2163 .is_some()
2164 .then_some(true),
2165 };
2166 let value = self
2167 .call(
2168 crate::generated::api_types::rpc_methods::CONNECT,
2169 Some(serde_json::to_value(params)?),
2170 )
2171 .await?;
2172 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2173 Ok(Some(u32::try_from(result.protocol_version).map_err(
2174 |_| ProtocolErrorKind::InvalidProtocolVersion {
2175 server: result.protocol_version,
2176 },
2177 )?))
2178 }
2179
2180 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2188 let params = match message {
2189 Some(m) => serde_json::json!({ "message": m }),
2190 None => serde_json::json!({}),
2191 };
2192 let value = self
2193 .call(generated::api_types::rpc_methods::PING, Some(params))
2194 .await?;
2195 Ok(serde_json::from_value(value)?)
2196 }
2197
2198 pub async fn list_sessions(
2201 &self,
2202 filter: Option<SessionListFilter>,
2203 ) -> Result<Vec<SessionMetadata>> {
2204 let params = match filter {
2205 Some(f) => serde_json::json!({ "filter": f }),
2206 None => serde_json::json!({}),
2207 };
2208 let result = self.call("session.list", Some(params)).await?;
2209 let response: ListSessionsResponse = serde_json::from_value(result)?;
2210 Ok(response.sessions)
2211 }
2212
2213 pub async fn get_session_metadata(
2231 &self,
2232 session_id: &SessionId,
2233 ) -> Result<Option<SessionMetadata>> {
2234 let result = self
2235 .call(
2236 "session.getMetadata",
2237 Some(serde_json::json!({ "sessionId": session_id })),
2238 )
2239 .await?;
2240 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2241 Ok(response.session)
2242 }
2243
2244 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2246 self.call(
2247 "session.delete",
2248 Some(serde_json::json!({ "sessionId": session_id })),
2249 )
2250 .await?;
2251 Ok(())
2252 }
2253
2254 #[cfg(feature = "test-support")]
2257 #[doc(hidden)]
2258 pub fn start_router_for_test(&self) {
2259 self.inner.router.ensure_started(
2260 &self.inner.notification_tx,
2261 &self.inner.request_rx,
2262 self.inner.llm_inference.get().cloned(),
2263 self.inner.on_github_telemetry.clone(),
2264 );
2265 }
2266
2267 #[cfg(feature = "test-support")]
2268 #[doc(hidden)]
2269 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2272 let mut first_error = None;
2273
2274 for session_id in self.inner.router.session_ids() {
2275 if let Err(error) = self
2276 .call(
2277 "session.destroy",
2278 Some(serde_json::json!({ "sessionId": session_id })),
2279 )
2280 .await
2281 && first_error.is_none()
2282 {
2283 first_error = Some(error);
2284 }
2285 self.inner.router.unregister(&session_id);
2286 }
2287
2288 match self.list_sessions(None).await {
2289 Ok(sessions) => {
2290 for session in sessions {
2291 if let Err(error) = self.delete_session(&session.session_id).await
2292 && first_error.is_none()
2293 {
2294 first_error = Some(error);
2295 }
2296 }
2297 }
2298 Err(error) if first_error.is_none() => first_error = Some(error),
2299 Err(_) => {}
2300 }
2301
2302 match first_error {
2303 Some(error) => Err(error),
2304 None => Ok(()),
2305 }
2306 }
2307
2308 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2324 let result = self
2325 .call("session.getLastId", Some(serde_json::json!({})))
2326 .await?;
2327 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2328 Ok(response.session_id)
2329 }
2330
2331 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2336 let result = self
2337 .call("session.getForeground", Some(serde_json::json!({})))
2338 .await?;
2339 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2340 Ok(response.session_id)
2341 }
2342
2343 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2348 self.call(
2349 "session.setForeground",
2350 Some(serde_json::json!({ "sessionId": session_id })),
2351 )
2352 .await?;
2353 Ok(())
2354 }
2355
2356 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2358 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2359 Ok(serde_json::from_value(result)?)
2360 }
2361
2362 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2364 let result = self
2365 .call("auth.getStatus", Some(serde_json::json!({})))
2366 .await?;
2367 Ok(serde_json::from_value(result)?)
2368 }
2369
2370 pub async fn list_models(&self) -> Result<Vec<Model>> {
2375 let cache = self.inner.models_cache.lock().clone();
2376 let models = cache
2377 .get_or_try_init(|| async {
2378 if let Some(handler) = &self.inner.on_list_models {
2379 handler.list_models().await
2380 } else {
2381 Ok(self.rpc().models().list().await?.models)
2382 }
2383 })
2384 .await?;
2385 Ok(models.clone())
2386 }
2387
2388 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2391 if let Some(provider) = &self.inner.on_get_trace_context {
2392 provider.get_trace_context().await
2393 } else {
2394 TraceContext::default()
2395 }
2396 }
2397
2398 pub fn pid(&self) -> Option<u32> {
2400 self.inner.child.lock().as_ref().and_then(|c| c.id())
2401 }
2402
2403 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2430 let pid = self.pid();
2431 info!(pid = ?pid, "stopping CLI process");
2432 let mut errors: Vec<Error> = Vec::new();
2433
2434 for session_id in self.inner.router.session_ids() {
2437 match self
2438 .call(
2439 "session.destroy",
2440 Some(serde_json::json!({ "sessionId": session_id })),
2441 )
2442 .await
2443 {
2444 Ok(_) => {}
2445 Err(e) => {
2446 warn!(
2447 session_id = %session_id,
2448 error = %e,
2449 "session.destroy failed during Client::stop",
2450 );
2451 errors.push(e);
2452 }
2453 }
2454 self.inner.router.unregister(&session_id);
2455 }
2456
2457 let should_shutdown_runtime = self.inner.child.lock().is_some();
2458 #[cfg(feature = "bundled-in-process")]
2459 let should_shutdown_runtime =
2460 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2461 if should_shutdown_runtime {
2462 let runtime_shutdown_start = Instant::now();
2463 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2464 .await
2465 {
2466 Ok(Ok(())) => {
2467 debug!(
2468 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2469 "Client::stop runtime shutdown complete"
2470 );
2471 }
2472 Ok(Err(e)) => {
2473 warn!(
2474 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2475 error = %e,
2476 "runtime.shutdown failed during Client::stop",
2477 );
2478 errors.push(e);
2479 }
2480 Err(_) => {
2481 let e = std::io::Error::new(
2482 std::io::ErrorKind::TimedOut,
2483 "runtime.shutdown timed out during Client::stop",
2484 );
2485 warn!(
2486 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2487 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2488 error = %e,
2489 "runtime.shutdown timed out during Client::stop",
2490 );
2491 errors.push(e.into());
2492 }
2493 }
2494 }
2495
2496 let child = self.inner.child.lock().take();
2497 *self.inner.state.lock() = ConnectionState::Disconnected;
2498 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2499 if let Some(mut child) = child {
2500 match child.try_wait() {
2501 Ok(Some(_status)) => {}
2502 Ok(None) => {
2503 if let Err(e) = child.kill().await {
2510 errors.push(e.into());
2511 }
2512 }
2513 Err(e) => errors.push(e.into()),
2514 }
2515 }
2516
2517 #[cfg(feature = "bundled-in-process")]
2520 {
2521 if let Some(host) = self.inner.ffi_host.lock().take() {
2522 self.inner.rpc.force_close();
2523 host.close();
2524 }
2525 }
2526
2527 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2528 if errors.is_empty() {
2529 Ok(())
2530 } else {
2531 Err(StopErrors(errors))
2532 }
2533 }
2534
2535 pub fn force_stop(&self) {
2565 let pid = self.pid();
2566 info!(pid = ?pid, "force-stopping CLI process");
2567 if let Some(mut child) = self.inner.child.lock().take()
2568 && let Err(e) = child.start_kill()
2569 {
2570 error!(pid = ?pid, error = %e, "failed to send kill signal");
2571 }
2572 self.inner.rpc.force_close();
2573 #[cfg(feature = "bundled-in-process")]
2574 {
2575 if let Some(host) = self.inner.ffi_host.lock().take() {
2576 host.close();
2577 }
2578 }
2579 self.inner.router.clear();
2582 *self.inner.state.lock() = ConnectionState::Disconnected;
2583 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2584 }
2585
2586 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2621 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2622 }
2623}
2624
2625impl Drop for ClientInner {
2626 fn drop(&mut self) {
2627 if let Some(ref mut child) = *self.child.lock() {
2628 let pid = child.id();
2629 if let Err(e) = child.start_kill() {
2630 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2631 } else {
2632 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2633 }
2634 }
2635 #[cfg(feature = "bundled-in-process")]
2636 {
2637 if let Some(host) = self.ffi_host.lock().take() {
2638 self.rpc.force_close();
2639 host.close();
2640 }
2641 }
2642 }
2643}
2644
2645#[cfg(test)]
2646mod tests {
2647 use super::*;
2648
2649 #[test]
2650 fn is_transport_failure_matches_request_cancelled() {
2651 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2652 assert!(err.is_transport_failure());
2653 }
2654
2655 #[test]
2656 fn is_transport_failure_matches_io_error() {
2657 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2658 assert!(err.is_transport_failure());
2659 }
2660
2661 #[test]
2662 fn is_transport_failure_rejects_rpc_error() {
2663 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2664 assert!(!err.is_transport_failure());
2665 }
2666
2667 #[test]
2668 fn is_transport_failure_rejects_session_error() {
2669 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2670 assert!(!err.is_transport_failure());
2671 }
2672
2673 #[test]
2674 fn client_options_builder_composes() {
2675 let opts = ClientOptions::new()
2676 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2677 .with_prefix_args(["node"])
2678 .with_cwd(PathBuf::from("/tmp"))
2679 .with_env([("KEY", "value")])
2680 .with_env_remove(["UNWANTED"])
2681 .with_extra_args(["--quiet"])
2682 .with_github_token("ghp_test")
2683 .with_use_logged_in_user(false)
2684 .with_log_level(LogLevel::Debug)
2685 .with_session_idle_timeout_seconds(120)
2686 .with_enable_remote_sessions(true);
2687 assert!(matches!(opts.program, CliProgram::Path(_)));
2688 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2689 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2690 assert_eq!(
2691 opts.env,
2692 vec![(
2693 std::ffi::OsString::from("KEY"),
2694 std::ffi::OsString::from("value")
2695 )]
2696 );
2697 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2698 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2699 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2700 assert_eq!(opts.use_logged_in_user, Some(false));
2701 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2702 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2703 assert!(opts.enable_remote_sessions);
2704 }
2705
2706 #[test]
2707 fn default_transport_values_resolve_without_process_state() {
2708 assert!(matches!(
2709 resolve_default_transport_value(None).unwrap(),
2710 Transport::Stdio
2711 ));
2712 assert!(matches!(
2713 resolve_default_transport_value(Some("stdio")).unwrap(),
2714 Transport::Stdio
2715 ));
2716 assert!(matches!(
2717 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2718 Transport::InProcess
2719 ));
2720 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2721 }
2722
2723 #[test]
2724 fn inprocess_rejects_process_scoped_options() {
2725 let invalid = [
2726 ClientOptions::new().with_cwd("."),
2727 ClientOptions::new().with_env([("KEY", "value")]),
2728 ClientOptions::new().with_env_remove(["KEY"]),
2729 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2730 ClientOptions::new().with_prefix_args(["index.js"]),
2731 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
2732 ClientOptions::new().with_extra_args(["--verbose"]),
2733 ];
2734
2735 for options in invalid {
2736 assert!(validate_inprocess_options(&options).is_err());
2737 }
2738 }
2739
2740 #[test]
2741 fn inprocess_allows_typed_runtime_options() {
2742 let options = ClientOptions::new()
2743 .with_base_directory("state")
2744 .with_log_level(LogLevel::Debug)
2745 .with_session_idle_timeout_seconds(10)
2746 .with_github_token("token")
2747 .with_use_logged_in_user(false)
2748 .with_enable_remote_sessions(true);
2749
2750 assert!(validate_inprocess_options(&options).is_ok());
2751 }
2752
2753 #[cfg(not(feature = "bundled-in-process"))]
2754 #[tokio::test]
2755 async fn inprocess_requires_cargo_feature() {
2756 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
2757 .await
2758 .unwrap_err();
2759
2760 assert!(error.to_string().contains("bundled-in-process"));
2761 }
2762
2763 #[test]
2764 fn is_transport_failure_rejects_other_protocol_errors() {
2765 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2766 assert!(!err.is_transport_failure());
2767 }
2768
2769 #[test]
2770 fn build_command_lets_env_remove_strip_injected_token() {
2771 let opts = ClientOptions {
2772 github_token: Some("secret".to_string()),
2773 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2774 ..Default::default()
2775 };
2776 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2777 let action = cmd
2779 .as_std()
2780 .get_envs()
2781 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2782 .map(|(_, v)| v);
2783 assert_eq!(
2784 action,
2785 Some(None),
2786 "env_remove should win over github_token"
2787 );
2788 }
2789
2790 #[test]
2791 fn build_command_lets_env_override_injected_token() {
2792 let opts = ClientOptions {
2793 github_token: Some("from-options".to_string()),
2794 env: vec![(
2795 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2796 std::ffi::OsString::from("from-env"),
2797 )],
2798 ..Default::default()
2799 };
2800 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2801 let value = cmd
2802 .as_std()
2803 .get_envs()
2804 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2805 .and_then(|(_, v)| v);
2806 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2807 }
2808
2809 #[test]
2810 fn build_command_injects_github_token_by_default() {
2811 let opts = ClientOptions {
2812 github_token: Some("just-the-token".to_string()),
2813 ..Default::default()
2814 };
2815 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2816 let value = cmd
2817 .as_std()
2818 .get_envs()
2819 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2820 .and_then(|(_, v)| v);
2821 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2822 }
2823
2824 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2825 cmd.as_std()
2826 .get_envs()
2827 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2828 .and_then(|(_, v)| v)
2829 }
2830
2831 #[test]
2832 fn telemetry_config_builder_composes() {
2833 let cfg = TelemetryConfig::new()
2834 .with_otlp_endpoint("http://collector:4318")
2835 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2836 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2837 .with_exporter_type(OtelExporterType::OtlpHttp)
2838 .with_source_name("my-app")
2839 .with_capture_content(true);
2840
2841 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
2842 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
2843 assert_eq!(
2844 cfg.file_path.as_deref(),
2845 Some(Path::new("/var/log/copilot.jsonl")),
2846 );
2847 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
2848 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
2849 assert_eq!(cfg.capture_content, Some(true));
2850 assert!(!cfg.is_empty());
2851 assert!(TelemetryConfig::new().is_empty());
2852 }
2853
2854 #[test]
2855 fn otlp_http_protocol_serde_matches_env_value() {
2856 for (protocol, wire) in [
2857 (OtlpHttpProtocol::HttpJson, "http/json"),
2858 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
2859 ] {
2860 assert_eq!(protocol.as_str(), wire);
2861
2862 let serialized = serde_json::to_string(&protocol).unwrap();
2863 assert_eq!(serialized, format!("\"{wire}\""));
2864
2865 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
2866 assert_eq!(deserialized, protocol);
2867 }
2868 }
2869
2870 #[test]
2871 fn build_command_sets_otel_env_when_telemetry_enabled() {
2872 let opts = ClientOptions {
2873 telemetry: Some(TelemetryConfig {
2874 otlp_endpoint: Some("http://collector:4318".to_string()),
2875 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
2876 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
2877 exporter_type: Some(OtelExporterType::OtlpHttp),
2878 source_name: Some("my-app".to_string()),
2879 capture_content: Some(true),
2880 }),
2881 ..Default::default()
2882 };
2883 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2884 assert_eq!(
2885 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2886 Some(std::ffi::OsStr::new("true")),
2887 );
2888 assert_eq!(
2889 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2890 Some(std::ffi::OsStr::new("http://collector:4318")),
2891 );
2892 assert_eq!(
2893 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
2894 Some(std::ffi::OsStr::new("http/protobuf")),
2895 );
2896 assert_eq!(
2897 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
2898 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
2899 );
2900 assert_eq!(
2901 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
2902 Some(std::ffi::OsStr::new("otlp-http")),
2903 );
2904 assert_eq!(
2905 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
2906 Some(std::ffi::OsStr::new("my-app")),
2907 );
2908 assert_eq!(
2909 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
2910 Some(std::ffi::OsStr::new("true")),
2911 );
2912 }
2913
2914 #[test]
2915 fn build_command_omits_otel_env_when_telemetry_none() {
2916 let opts = ClientOptions::default();
2917 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2918 for key in [
2919 "COPILOT_OTEL_ENABLED",
2920 "OTEL_EXPORTER_OTLP_ENDPOINT",
2921 "OTEL_EXPORTER_OTLP_PROTOCOL",
2922 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2923 "COPILOT_OTEL_EXPORTER_TYPE",
2924 "COPILOT_OTEL_SOURCE_NAME",
2925 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2926 ] {
2927 assert!(
2928 env_value(&cmd, key).is_none(),
2929 "expected {key} to be unset when telemetry is None",
2930 );
2931 }
2932 }
2933
2934 #[test]
2935 fn build_command_omits_unset_telemetry_fields() {
2936 let opts = ClientOptions {
2937 telemetry: Some(TelemetryConfig {
2938 otlp_endpoint: Some("http://collector:4318".to_string()),
2939 ..Default::default()
2940 }),
2941 ..Default::default()
2942 };
2943 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2944 assert_eq!(
2946 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2947 Some(std::ffi::OsStr::new("true")),
2948 );
2949 assert_eq!(
2950 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2951 Some(std::ffi::OsStr::new("http://collector:4318")),
2952 );
2953 for key in [
2955 "OTEL_EXPORTER_OTLP_PROTOCOL",
2956 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2957 "COPILOT_OTEL_EXPORTER_TYPE",
2958 "COPILOT_OTEL_SOURCE_NAME",
2959 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2960 ] {
2961 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
2962 }
2963 }
2964
2965 #[test]
2966 fn build_command_lets_user_env_override_telemetry() {
2967 let opts = ClientOptions {
2968 telemetry: Some(TelemetryConfig {
2969 otlp_endpoint: Some("http://from-config:4318".to_string()),
2970 ..Default::default()
2971 }),
2972 env: vec![(
2973 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
2974 std::ffi::OsString::from("http://from-user-env:4318"),
2975 )],
2976 ..Default::default()
2977 };
2978 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2979 assert_eq!(
2980 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2981 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
2982 "user-supplied options.env should override telemetry config",
2983 );
2984 }
2985
2986 #[test]
2987 fn build_command_sets_copilot_home_env_when_configured() {
2988 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
2989 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2990 assert_eq!(
2991 env_value(&cmd, "COPILOT_HOME"),
2992 Some(std::ffi::OsStr::new("/custom/copilot")),
2993 );
2994
2995 let opts = ClientOptions::default();
2996 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2997 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
2998 }
2999
3000 #[test]
3001 fn build_command_sets_connection_token_env_when_configured() {
3002 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3003 port: 0,
3004 connection_token: Some("secret-token".to_string()),
3005 });
3006 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3007 assert_eq!(
3008 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3009 Some(std::ffi::OsStr::new("secret-token")),
3010 );
3011
3012 let opts = ClientOptions::default();
3013 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3014 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3015 }
3016
3017 #[tokio::test]
3018 async fn start_rejects_empty_connection_token() {
3019 let opts = ClientOptions::new()
3020 .with_transport(Transport::Tcp {
3021 port: 0,
3022 connection_token: Some(String::new()),
3023 })
3024 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3025 let err = Client::start(opts).await.unwrap_err();
3026 assert!(
3027 matches!(err.kind(), ErrorKind::InvalidConfig),
3028 "got {err:?}"
3029 );
3030 }
3031
3032 #[tokio::test]
3033 async fn start_rejects_empty_external_connection_token() {
3034 let opts = ClientOptions::new()
3035 .with_transport(Transport::External {
3036 host: "127.0.0.1".to_string(),
3037 port: 1,
3038 connection_token: Some(String::new()),
3039 })
3040 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3041 let err = Client::start(opts).await.unwrap_err();
3042 assert!(
3043 matches!(err.kind(), ErrorKind::InvalidConfig),
3044 "got {err:?}"
3045 );
3046 }
3047
3048 #[test]
3049 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3050 let opts_true = ClientOptions {
3051 telemetry: Some(TelemetryConfig {
3052 capture_content: Some(true),
3053 ..Default::default()
3054 }),
3055 ..Default::default()
3056 };
3057 let opts_false = ClientOptions {
3058 telemetry: Some(TelemetryConfig {
3059 capture_content: Some(false),
3060 ..Default::default()
3061 }),
3062 ..Default::default()
3063 };
3064 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3065 let cmd_false =
3066 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3067 assert_eq!(
3068 env_value(
3069 &cmd_true,
3070 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3071 ),
3072 Some(std::ffi::OsStr::new("true")),
3073 );
3074 assert_eq!(
3075 env_value(
3076 &cmd_false,
3077 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3078 ),
3079 Some(std::ffi::OsStr::new("false")),
3080 );
3081 }
3082
3083 #[test]
3084 fn session_idle_timeout_args_are_omitted_by_default() {
3085 let opts = ClientOptions::default();
3086 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3087 }
3088
3089 #[test]
3090 fn session_idle_timeout_args_omitted_for_zero() {
3091 let opts = ClientOptions {
3092 session_idle_timeout_seconds: Some(0),
3093 ..Default::default()
3094 };
3095 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3096 }
3097
3098 #[test]
3099 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3100 let opts = ClientOptions {
3101 session_idle_timeout_seconds: Some(300),
3102 ..Default::default()
3103 };
3104 assert_eq!(
3105 Client::session_idle_timeout_args(&opts),
3106 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3107 );
3108 }
3109
3110 #[test]
3111 fn remote_args_omitted_by_default() {
3112 let opts = ClientOptions::default();
3113 assert!(Client::remote_args(&opts).is_empty());
3114 }
3115
3116 #[test]
3117 fn remote_args_emit_flag_when_enabled() {
3118 let opts = ClientOptions {
3119 enable_remote_sessions: true,
3120 ..Default::default()
3121 };
3122 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3123 }
3124
3125 #[test]
3126 fn log_level_args_omitted_when_unset() {
3127 let opts = ClientOptions::default();
3128 assert!(opts.log_level.is_none());
3129 assert!(
3130 Client::log_level_args(&opts).is_empty(),
3131 "with no caller-supplied log_level the SDK must not pass --log-level"
3132 );
3133 }
3134
3135 #[test]
3136 fn log_level_args_emit_flag_when_set() {
3137 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3138 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3139 }
3140
3141 #[test]
3142 fn log_level_str_round_trips() {
3143 for level in [
3144 LogLevel::None,
3145 LogLevel::Error,
3146 LogLevel::Warning,
3147 LogLevel::Info,
3148 LogLevel::Debug,
3149 LogLevel::All,
3150 ] {
3151 let s = level.as_str();
3152 let json = serde_json::to_string(&level).unwrap();
3153 assert_eq!(json, format!("\"{s}\""));
3154 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3155 assert_eq!(parsed, level);
3156 }
3157 }
3158
3159 #[test]
3160 fn client_options_debug_redacts_handler() {
3161 struct StubHandler;
3162 #[async_trait]
3163 impl ListModelsHandler for StubHandler {
3164 async fn list_models(&self) -> Result<Vec<Model>> {
3165 Ok(vec![])
3166 }
3167 }
3168 let opts = ClientOptions {
3169 on_list_models: Some(Arc::new(StubHandler)),
3170 github_token: Some("secret-token".into()),
3171 ..Default::default()
3172 };
3173 let debug = format!("{opts:?}");
3174 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3175 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3176 assert!(!debug.contains("secret-token"));
3177 }
3178
3179 #[tokio::test]
3180 async fn list_models_uses_on_list_models_handler_when_set() {
3181 use std::sync::atomic::{AtomicUsize, Ordering};
3182
3183 struct CountingHandler {
3184 calls: Arc<AtomicUsize>,
3185 models: Vec<Model>,
3186 }
3187 #[async_trait]
3188 impl ListModelsHandler for CountingHandler {
3189 async fn list_models(&self) -> Result<Vec<Model>> {
3190 self.calls.fetch_add(1, Ordering::SeqCst);
3191 Ok(self.models.clone())
3192 }
3193 }
3194
3195 let calls = Arc::new(AtomicUsize::new(0));
3196 let model = Model {
3197 id: "byok-gpt-4".into(),
3198 name: "BYOK GPT-4".into(),
3199 ..Default::default()
3200 };
3201 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3202 calls: Arc::clone(&calls),
3203 models: vec![model.clone()],
3204 });
3205
3206 let client = client_with_list_models_handler(handler);
3207
3208 let result = client.list_models().await.unwrap();
3209 assert_eq!(result.len(), 1);
3210 assert_eq!(result[0].id, "byok-gpt-4");
3211 assert_eq!(calls.load(Ordering::SeqCst), 1);
3212 }
3213
3214 #[tokio::test]
3215 async fn list_models_serializes_concurrent_cache_misses() {
3216 use std::sync::atomic::{AtomicUsize, Ordering};
3217
3218 struct SlowCountingHandler {
3219 calls: Arc<AtomicUsize>,
3220 models: Vec<Model>,
3221 }
3222 #[async_trait]
3223 impl ListModelsHandler for SlowCountingHandler {
3224 async fn list_models(&self) -> Result<Vec<Model>> {
3225 self.calls.fetch_add(1, Ordering::SeqCst);
3226 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3227 Ok(self.models.clone())
3228 }
3229 }
3230
3231 let calls = Arc::new(AtomicUsize::new(0));
3232 let model = Model {
3233 id: "single-flight-model".into(),
3234 name: "Single Flight Model".into(),
3235 ..Default::default()
3236 };
3237 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3238 calls: Arc::clone(&calls),
3239 models: vec![model],
3240 });
3241 let client = client_with_list_models_handler(handler);
3242
3243 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3244 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3245 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3246 assert_eq!(calls.load(Ordering::SeqCst), 1);
3247 }
3248
3249 #[tokio::test]
3250 async fn cancelled_resume_session_unregisters_pending_session() {
3251 let (client_write, _server_read) = tokio::io::duplex(8192);
3252 let (_server_write, client_read) = tokio::io::duplex(8192);
3253 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3254 assert!(client.startup_timings().is_none());
3255 let session_id = SessionId::new("resume-cancel-test");
3256 let handle = tokio::spawn({
3257 let client = client.clone();
3258 async move {
3259 client
3260 .resume_session(ResumeSessionConfig::new(session_id))
3261 .await
3262 }
3263 });
3264
3265 wait_for_pending_session_registration(&client).await;
3266 handle.abort();
3267 let _ = handle.await;
3268
3269 assert!(client.inner.router.session_ids().is_empty());
3270 client.force_stop();
3271 }
3272
3273 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3274 Client {
3275 inner: Arc::new(ClientInner {
3276 child: parking_lot::Mutex::new(None),
3277 #[cfg(feature = "bundled-in-process")]
3278 ffi_host: parking_lot::Mutex::new(None),
3279 rpc: {
3280 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3281 let (notif_tx, _notif_rx) = broadcast::channel(16);
3282 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3283 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3284 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3285 },
3286 cwd: PathBuf::from("."),
3287 request_rx: parking_lot::Mutex::new(None),
3288 notification_tx: broadcast::channel(16).0,
3289 router: router::SessionRouter::new(),
3290 negotiated_protocol_version: OnceLock::new(),
3291 state: parking_lot::Mutex::new(ConnectionState::Connected),
3292 lifecycle_tx: broadcast::channel(16).0,
3293 on_list_models: Some(handler),
3294 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3295 session_fs_configured: false,
3296 session_fs_sqlite_declared: false,
3297 llm_inference: OnceLock::new(),
3298 on_github_telemetry: None,
3299 on_get_trace_context: None,
3300 effective_connection_token: None,
3301 mode: ClientMode::default(),
3302 startup_timings: OnceLock::new(),
3303 }),
3304 }
3305 }
3306
3307 async fn wait_for_pending_session_registration(client: &Client) {
3308 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3309 while client.inner.router.session_ids().is_empty() {
3310 assert!(
3311 tokio::time::Instant::now() < deadline,
3312 "session was not registered"
3313 );
3314 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3315 }
3316 }
3317}