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 subscription;
45pub mod tool;
47pub mod trace_context;
49pub mod transforms;
51pub mod types;
53mod wire;
54
55pub mod session_events;
57
58pub mod rpc;
61
62pub(crate) mod generated;
67
68pub mod mode;
71
72use std::ffi::OsString;
73use std::path::{Path, PathBuf};
74use std::process::Stdio;
75use std::sync::{Arc, OnceLock};
76use std::time::{Duration, Instant};
77
78use async_trait::async_trait;
79pub use indexmap::IndexMap;
83pub(crate) use jsonrpc::{
86 JsonRpcClient, JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, error_codes,
87};
88pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet};
89pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs};
90
91#[cfg(feature = "test-support")]
93pub mod test_support {
94 pub use crate::jsonrpc::{
95 JsonRpcClient, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
96 error_codes,
97 };
98}
99use serde::{Deserialize, Serialize};
100use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader};
101use tokio::net::TcpStream;
102use tokio::process::{Child, Command};
103use tokio::sync::{broadcast, mpsc, oneshot};
104use tracing::{Instrument, debug, error, info, warn};
105pub use types::*;
106
107mod sdk_protocol_version;
108pub use sdk_protocol_version::{SDK_PROTOCOL_VERSION, get_sdk_protocol_version};
109pub use subscription::{EventSubscription, LifecycleSubscription};
110
111const MIN_PROTOCOL_VERSION: u32 = 3;
113const RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
114
115#[derive(Debug, Default)]
117#[non_exhaustive]
118pub enum Transport {
119 #[default]
122 Default,
123 Stdio,
125 InProcess,
139 Tcp {
141 port: u16,
143 connection_token: Option<String>,
147 },
148 External {
150 host: String,
152 port: u16,
154 connection_token: Option<String>,
157 },
158}
159
160#[derive(Debug, Clone, Default)]
162pub enum CliProgram {
163 #[default]
166 Resolve,
167 Path(PathBuf),
169}
170
171impl From<PathBuf> for CliProgram {
172 fn from(path: PathBuf) -> Self {
173 Self::Path(path)
174 }
175}
176
177pub const HAS_BUNDLED_CLI: bool = cfg!(has_bundled_cli);
184
185pub fn install_bundled_cli() -> Option<PathBuf> {
209 #[cfg(feature = "bundled-cli")]
210 {
211 embeddedcli::path()
212 }
213 #[cfg(not(feature = "bundled-cli"))]
214 {
215 None
216 }
217}
218
219#[non_exhaustive]
229pub struct ClientOptions {
230 pub program: CliProgram,
232 pub prefix_args: Vec<OsString>,
234 pub working_directory: PathBuf,
238 pub env: Vec<(OsString, OsString)>,
240 pub env_remove: Vec<OsString>,
242 pub extra_args: Vec<String>,
244 pub transport: Transport,
246 pub github_token: Option<String>,
251 pub use_logged_in_user: Option<bool>,
255 pub log_level: Option<LogLevel>,
259 pub session_idle_timeout_seconds: Option<u64>,
265 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
273 pub session_fs: Option<SessionFsConfig>,
281 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
290 #[doc(hidden)]
298 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
299 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
309 pub telemetry: Option<TelemetryConfig>,
313 pub base_directory: Option<PathBuf>,
318 pub enable_remote_sessions: bool,
324 pub bundled_cli_extract_dir: Option<PathBuf>,
343 pub mode: ClientMode,
347}
348
349impl std::fmt::Debug for ClientOptions {
350 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
351 f.debug_struct("ClientOptions")
352 .field("program", &self.program)
353 .field("prefix_args", &self.prefix_args)
354 .field("working_directory", &self.working_directory)
355 .field("env", &self.env)
356 .field("env_remove", &self.env_remove)
357 .field("extra_args", &self.extra_args)
358 .field("transport", &self.transport)
359 .field(
360 "github_token",
361 &self.github_token.as_ref().map(|_| "<redacted>"),
362 )
363 .field("use_logged_in_user", &self.use_logged_in_user)
364 .field("log_level", &self.log_level)
365 .field(
366 "session_idle_timeout_seconds",
367 &self.session_idle_timeout_seconds,
368 )
369 .field(
370 "on_list_models",
371 &self.on_list_models.as_ref().map(|_| "<set>"),
372 )
373 .field("session_fs", &self.session_fs)
374 .field(
375 "request_handler",
376 &self.request_handler.as_ref().map(|_| "<set>"),
377 )
378 .field(
379 "on_github_telemetry",
380 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
381 )
382 .field(
383 "on_get_trace_context",
384 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
385 )
386 .field("telemetry", &self.telemetry)
387 .field("base_directory", &self.base_directory)
388 .field("enable_remote_sessions", &self.enable_remote_sessions)
389 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
390 .finish()
391 }
392}
393
394#[async_trait]
403pub trait ListModelsHandler: Send + Sync + 'static {
404 async fn list_models(&self) -> Result<Vec<Model>>;
406}
407
408#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
410#[serde(rename_all = "lowercase")]
411pub enum LogLevel {
412 None,
414 Error,
416 Warning,
418 Info,
420 Debug,
422 All,
424}
425
426impl LogLevel {
427 pub fn as_str(self) -> &'static str {
429 match self {
430 Self::None => "none",
431 Self::Error => "error",
432 Self::Warning => "warning",
433 Self::Info => "info",
434 Self::Debug => "debug",
435 Self::All => "all",
436 }
437 }
438}
439
440impl std::fmt::Display for LogLevel {
441 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
442 f.write_str(self.as_str())
443 }
444}
445
446#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
451#[serde(rename_all = "kebab-case")]
452#[non_exhaustive]
453pub enum OtelExporterType {
454 OtlpHttp,
457 File,
460}
461
462impl OtelExporterType {
463 pub fn as_str(self) -> &'static str {
465 match self {
466 Self::OtlpHttp => "otlp-http",
467 Self::File => "file",
468 }
469 }
470}
471
472#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
478#[non_exhaustive]
479pub enum OtlpHttpProtocol {
480 #[serde(rename = "http/json")]
482 HttpJson,
483 #[serde(rename = "http/protobuf")]
485 HttpProtobuf,
486}
487
488impl OtlpHttpProtocol {
489 pub fn as_str(self) -> &'static str {
491 match self {
492 Self::HttpJson => "http/json",
493 Self::HttpProtobuf => "http/protobuf",
494 }
495 }
496}
497
498#[derive(Debug, Clone, Default)]
533#[non_exhaustive]
534pub struct TelemetryConfig {
535 pub otlp_endpoint: Option<String>,
537 pub otlp_protocol: Option<OtlpHttpProtocol>,
539 pub file_path: Option<PathBuf>,
541 pub exporter_type: Option<OtelExporterType>,
544 pub source_name: Option<String>,
548 pub capture_content: Option<bool>,
552}
553
554impl TelemetryConfig {
555 pub fn new() -> Self {
558 Self::default()
559 }
560
561 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
563 self.otlp_endpoint = Some(endpoint.into());
564 self
565 }
566
567 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
569 self.otlp_protocol = Some(protocol);
570 self
571 }
572
573 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
575 self.file_path = Some(path.into());
576 self
577 }
578
579 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
581 self.exporter_type = Some(exporter_type);
582 self
583 }
584
585 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
589 self.source_name = Some(source_name.into());
590 self
591 }
592
593 pub fn with_capture_content(mut self, capture: bool) -> Self {
597 self.capture_content = Some(capture);
598 self
599 }
600
601 pub fn is_empty(&self) -> bool {
604 self.otlp_endpoint.is_none()
605 && self.otlp_protocol.is_none()
606 && self.file_path.is_none()
607 && self.exporter_type.is_none()
608 && self.source_name.is_none()
609 && self.capture_content.is_none()
610 }
611}
612
613impl Default for ClientOptions {
614 fn default() -> Self {
615 Self {
616 program: CliProgram::Resolve,
617 prefix_args: Vec::new(),
618 working_directory: PathBuf::new(),
619 env: Vec::new(),
620 env_remove: Vec::new(),
621 extra_args: Vec::new(),
622 transport: Transport::default(),
623 github_token: None,
624 use_logged_in_user: None,
625 log_level: None,
626 session_idle_timeout_seconds: None,
627 on_list_models: None,
628 session_fs: None,
629 request_handler: None,
630 on_github_telemetry: None,
631 on_get_trace_context: None,
632 telemetry: None,
633 base_directory: None,
634 enable_remote_sessions: false,
635 bundled_cli_extract_dir: None,
636 mode: ClientMode::default(),
637 }
638 }
639}
640
641impl ClientOptions {
642 pub fn new() -> Self {
658 Self::default()
659 }
660
661 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
663 self.program = program.into();
664 self
665 }
666
667 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
669 where
670 I: IntoIterator<Item = S>,
671 S: Into<OsString>,
672 {
673 self.prefix_args = args.into_iter().map(Into::into).collect();
674 self
675 }
676
677 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
679 self.working_directory = cwd.into();
680 self
681 }
682
683 pub fn with_env<I, K, V>(mut self, env: I) -> Self
685 where
686 I: IntoIterator<Item = (K, V)>,
687 K: Into<OsString>,
688 V: Into<OsString>,
689 {
690 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
691 self
692 }
693
694 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
696 where
697 I: IntoIterator<Item = S>,
698 S: Into<OsString>,
699 {
700 self.env_remove = names.into_iter().map(Into::into).collect();
701 self
702 }
703
704 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
706 where
707 I: IntoIterator<Item = S>,
708 S: Into<String>,
709 {
710 self.extra_args = args.into_iter().map(Into::into).collect();
711 self
712 }
713
714 pub fn with_transport(mut self, transport: Transport) -> Self {
716 self.transport = transport;
717 self
718 }
719
720 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
723 self.github_token = Some(token.into());
724 self
725 }
726
727 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
730 self.use_logged_in_user = Some(use_logged_in);
731 self
732 }
733
734 pub fn with_log_level(mut self, level: LogLevel) -> Self {
736 self.log_level = Some(level);
737 self
738 }
739
740 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
743 self.session_idle_timeout_seconds = Some(seconds);
744 self
745 }
746
747 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
750 where
751 H: ListModelsHandler + 'static,
752 {
753 self.on_list_models = Some(Arc::new(handler));
754 self
755 }
756
757 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
759 self.session_fs = Some(config);
760 self
761 }
762
763 pub fn with_request_handler<H>(mut self, handler: H) -> Self
768 where
769 H: crate::copilot_request_handler::CopilotRequestHandler,
770 {
771 self.request_handler = Some(Arc::new(handler));
772 self
773 }
774
775 #[doc(hidden)]
781 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
782 where
783 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
784 {
785 self.on_github_telemetry = Some(Arc::new(callback));
786 self
787 }
788
789 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
793 where
794 P: TraceContextProvider + 'static,
795 {
796 self.on_get_trace_context = Some(Arc::new(provider));
797 self
798 }
799
800 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
802 self.telemetry = Some(config);
803 self
804 }
805
806 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
809 self.base_directory = Some(dir.into());
810 self
811 }
812
813 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
816 self.enable_remote_sessions = enabled;
817 self
818 }
819
820 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
830 self.bundled_cli_extract_dir = Some(dir.into());
831 self
832 }
833
834 pub fn with_mode(mut self, mode: ClientMode) -> Self {
839 self.mode = mode;
840 self
841 }
842}
843
844fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
846 if cfg.initial_cwd.trim().is_empty() {
847 return Err(Error::with_message(
848 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
849 "invalid SessionFsConfig: initial_cwd must not be empty",
850 ));
851 }
852 if cfg.session_state_path.trim().is_empty() {
853 return Err(Error::with_message(
854 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
855 "invalid SessionFsConfig: session_state_path must not be empty",
856 ));
857 }
858 Ok(())
859}
860
861fn generate_connection_token() -> String {
868 let mut bytes = [0u8; 16];
869 getrandom::getrandom(&mut bytes)
870 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
871 let mut hex = String::with_capacity(32);
872 for byte in bytes {
873 use std::fmt::Write;
874 let _ = write!(hex, "{byte:02x}");
875 }
876 hex
877}
878
879const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
884
885fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
887 let configured = options
888 .env
889 .iter()
890 .find(|(key, _)| {
891 key.to_string_lossy()
892 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
893 })
894 .map(|(_, value)| value.to_string_lossy().into_owned());
895 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
896 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
897}
898
899fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
900 match value {
901 None => Ok(Transport::Stdio),
902 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
903 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
904 Some(v) => Err(Error::with_message(
905 ErrorKind::InvalidConfig,
906 format!(
907 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
908 Expected 'inprocess', 'stdio', or unset."
909 ),
910 )),
911 }
912}
913
914#[cfg(any(feature = "bundled-in-process", test))]
915fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
916 let unsupported = if !options.working_directory.as_os_str().is_empty() {
917 Some("working_directory")
918 } else if !options.env.is_empty() {
919 Some("env")
920 } else if !options.env_remove.is_empty() {
921 Some("env_remove")
922 } else if options.telemetry.is_some() {
923 Some("telemetry")
924 } else if options.github_token.is_some() {
925 Some("github_token")
926 } else if !options.prefix_args.is_empty() {
927 Some("prefix_args")
928 } else {
929 None
930 };
931
932 if let Some(option) = unsupported {
933 return Err(Error::with_message(
934 ErrorKind::InvalidConfig,
935 format!(
936 "ClientOptions::{option} is not supported with Transport::InProcess; \
937 configure process-global settings on the host process instead"
938 ),
939 ));
940 }
941
942 Ok(())
943}
944
945#[derive(Clone)]
950pub struct Client {
951 inner: Arc<ClientInner>,
952}
953
954impl std::fmt::Debug for Client {
955 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
956 f.debug_struct("Client")
957 .field("working_directory", &self.inner.cwd)
958 .field("pid", &self.pid())
959 .finish()
960 }
961}
962
963struct ClientInner {
964 child: parking_lot::Mutex<Option<Child>>,
965 #[cfg(feature = "bundled-in-process")]
966 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
969 rpc: JsonRpcClient,
970 cwd: PathBuf,
971 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
972 notification_tx: broadcast::Sender<JsonRpcNotification>,
973 router: router::SessionRouter,
974 negotiated_protocol_version: OnceLock<u32>,
975 state: parking_lot::Mutex<ConnectionState>,
976 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
977 on_list_models: Option<Arc<dyn ListModelsHandler>>,
978 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
979 session_fs_configured: bool,
980 session_fs_sqlite_declared: bool,
981 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
984 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
989 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
990 effective_connection_token: Option<String>,
995 pub(crate) mode: ClientMode,
998}
999
1000impl Client {
1001 pub async fn start(options: ClientOptions) -> Result<Self> {
1014 let start_time = Instant::now();
1015 let mut options = options;
1016 if matches!(options.transport, Transport::Default) {
1017 options.transport = resolve_default_transport(&options)?;
1018 }
1019 if matches!(options.transport, Transport::InProcess) {
1020 #[cfg(not(feature = "bundled-in-process"))]
1021 {
1022 return Err(Error::with_message(
1023 ErrorKind::InvalidConfig,
1024 "Transport::InProcess requires the `bundled-in-process` Cargo feature",
1025 ));
1026 }
1027 #[cfg(feature = "bundled-in-process")]
1028 validate_inprocess_options(&options)?;
1029 }
1030 if options.mode == ClientMode::Empty
1031 && options.base_directory.is_none()
1032 && options.session_fs.is_none()
1033 {
1034 return Err(Error::with_message(
1035 ErrorKind::InvalidConfig,
1036 "ClientMode::Empty requires either `base_directory` or \
1037 `session_fs` to be set (no implicit ~/.copilot fallback).",
1038 ));
1039 }
1040 if let Some(cfg) = &options.session_fs {
1041 validate_session_fs_config(cfg)?;
1042 }
1043 if matches!(options.transport, Transport::External { .. }) {
1046 if options.github_token.is_some() {
1047 return Err(Error::with_message(
1048 ErrorKind::InvalidConfig,
1049 "invalid client configuration: github_token cannot be used with \
1050 Transport::External (external server manages its own auth)",
1051 ));
1052 }
1053 if options.use_logged_in_user == Some(true) {
1054 return Err(Error::with_message(
1055 ErrorKind::InvalidConfig,
1056 "invalid client configuration: use_logged_in_user cannot be used with \
1057 Transport::External (external server manages its own auth)",
1058 ));
1059 }
1060 }
1061 match &options.transport {
1065 Transport::Tcp {
1066 connection_token: Some(t),
1067 ..
1068 }
1069 | Transport::External {
1070 connection_token: Some(t),
1071 ..
1072 } if t.is_empty() => {
1073 return Err(Error::with_message(
1074 ErrorKind::InvalidConfig,
1075 "invalid client configuration: connection_token must be a non-empty string",
1076 ));
1077 }
1078 _ => {}
1079 }
1080 let effective_connection_token: Option<String> = match &mut options.transport {
1085 Transport::Default => unreachable!("default transport resolved above"),
1086 Transport::Stdio | Transport::InProcess => None,
1087 Transport::Tcp {
1088 connection_token, ..
1089 } => Some(
1090 connection_token
1091 .get_or_insert_with(generate_connection_token)
1092 .clone(),
1093 ),
1094 Transport::External {
1095 connection_token, ..
1096 } => connection_token.clone(),
1097 };
1098 let session_fs_config = options.session_fs.clone();
1099 let request_handler = options.request_handler.clone();
1100 let session_fs_sqlite_declared = session_fs_config
1101 .as_ref()
1102 .and_then(|c| c.capabilities.as_ref())
1103 .is_some_and(|caps| caps.sqlite);
1104 let program = match &options.program {
1105 CliProgram::Path(path) => {
1106 info!(path = %path.display(), "using explicit copilot CLI path");
1107 path.clone()
1108 }
1109 CliProgram::Resolve => {
1110 let resolved = resolve::copilot_binary_with_extract_dir(
1111 options.bundled_cli_extract_dir.as_deref(),
1112 )?;
1113 info!(path = %resolved.display(), "resolved copilot CLI");
1114 #[cfg(windows)]
1115 {
1116 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1117 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1118 }) {
1119 warn!(
1120 path = %resolved.display(),
1121 ext = %ext,
1122 "resolved copilot CLI is a .cmd/.bat wrapper; \
1123 this may cause console window flashes on Windows"
1124 );
1125 }
1126 }
1127 resolved
1128 }
1129 };
1130 let working_directory = {
1131 let cwd = options.working_directory.clone();
1132 if cwd.as_os_str().is_empty() {
1133 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1134 } else {
1135 cwd
1136 }
1137 };
1138
1139 let client = match options.transport {
1140 Transport::Default => unreachable!("default transport resolved above"),
1141 Transport::External {
1142 ref host,
1143 port,
1144 connection_token: _,
1145 } => {
1146 info!(host = %host, port = %port, "connecting to external CLI server");
1147 let connect_start = Instant::now();
1148 let stream = TcpStream::connect((host.as_str(), port)).await?;
1149 debug!(
1150 elapsed_ms = connect_start.elapsed().as_millis(),
1151 host = %host,
1152 port,
1153 "Client::start TCP connect complete"
1154 );
1155 let (reader, writer) = tokio::io::split(stream);
1156 Self::from_transport(
1157 reader,
1158 writer,
1159 None,
1160 working_directory,
1161 options.on_list_models,
1162 session_fs_config.is_some(),
1163 session_fs_sqlite_declared,
1164 options.on_get_trace_context,
1165 options.on_github_telemetry,
1166 effective_connection_token.clone(),
1167 options.mode,
1168 )?
1169 }
1170 Transport::Tcp {
1171 port,
1172 connection_token: _,
1173 } => {
1174 let (mut child, actual_port) =
1175 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1176 let connect_start = Instant::now();
1177 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1178 debug!(
1179 elapsed_ms = connect_start.elapsed().as_millis(),
1180 port = actual_port,
1181 "Client::start TCP connect complete"
1182 );
1183 let (reader, writer) = tokio::io::split(stream);
1184 Self::drain_stderr(&mut child);
1185 Self::from_transport(
1186 reader,
1187 writer,
1188 Some(child),
1189 working_directory,
1190 options.on_list_models,
1191 session_fs_config.is_some(),
1192 session_fs_sqlite_declared,
1193 options.on_get_trace_context,
1194 options.on_github_telemetry,
1195 effective_connection_token.clone(),
1196 options.mode,
1197 )?
1198 }
1199 Transport::Stdio => {
1200 let mut child = Self::spawn_stdio(&program, &options, &working_directory)?;
1201 let stdin = child.stdin.take().expect("stdin is piped");
1202 let stdout = child.stdout.take().expect("stdout is piped");
1203 Self::drain_stderr(&mut child);
1204 Self::from_transport(
1205 stdout,
1206 stdin,
1207 Some(child),
1208 working_directory,
1209 options.on_list_models,
1210 session_fs_config.is_some(),
1211 session_fs_sqlite_declared,
1212 options.on_get_trace_context,
1213 options.on_github_telemetry,
1214 effective_connection_token.clone(),
1215 options.mode,
1216 )?
1217 }
1218 Transport::InProcess => {
1219 #[cfg(feature = "bundled-in-process")]
1220 {
1221 info!(entrypoint = %program.display(), "hosting copilot runtime in-process (FFI)");
1222 let mut environment = Vec::new();
1223 if let Some(base_directory) = &options.base_directory {
1224 let value = base_directory.to_str().ok_or_else(|| {
1225 Error::with_message(
1226 ErrorKind::InvalidConfig,
1227 "base_directory must be valid UTF-8 for Transport::InProcess",
1228 )
1229 })?;
1230 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1231 }
1232 if options.mode == ClientMode::Empty {
1233 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1234 }
1235 let mut args = Vec::new();
1236 args.extend(
1237 Self::log_level_args(&options)
1238 .into_iter()
1239 .map(str::to_string),
1240 );
1241 args.extend(Self::session_idle_timeout_args(&options));
1242 args.extend(Self::remote_args(&options));
1243 if options.use_logged_in_user == Some(false) {
1244 args.push("--no-auto-login".to_string());
1245 }
1246 args.extend(options.extra_args.clone());
1247 let host = crate::ffi::FfiHost::create(&program, environment, args)?;
1248 let (reader, writer, shared) = host.start().await?;
1249 let client = Self::from_transport(
1250 reader,
1251 writer,
1252 None,
1253 working_directory,
1254 options.on_list_models,
1255 session_fs_config.is_some(),
1256 session_fs_sqlite_declared,
1257 options.on_get_trace_context,
1258 options.on_github_telemetry,
1259 effective_connection_token.clone(),
1260 options.mode,
1261 )?;
1262 *client.inner.ffi_host.lock() = Some(shared);
1263 client
1264 }
1265 #[cfg(not(feature = "bundled-in-process"))]
1266 unreachable!("in-process feature validation returned above")
1267 }
1268 };
1269 debug!(
1270 elapsed_ms = start_time.elapsed().as_millis(),
1271 "Client::start transport setup complete"
1272 );
1273 client.verify_protocol_version().await?;
1274 debug!(
1275 elapsed_ms = start_time.elapsed().as_millis(),
1276 "Client::start protocol verification complete"
1277 );
1278 if let Some(cfg) = session_fs_config {
1279 let session_fs_start = Instant::now();
1280 let capabilities = cfg.capabilities.as_ref().map(|c| {
1281 crate::generated::api_types::SessionFsSetProviderCapabilities {
1282 sqlite: Some(c.sqlite),
1283 }
1284 });
1285 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1286 capabilities,
1287 conventions: cfg.conventions.into_wire(),
1288 initial_cwd: cfg.initial_cwd,
1289 session_state_path: cfg.session_state_path,
1290 };
1291 client.rpc().session_fs().set_provider(request).await?;
1292 debug!(
1293 elapsed_ms = session_fs_start.elapsed().as_millis(),
1294 "Client::start session filesystem setup complete"
1295 );
1296 }
1297 if let Some(handler) = request_handler {
1298 let llm_inference_start = Instant::now();
1299 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1300 handler,
1301 ));
1302 dispatcher.set_client(Arc::downgrade(&client.inner));
1303 let _ = client.inner.llm_inference.set(dispatcher.clone());
1304 client.inner.router.ensure_started(
1307 &client.inner.notification_tx,
1308 &client.inner.request_rx,
1309 Some(dispatcher.clone()),
1310 client.inner.on_github_telemetry.clone(),
1311 );
1312 client.rpc().llm_inference().set_provider().await?;
1313 debug!(
1314 elapsed_ms = llm_inference_start.elapsed().as_millis(),
1315 "Client::start Copilot request handler registration complete"
1316 );
1317 }
1318 debug!(
1319 elapsed_ms = start_time.elapsed().as_millis(),
1320 "Client::start complete"
1321 );
1322 Ok(client)
1323 }
1324
1325 pub fn from_streams(
1329 reader: impl AsyncRead + Unpin + Send + 'static,
1330 writer: impl AsyncWrite + Unpin + Send + 'static,
1331 cwd: PathBuf,
1332 ) -> Result<Self> {
1333 Self::from_transport(
1334 reader,
1335 writer,
1336 None,
1337 cwd,
1338 None,
1339 false,
1340 false,
1341 None,
1342 None,
1343 None,
1344 ClientMode::default(),
1345 )
1346 }
1347
1348 #[cfg(any(test, feature = "test-support"))]
1356 pub fn from_streams_with_trace_provider(
1357 reader: impl AsyncRead + Unpin + Send + 'static,
1358 writer: impl AsyncWrite + Unpin + Send + 'static,
1359 cwd: PathBuf,
1360 provider: Arc<dyn TraceContextProvider>,
1361 ) -> Result<Self> {
1362 Self::from_transport(
1363 reader,
1364 writer,
1365 None,
1366 cwd,
1367 None,
1368 false,
1369 false,
1370 Some(provider),
1371 None,
1372 None,
1373 ClientMode::default(),
1374 )
1375 }
1376
1377 #[cfg(any(test, feature = "test-support"))]
1381 pub fn from_streams_with_connection_token(
1382 reader: impl AsyncRead + Unpin + Send + 'static,
1383 writer: impl AsyncWrite + Unpin + Send + 'static,
1384 cwd: PathBuf,
1385 token: Option<String>,
1386 ) -> Result<Self> {
1387 Self::from_transport(
1388 reader,
1389 writer,
1390 None,
1391 cwd,
1392 None,
1393 false,
1394 false,
1395 None,
1396 None,
1397 token,
1398 ClientMode::default(),
1399 )
1400 }
1401
1402 #[doc(hidden)]
1405 #[cfg(any(test, feature = "test-support"))]
1406 pub fn from_streams_with_github_telemetry(
1407 reader: impl AsyncRead + Unpin + Send + 'static,
1408 writer: impl AsyncWrite + Unpin + Send + 'static,
1409 cwd: PathBuf,
1410 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1411 ) -> Result<Self> {
1412 Self::from_transport(
1413 reader,
1414 writer,
1415 None,
1416 cwd,
1417 None,
1418 false,
1419 false,
1420 None,
1421 Some(on_github_telemetry),
1422 None,
1423 ClientMode::default(),
1424 )
1425 }
1426
1427 #[cfg(any(test, feature = "test-support"))]
1433 pub fn generate_connection_token_for_test() -> String {
1434 generate_connection_token()
1435 }
1436
1437 #[allow(clippy::too_many_arguments)]
1438 fn from_transport(
1439 reader: impl AsyncRead + Unpin + Send + 'static,
1440 writer: impl AsyncWrite + Unpin + Send + 'static,
1441 child: Option<Child>,
1442 cwd: PathBuf,
1443 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1444 session_fs_configured: bool,
1445 session_fs_sqlite_declared: bool,
1446 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1447 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1448 effective_connection_token: Option<String>,
1449 mode: ClientMode,
1450 ) -> Result<Self> {
1451 let setup_start = Instant::now();
1452 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1453 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1454 let rpc = JsonRpcClient::new(
1455 writer,
1456 reader,
1457 notification_broadcast_tx.clone(),
1458 request_tx,
1459 );
1460
1461 let pid = child.as_ref().and_then(|c| c.id());
1462 info!(pid = ?pid, "copilot CLI client ready");
1463
1464 let client = Self {
1465 inner: Arc::new(ClientInner {
1466 child: parking_lot::Mutex::new(child),
1467 #[cfg(feature = "bundled-in-process")]
1468 ffi_host: parking_lot::Mutex::new(None),
1469 rpc,
1470 cwd,
1471 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1472 notification_tx: notification_broadcast_tx,
1473 router: router::SessionRouter::new(),
1474 negotiated_protocol_version: OnceLock::new(),
1475 state: parking_lot::Mutex::new(ConnectionState::Connected),
1476 lifecycle_tx: broadcast::channel(256).0,
1477 on_list_models,
1478 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1479 session_fs_configured,
1480 session_fs_sqlite_declared,
1481 llm_inference: OnceLock::new(),
1482 on_github_telemetry,
1483 on_get_trace_context,
1484 effective_connection_token,
1485 mode,
1486 }),
1487 };
1488 client.spawn_lifecycle_dispatcher();
1489 debug!(
1490 elapsed_ms = setup_start.elapsed().as_millis(),
1491 pid = ?pid,
1492 "Client::from_transport setup complete"
1493 );
1494 Ok(client)
1495 }
1496
1497 fn spawn_lifecycle_dispatcher(&self) {
1501 let inner = Arc::clone(&self.inner);
1502 let mut notif_rx = inner.notification_tx.subscribe();
1503 tokio::spawn(async move {
1504 loop {
1505 match notif_rx.recv().await {
1506 Ok(notification) => {
1507 if notification.method != "session.lifecycle" {
1508 continue;
1509 }
1510 let Some(params) = notification.params.as_ref() else {
1511 continue;
1512 };
1513 let event: SessionLifecycleEvent =
1514 match serde_json::from_value(params.clone()) {
1515 Ok(e) => e,
1516 Err(e) => {
1517 warn!(
1518 error = %e,
1519 "failed to deserialize session.lifecycle notification"
1520 );
1521 continue;
1522 }
1523 };
1524 let _ = inner.lifecycle_tx.send(event);
1527 }
1528 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
1529 warn!(missed = n, "lifecycle dispatcher lagged");
1530 }
1531 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
1532 }
1533 }
1534 });
1535 }
1536
1537 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
1538 let mut command = Command::new(program);
1539 for arg in &options.prefix_args {
1540 command.arg(arg);
1541 }
1542 if let Some(token) = &options.github_token {
1545 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1546 }
1547 if let Some(telemetry) = &options.telemetry {
1550 command.env("COPILOT_OTEL_ENABLED", "true");
1551 if let Some(endpoint) = &telemetry.otlp_endpoint {
1552 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1553 }
1554 if let Some(protocol) = telemetry.otlp_protocol {
1555 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
1556 }
1557 if let Some(path) = &telemetry.file_path {
1558 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
1559 }
1560 if let Some(exporter) = telemetry.exporter_type {
1561 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
1562 }
1563 if let Some(source) = &telemetry.source_name {
1564 command.env("COPILOT_OTEL_SOURCE_NAME", source);
1565 }
1566 if let Some(capture) = telemetry.capture_content {
1567 command.env(
1568 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
1569 if capture { "true" } else { "false" },
1570 );
1571 }
1572 }
1573 if let Some(dir) = &options.base_directory {
1574 command.env("COPILOT_HOME", dir);
1575 }
1576 if options.mode == ClientMode::Empty {
1579 command.env("COPILOT_DISABLE_KEYTAR", "1");
1580 }
1581 if let Transport::Tcp {
1582 connection_token: Some(token),
1583 ..
1584 } = &options.transport
1585 {
1586 command.env("COPILOT_CONNECTION_TOKEN", token);
1587 }
1588 for (key, value) in &options.env {
1589 command.env(key, value);
1590 }
1591 for key in &options.env_remove {
1592 command.env_remove(key);
1593 }
1594 command
1595 .current_dir(working_directory)
1596 .stdout(Stdio::piped())
1597 .stderr(Stdio::piped());
1598
1599 #[cfg(windows)]
1600 {
1601 use std::os::windows::process::CommandExt;
1602 const CREATE_NO_WINDOW: u32 = 0x08000000;
1603 command.as_std_mut().creation_flags(CREATE_NO_WINDOW);
1604 }
1605
1606 command
1607 }
1608
1609 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
1617 let mut args: Vec<&'static str> = Vec::new();
1618 if options.github_token.is_some() {
1619 args.push("--auth-token-env");
1620 args.push("COPILOT_SDK_AUTH_TOKEN");
1621 }
1622 let use_logged_in = options
1623 .use_logged_in_user
1624 .unwrap_or(options.github_token.is_none());
1625 if !use_logged_in {
1626 args.push("--no-auto-login");
1627 }
1628 args
1629 }
1630
1631 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
1635 match options.session_idle_timeout_seconds {
1636 Some(secs) if secs > 0 => {
1637 vec!["--session-idle-timeout".to_string(), secs.to_string()]
1638 }
1639 _ => Vec::new(),
1640 }
1641 }
1642
1643 fn remote_args(options: &ClientOptions) -> Vec<String> {
1644 if options.enable_remote_sessions {
1645 vec!["--remote".to_string()]
1646 } else {
1647 Vec::new()
1648 }
1649 }
1650
1651 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
1652 match options.log_level {
1653 Some(level) => vec!["--log-level", level.as_str()],
1654 None => Vec::new(),
1655 }
1656 }
1657
1658 fn spawn_stdio(
1659 program: &Path,
1660 options: &ClientOptions,
1661 working_directory: &Path,
1662 ) -> Result<Child> {
1663 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
1664 let mut command = Self::build_command(program, options, working_directory);
1665 command
1666 .args(["--server", "--stdio", "--no-auto-update"])
1667 .args(Self::log_level_args(options))
1668 .args(Self::auth_args(options))
1669 .args(Self::session_idle_timeout_args(options))
1670 .args(Self::remote_args(options))
1671 .args(&options.extra_args)
1672 .stdin(Stdio::piped());
1673 let spawn_start = Instant::now();
1674 let child = command.spawn()?;
1675 debug!(
1676 elapsed_ms = spawn_start.elapsed().as_millis(),
1677 "Client::spawn_stdio subprocess spawned"
1678 );
1679 Ok(child)
1680 }
1681
1682 async fn spawn_tcp(
1683 program: &Path,
1684 options: &ClientOptions,
1685 working_directory: &Path,
1686 port: u16,
1687 ) -> Result<(Child, u16)> {
1688 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
1689 let mut command = Self::build_command(program, options, working_directory);
1690 command
1691 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
1692 .args(Self::log_level_args(options))
1693 .args(Self::auth_args(options))
1694 .args(Self::session_idle_timeout_args(options))
1695 .args(Self::remote_args(options))
1696 .args(&options.extra_args)
1697 .stdin(Stdio::null());
1698 let spawn_start = Instant::now();
1699 let mut child = command.spawn()?;
1700 debug!(
1701 elapsed_ms = spawn_start.elapsed().as_millis(),
1702 "Client::spawn_tcp subprocess spawned"
1703 );
1704 let stdout = child.stdout.take().expect("stdout is piped");
1705
1706 let (port_tx, port_rx) = oneshot::channel::<u16>();
1707 let span = tracing::error_span!("copilot_cli_port_scan");
1708 tokio::spawn(
1709 async move {
1710 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
1712 let mut lines = BufReader::new(stdout).lines();
1713 let mut port_tx = Some(port_tx);
1714 while let Ok(Some(line)) = lines.next_line().await {
1715 debug!(line = %line, "CLI stdout");
1716 if let Some(tx) = port_tx.take() {
1717 if let Some(caps) = port_re.captures(&line)
1718 && let Some(p) =
1719 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
1720 {
1721 let _ = tx.send(p);
1722 continue;
1723 }
1724 port_tx = Some(tx);
1726 }
1727 }
1728 }
1729 .instrument(span),
1730 );
1731
1732 let port_wait_start = Instant::now();
1733 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
1734 .await
1735 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
1736 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
1737
1738 debug!(
1739 elapsed_ms = port_wait_start.elapsed().as_millis(),
1740 port = actual_port,
1741 "Client::spawn_tcp TCP port wait complete"
1742 );
1743 info!(port = %actual_port, "CLI server listening");
1744 Ok((child, actual_port))
1745 }
1746
1747 fn drain_stderr(child: &mut Child) {
1748 if let Some(stderr) = child.stderr.take() {
1749 let span = tracing::error_span!("copilot_cli");
1750 tokio::spawn(
1751 async move {
1752 let mut reader = BufReader::new(stderr).lines();
1753 while let Ok(Some(line)) = reader.next_line().await {
1754 warn!(line = %line, "CLI stderr");
1755 }
1756 }
1757 .instrument(span),
1758 );
1759 }
1760 }
1761
1762 pub fn cwd(&self) -> &PathBuf {
1764 &self.inner.cwd
1765 }
1766
1767 pub fn mode(&self) -> ClientMode {
1769 self.inner.mode
1770 }
1771
1772 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
1783 crate::generated::rpc::ClientRpc { client: self }
1784 }
1785
1786 #[allow(dead_code, reason = "convenience for future internal use")]
1788 pub(crate) async fn send_request(
1789 &self,
1790 method: &str,
1791 params: Option<serde_json::Value>,
1792 ) -> Result<JsonRpcResponse> {
1793 self.inner.rpc.send_request(method, params).await
1794 }
1795
1796 pub async fn call(
1816 &self,
1817 method: &str,
1818 params: Option<serde_json::Value>,
1819 ) -> Result<serde_json::Value> {
1820 self.call_with_inline_callback(method, params, None).await
1821 }
1822
1823 pub(crate) async fn call_with_inline_callback(
1838 &self,
1839 method: &str,
1840 params: Option<serde_json::Value>,
1841 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
1842 ) -> Result<serde_json::Value> {
1843 let session_id: Option<SessionId> = params
1844 .as_ref()
1845 .and_then(|p| p.get("sessionId"))
1846 .and_then(|v| v.as_str())
1847 .map(SessionId::from);
1848 let response = self
1849 .inner
1850 .rpc
1851 .send_request_with_inline_callback(method, params, inline_callback)
1852 .await?;
1853 if let Some(err) = response.error {
1854 if err.message.contains("Session not found") {
1855 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
1856 session_id.unwrap_or_else(|| "unknown".into()),
1857 ))
1858 .into());
1859 }
1860 return Err(Error::with_message(
1861 ErrorKind::Rpc { code: err.code },
1862 err.message,
1863 ));
1864 }
1865 Ok(response.result.unwrap_or(serde_json::Value::Null))
1866 }
1867
1868 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
1870 self.inner.rpc.write(response).await
1871 }
1872
1873 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
1875 Self { inner }
1876 }
1877
1878 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
1882 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
1883 self.inner.request_rx.lock().take()
1884 }
1885
1886 pub(crate) fn register_session(
1894 &self,
1895 session_id: &SessionId,
1896 ) -> crate::router::SessionChannels {
1897 self.inner.router.ensure_started(
1898 &self.inner.notification_tx,
1899 &self.inner.request_rx,
1900 self.inner.llm_inference.get().cloned(),
1901 self.inner.on_github_telemetry.clone(),
1902 );
1903 self.inner.router.register(session_id)
1904 }
1905
1906 pub(crate) fn unregister_session(&self, session_id: &SessionId) {
1908 self.inner.router.unregister(session_id);
1909 }
1910
1911 pub fn protocol_version(&self) -> Option<u32> {
1918 self.inner.negotiated_protocol_version.get().copied()
1919 }
1920
1921 pub async fn verify_protocol_version(&self) -> Result<()> {
1945 let handshake_start = Instant::now();
1946 let mut used_fallback_ping = false;
1947 let server_version = match self.connect_handshake().await {
1951 Ok(v) => v,
1952 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
1953 used_fallback_ping = true;
1954 self.ping(None).await?.protocol_version
1955 }
1956 Err(e) => return Err(e),
1957 };
1958
1959 match server_version {
1960 None => {
1961 warn!("CLI server did not report protocolVersion; skipping version check");
1962 }
1963 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
1964 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
1965 server: v,
1966 min: MIN_PROTOCOL_VERSION,
1967 max: SDK_PROTOCOL_VERSION,
1968 })
1969 .into());
1970 }
1971 Some(v) => {
1972 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
1973 if existing != v {
1974 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
1975 previous: existing,
1976 current: v,
1977 })
1978 .into());
1979 }
1980 } else {
1981 let _ = self.inner.negotiated_protocol_version.set(v);
1982 }
1983 }
1984 }
1985
1986 debug!(
1987 elapsed_ms = handshake_start.elapsed().as_millis(),
1988 protocol_version = ?server_version,
1989 used_fallback_ping,
1990 "Client::verify_protocol_version protocol handshake complete"
1991 );
1992 Ok(())
1993 }
1994
1995 async fn connect_handshake(&self) -> Result<Option<u32>> {
2002 let params = crate::generated::api_types::ConnectRequest {
2003 token: self.inner.effective_connection_token.clone(),
2004 enable_git_hub_telemetry_forwarding: self
2005 .inner
2006 .on_github_telemetry
2007 .is_some()
2008 .then_some(true),
2009 };
2010 let value = self
2011 .call(
2012 crate::generated::api_types::rpc_methods::CONNECT,
2013 Some(serde_json::to_value(params)?),
2014 )
2015 .await?;
2016 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2017 Ok(Some(u32::try_from(result.protocol_version).map_err(
2018 |_| ProtocolErrorKind::InvalidProtocolVersion {
2019 server: result.protocol_version,
2020 },
2021 )?))
2022 }
2023
2024 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2032 let params = match message {
2033 Some(m) => serde_json::json!({ "message": m }),
2034 None => serde_json::json!({}),
2035 };
2036 let value = self
2037 .call(generated::api_types::rpc_methods::PING, Some(params))
2038 .await?;
2039 Ok(serde_json::from_value(value)?)
2040 }
2041
2042 pub async fn list_sessions(
2045 &self,
2046 filter: Option<SessionListFilter>,
2047 ) -> Result<Vec<SessionMetadata>> {
2048 let params = match filter {
2049 Some(f) => serde_json::json!({ "filter": f }),
2050 None => serde_json::json!({}),
2051 };
2052 let result = self.call("session.list", Some(params)).await?;
2053 let response: ListSessionsResponse = serde_json::from_value(result)?;
2054 Ok(response.sessions)
2055 }
2056
2057 pub async fn get_session_metadata(
2075 &self,
2076 session_id: &SessionId,
2077 ) -> Result<Option<SessionMetadata>> {
2078 let result = self
2079 .call(
2080 "session.getMetadata",
2081 Some(serde_json::json!({ "sessionId": session_id })),
2082 )
2083 .await?;
2084 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2085 Ok(response.session)
2086 }
2087
2088 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2090 self.call(
2091 "session.delete",
2092 Some(serde_json::json!({ "sessionId": session_id })),
2093 )
2094 .await?;
2095 Ok(())
2096 }
2097
2098 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2114 let result = self
2115 .call("session.getLastId", Some(serde_json::json!({})))
2116 .await?;
2117 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2118 Ok(response.session_id)
2119 }
2120
2121 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2126 let result = self
2127 .call("session.getForeground", Some(serde_json::json!({})))
2128 .await?;
2129 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2130 Ok(response.session_id)
2131 }
2132
2133 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2138 self.call(
2139 "session.setForeground",
2140 Some(serde_json::json!({ "sessionId": session_id })),
2141 )
2142 .await?;
2143 Ok(())
2144 }
2145
2146 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2148 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2149 Ok(serde_json::from_value(result)?)
2150 }
2151
2152 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2154 let result = self
2155 .call("auth.getStatus", Some(serde_json::json!({})))
2156 .await?;
2157 Ok(serde_json::from_value(result)?)
2158 }
2159
2160 pub async fn list_models(&self) -> Result<Vec<Model>> {
2165 let cache = self.inner.models_cache.lock().clone();
2166 let models = cache
2167 .get_or_try_init(|| async {
2168 if let Some(handler) = &self.inner.on_list_models {
2169 handler.list_models().await
2170 } else {
2171 Ok(self.rpc().models().list().await?.models)
2172 }
2173 })
2174 .await?;
2175 Ok(models.clone())
2176 }
2177
2178 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2181 if let Some(provider) = &self.inner.on_get_trace_context {
2182 provider.get_trace_context().await
2183 } else {
2184 TraceContext::default()
2185 }
2186 }
2187
2188 pub fn pid(&self) -> Option<u32> {
2190 self.inner.child.lock().as_ref().and_then(|c| c.id())
2191 }
2192
2193 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2220 let pid = self.pid();
2221 info!(pid = ?pid, "stopping CLI process");
2222 let mut errors: Vec<Error> = Vec::new();
2223
2224 for session_id in self.inner.router.session_ids() {
2227 match self
2228 .call(
2229 "session.destroy",
2230 Some(serde_json::json!({ "sessionId": session_id })),
2231 )
2232 .await
2233 {
2234 Ok(_) => {}
2235 Err(e) => {
2236 warn!(
2237 session_id = %session_id,
2238 error = %e,
2239 "session.destroy failed during Client::stop",
2240 );
2241 errors.push(e);
2242 }
2243 }
2244 self.inner.router.unregister(&session_id);
2245 }
2246
2247 let should_shutdown_runtime = self.inner.child.lock().is_some();
2248 #[cfg(feature = "bundled-in-process")]
2249 let should_shutdown_runtime =
2250 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2251 if should_shutdown_runtime {
2252 let runtime_shutdown_start = Instant::now();
2253 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2254 .await
2255 {
2256 Ok(Ok(())) => {
2257 debug!(
2258 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2259 "Client::stop runtime shutdown complete"
2260 );
2261 }
2262 Ok(Err(e)) => {
2263 warn!(
2264 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2265 error = %e,
2266 "runtime.shutdown failed during Client::stop",
2267 );
2268 errors.push(e);
2269 }
2270 Err(_) => {
2271 let e = std::io::Error::new(
2272 std::io::ErrorKind::TimedOut,
2273 "runtime.shutdown timed out during Client::stop",
2274 );
2275 warn!(
2276 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2277 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2278 error = %e,
2279 "runtime.shutdown timed out during Client::stop",
2280 );
2281 errors.push(e.into());
2282 }
2283 }
2284 }
2285
2286 let child = self.inner.child.lock().take();
2287 *self.inner.state.lock() = ConnectionState::Disconnected;
2288 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2289 if let Some(mut child) = child {
2290 match child.try_wait() {
2291 Ok(Some(_status)) => {}
2292 Ok(None) => {
2293 if let Err(e) = child.kill().await {
2300 errors.push(e.into());
2301 }
2302 }
2303 Err(e) => errors.push(e.into()),
2304 }
2305 }
2306
2307 #[cfg(feature = "bundled-in-process")]
2310 {
2311 if let Some(host) = self.inner.ffi_host.lock().take() {
2312 self.inner.rpc.force_close();
2313 host.close();
2314 }
2315 }
2316
2317 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2318 if errors.is_empty() {
2319 Ok(())
2320 } else {
2321 Err(StopErrors(errors))
2322 }
2323 }
2324
2325 pub fn force_stop(&self) {
2355 let pid = self.pid();
2356 info!(pid = ?pid, "force-stopping CLI process");
2357 if let Some(mut child) = self.inner.child.lock().take()
2358 && let Err(e) = child.start_kill()
2359 {
2360 error!(pid = ?pid, error = %e, "failed to send kill signal");
2361 }
2362 self.inner.rpc.force_close();
2363 #[cfg(feature = "bundled-in-process")]
2364 {
2365 if let Some(host) = self.inner.ffi_host.lock().take() {
2366 host.close();
2367 }
2368 }
2369 self.inner.router.clear();
2372 *self.inner.state.lock() = ConnectionState::Disconnected;
2373 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2374 }
2375
2376 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2411 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2412 }
2413}
2414
2415impl Drop for ClientInner {
2416 fn drop(&mut self) {
2417 if let Some(ref mut child) = *self.child.lock() {
2418 let pid = child.id();
2419 if let Err(e) = child.start_kill() {
2420 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2421 } else {
2422 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2423 }
2424 }
2425 #[cfg(feature = "bundled-in-process")]
2426 {
2427 if let Some(host) = self.ffi_host.lock().take() {
2428 self.rpc.force_close();
2429 host.close();
2430 }
2431 }
2432 }
2433}
2434
2435#[cfg(test)]
2436mod tests {
2437 use super::*;
2438
2439 #[test]
2440 fn is_transport_failure_matches_request_cancelled() {
2441 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2442 assert!(err.is_transport_failure());
2443 }
2444
2445 #[test]
2446 fn is_transport_failure_matches_io_error() {
2447 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2448 assert!(err.is_transport_failure());
2449 }
2450
2451 #[test]
2452 fn is_transport_failure_rejects_rpc_error() {
2453 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2454 assert!(!err.is_transport_failure());
2455 }
2456
2457 #[test]
2458 fn is_transport_failure_rejects_session_error() {
2459 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2460 assert!(!err.is_transport_failure());
2461 }
2462
2463 #[test]
2464 fn client_options_builder_composes() {
2465 let opts = ClientOptions::new()
2466 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2467 .with_prefix_args(["node"])
2468 .with_cwd(PathBuf::from("/tmp"))
2469 .with_env([("KEY", "value")])
2470 .with_env_remove(["UNWANTED"])
2471 .with_extra_args(["--quiet"])
2472 .with_github_token("ghp_test")
2473 .with_use_logged_in_user(false)
2474 .with_log_level(LogLevel::Debug)
2475 .with_session_idle_timeout_seconds(120)
2476 .with_enable_remote_sessions(true);
2477 assert!(matches!(opts.program, CliProgram::Path(_)));
2478 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2479 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2480 assert_eq!(
2481 opts.env,
2482 vec![(
2483 std::ffi::OsString::from("KEY"),
2484 std::ffi::OsString::from("value")
2485 )]
2486 );
2487 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2488 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2489 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2490 assert_eq!(opts.use_logged_in_user, Some(false));
2491 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2492 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2493 assert!(opts.enable_remote_sessions);
2494 }
2495
2496 #[test]
2497 fn default_transport_values_resolve_without_process_state() {
2498 assert!(matches!(
2499 resolve_default_transport_value(None).unwrap(),
2500 Transport::Stdio
2501 ));
2502 assert!(matches!(
2503 resolve_default_transport_value(Some("stdio")).unwrap(),
2504 Transport::Stdio
2505 ));
2506 assert!(matches!(
2507 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2508 Transport::InProcess
2509 ));
2510 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2511 }
2512
2513 #[test]
2514 fn inprocess_rejects_process_scoped_options() {
2515 let invalid = [
2516 ClientOptions::new().with_cwd("."),
2517 ClientOptions::new().with_env([("KEY", "value")]),
2518 ClientOptions::new().with_env_remove(["KEY"]),
2519 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2520 ClientOptions::new().with_github_token("token"),
2521 ClientOptions::new().with_prefix_args(["index.js"]),
2522 ];
2523
2524 for options in invalid {
2525 assert!(validate_inprocess_options(&options).is_err());
2526 }
2527 }
2528
2529 #[test]
2530 fn inprocess_allows_worker_and_rpc_options() {
2531 let options = ClientOptions::new()
2532 .with_base_directory("state")
2533 .with_log_level(LogLevel::Debug)
2534 .with_session_idle_timeout_seconds(10)
2535 .with_use_logged_in_user(false)
2536 .with_enable_remote_sessions(true)
2537 .with_extra_args(["--verbose"]);
2538
2539 assert!(validate_inprocess_options(&options).is_ok());
2540 }
2541
2542 #[cfg(not(feature = "bundled-in-process"))]
2543 #[tokio::test]
2544 async fn inprocess_requires_cargo_feature() {
2545 let error = Client::start(
2546 ClientOptions::new()
2547 .with_program(CliProgram::Path("copilot".into()))
2548 .with_transport(Transport::InProcess),
2549 )
2550 .await
2551 .unwrap_err();
2552
2553 assert!(error.to_string().contains("bundled-in-process"));
2554 }
2555
2556 #[test]
2557 fn is_transport_failure_rejects_other_protocol_errors() {
2558 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2559 assert!(!err.is_transport_failure());
2560 }
2561
2562 #[test]
2563 fn build_command_lets_env_remove_strip_injected_token() {
2564 let opts = ClientOptions {
2565 github_token: Some("secret".to_string()),
2566 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2567 ..Default::default()
2568 };
2569 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2570 let action = cmd
2572 .as_std()
2573 .get_envs()
2574 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2575 .map(|(_, v)| v);
2576 assert_eq!(
2577 action,
2578 Some(None),
2579 "env_remove should win over github_token"
2580 );
2581 }
2582
2583 #[test]
2584 fn build_command_lets_env_override_injected_token() {
2585 let opts = ClientOptions {
2586 github_token: Some("from-options".to_string()),
2587 env: vec![(
2588 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2589 std::ffi::OsString::from("from-env"),
2590 )],
2591 ..Default::default()
2592 };
2593 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2594 let value = cmd
2595 .as_std()
2596 .get_envs()
2597 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2598 .and_then(|(_, v)| v);
2599 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2600 }
2601
2602 #[test]
2603 fn build_command_injects_github_token_by_default() {
2604 let opts = ClientOptions {
2605 github_token: Some("just-the-token".to_string()),
2606 ..Default::default()
2607 };
2608 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2609 let value = cmd
2610 .as_std()
2611 .get_envs()
2612 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2613 .and_then(|(_, v)| v);
2614 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2615 }
2616
2617 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2618 cmd.as_std()
2619 .get_envs()
2620 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2621 .and_then(|(_, v)| v)
2622 }
2623
2624 #[test]
2625 fn telemetry_config_builder_composes() {
2626 let cfg = TelemetryConfig::new()
2627 .with_otlp_endpoint("http://collector:4318")
2628 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2629 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2630 .with_exporter_type(OtelExporterType::OtlpHttp)
2631 .with_source_name("my-app")
2632 .with_capture_content(true);
2633
2634 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
2635 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
2636 assert_eq!(
2637 cfg.file_path.as_deref(),
2638 Some(Path::new("/var/log/copilot.jsonl")),
2639 );
2640 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
2641 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
2642 assert_eq!(cfg.capture_content, Some(true));
2643 assert!(!cfg.is_empty());
2644 assert!(TelemetryConfig::new().is_empty());
2645 }
2646
2647 #[test]
2648 fn otlp_http_protocol_serde_matches_env_value() {
2649 for (protocol, wire) in [
2650 (OtlpHttpProtocol::HttpJson, "http/json"),
2651 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
2652 ] {
2653 assert_eq!(protocol.as_str(), wire);
2654
2655 let serialized = serde_json::to_string(&protocol).unwrap();
2656 assert_eq!(serialized, format!("\"{wire}\""));
2657
2658 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
2659 assert_eq!(deserialized, protocol);
2660 }
2661 }
2662
2663 #[test]
2664 fn build_command_sets_otel_env_when_telemetry_enabled() {
2665 let opts = ClientOptions {
2666 telemetry: Some(TelemetryConfig {
2667 otlp_endpoint: Some("http://collector:4318".to_string()),
2668 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
2669 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
2670 exporter_type: Some(OtelExporterType::OtlpHttp),
2671 source_name: Some("my-app".to_string()),
2672 capture_content: Some(true),
2673 }),
2674 ..Default::default()
2675 };
2676 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2677 assert_eq!(
2678 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2679 Some(std::ffi::OsStr::new("true")),
2680 );
2681 assert_eq!(
2682 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2683 Some(std::ffi::OsStr::new("http://collector:4318")),
2684 );
2685 assert_eq!(
2686 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
2687 Some(std::ffi::OsStr::new("http/protobuf")),
2688 );
2689 assert_eq!(
2690 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
2691 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
2692 );
2693 assert_eq!(
2694 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
2695 Some(std::ffi::OsStr::new("otlp-http")),
2696 );
2697 assert_eq!(
2698 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
2699 Some(std::ffi::OsStr::new("my-app")),
2700 );
2701 assert_eq!(
2702 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
2703 Some(std::ffi::OsStr::new("true")),
2704 );
2705 }
2706
2707 #[test]
2708 fn build_command_omits_otel_env_when_telemetry_none() {
2709 let opts = ClientOptions::default();
2710 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2711 for key in [
2712 "COPILOT_OTEL_ENABLED",
2713 "OTEL_EXPORTER_OTLP_ENDPOINT",
2714 "OTEL_EXPORTER_OTLP_PROTOCOL",
2715 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2716 "COPILOT_OTEL_EXPORTER_TYPE",
2717 "COPILOT_OTEL_SOURCE_NAME",
2718 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2719 ] {
2720 assert!(
2721 env_value(&cmd, key).is_none(),
2722 "expected {key} to be unset when telemetry is None",
2723 );
2724 }
2725 }
2726
2727 #[test]
2728 fn build_command_omits_unset_telemetry_fields() {
2729 let opts = ClientOptions {
2730 telemetry: Some(TelemetryConfig {
2731 otlp_endpoint: Some("http://collector:4318".to_string()),
2732 ..Default::default()
2733 }),
2734 ..Default::default()
2735 };
2736 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2737 assert_eq!(
2739 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2740 Some(std::ffi::OsStr::new("true")),
2741 );
2742 assert_eq!(
2743 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2744 Some(std::ffi::OsStr::new("http://collector:4318")),
2745 );
2746 for key in [
2748 "OTEL_EXPORTER_OTLP_PROTOCOL",
2749 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2750 "COPILOT_OTEL_EXPORTER_TYPE",
2751 "COPILOT_OTEL_SOURCE_NAME",
2752 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2753 ] {
2754 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
2755 }
2756 }
2757
2758 #[test]
2759 fn build_command_lets_user_env_override_telemetry() {
2760 let opts = ClientOptions {
2761 telemetry: Some(TelemetryConfig {
2762 otlp_endpoint: Some("http://from-config:4318".to_string()),
2763 ..Default::default()
2764 }),
2765 env: vec![(
2766 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
2767 std::ffi::OsString::from("http://from-user-env:4318"),
2768 )],
2769 ..Default::default()
2770 };
2771 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2772 assert_eq!(
2773 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2774 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
2775 "user-supplied options.env should override telemetry config",
2776 );
2777 }
2778
2779 #[test]
2780 fn build_command_sets_copilot_home_env_when_configured() {
2781 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
2782 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2783 assert_eq!(
2784 env_value(&cmd, "COPILOT_HOME"),
2785 Some(std::ffi::OsStr::new("/custom/copilot")),
2786 );
2787
2788 let opts = ClientOptions::default();
2789 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2790 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
2791 }
2792
2793 #[test]
2794 fn build_command_sets_connection_token_env_when_configured() {
2795 let opts = ClientOptions::new().with_transport(Transport::Tcp {
2796 port: 0,
2797 connection_token: Some("secret-token".to_string()),
2798 });
2799 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2800 assert_eq!(
2801 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
2802 Some(std::ffi::OsStr::new("secret-token")),
2803 );
2804
2805 let opts = ClientOptions::default();
2806 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2807 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
2808 }
2809
2810 #[tokio::test]
2811 async fn start_rejects_empty_connection_token() {
2812 let opts = ClientOptions::new()
2813 .with_transport(Transport::Tcp {
2814 port: 0,
2815 connection_token: Some(String::new()),
2816 })
2817 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
2818 let err = Client::start(opts).await.unwrap_err();
2819 assert!(
2820 matches!(err.kind(), ErrorKind::InvalidConfig),
2821 "got {err:?}"
2822 );
2823 }
2824
2825 #[tokio::test]
2826 async fn start_rejects_empty_external_connection_token() {
2827 let opts = ClientOptions::new()
2828 .with_transport(Transport::External {
2829 host: "127.0.0.1".to_string(),
2830 port: 1,
2831 connection_token: Some(String::new()),
2832 })
2833 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
2834 let err = Client::start(opts).await.unwrap_err();
2835 assert!(
2836 matches!(err.kind(), ErrorKind::InvalidConfig),
2837 "got {err:?}"
2838 );
2839 }
2840
2841 #[test]
2842 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
2843 let opts_true = ClientOptions {
2844 telemetry: Some(TelemetryConfig {
2845 capture_content: Some(true),
2846 ..Default::default()
2847 }),
2848 ..Default::default()
2849 };
2850 let opts_false = ClientOptions {
2851 telemetry: Some(TelemetryConfig {
2852 capture_content: Some(false),
2853 ..Default::default()
2854 }),
2855 ..Default::default()
2856 };
2857 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
2858 let cmd_false =
2859 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
2860 assert_eq!(
2861 env_value(
2862 &cmd_true,
2863 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
2864 ),
2865 Some(std::ffi::OsStr::new("true")),
2866 );
2867 assert_eq!(
2868 env_value(
2869 &cmd_false,
2870 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
2871 ),
2872 Some(std::ffi::OsStr::new("false")),
2873 );
2874 }
2875
2876 #[test]
2877 fn session_idle_timeout_args_are_omitted_by_default() {
2878 let opts = ClientOptions::default();
2879 assert!(Client::session_idle_timeout_args(&opts).is_empty());
2880 }
2881
2882 #[test]
2883 fn session_idle_timeout_args_omitted_for_zero() {
2884 let opts = ClientOptions {
2885 session_idle_timeout_seconds: Some(0),
2886 ..Default::default()
2887 };
2888 assert!(Client::session_idle_timeout_args(&opts).is_empty());
2889 }
2890
2891 #[test]
2892 fn session_idle_timeout_args_emit_flag_for_positive_value() {
2893 let opts = ClientOptions {
2894 session_idle_timeout_seconds: Some(300),
2895 ..Default::default()
2896 };
2897 assert_eq!(
2898 Client::session_idle_timeout_args(&opts),
2899 vec!["--session-idle-timeout".to_string(), "300".to_string()]
2900 );
2901 }
2902
2903 #[test]
2904 fn remote_args_omitted_by_default() {
2905 let opts = ClientOptions::default();
2906 assert!(Client::remote_args(&opts).is_empty());
2907 }
2908
2909 #[test]
2910 fn remote_args_emit_flag_when_enabled() {
2911 let opts = ClientOptions {
2912 enable_remote_sessions: true,
2913 ..Default::default()
2914 };
2915 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
2916 }
2917
2918 #[test]
2919 fn log_level_args_omitted_when_unset() {
2920 let opts = ClientOptions::default();
2921 assert!(opts.log_level.is_none());
2922 assert!(
2923 Client::log_level_args(&opts).is_empty(),
2924 "with no caller-supplied log_level the SDK must not pass --log-level"
2925 );
2926 }
2927
2928 #[test]
2929 fn log_level_args_emit_flag_when_set() {
2930 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
2931 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
2932 }
2933
2934 #[test]
2935 fn log_level_str_round_trips() {
2936 for level in [
2937 LogLevel::None,
2938 LogLevel::Error,
2939 LogLevel::Warning,
2940 LogLevel::Info,
2941 LogLevel::Debug,
2942 LogLevel::All,
2943 ] {
2944 let s = level.as_str();
2945 let json = serde_json::to_string(&level).unwrap();
2946 assert_eq!(json, format!("\"{s}\""));
2947 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
2948 assert_eq!(parsed, level);
2949 }
2950 }
2951
2952 #[test]
2953 fn client_options_debug_redacts_handler() {
2954 struct StubHandler;
2955 #[async_trait]
2956 impl ListModelsHandler for StubHandler {
2957 async fn list_models(&self) -> Result<Vec<Model>> {
2958 Ok(vec![])
2959 }
2960 }
2961 let opts = ClientOptions {
2962 on_list_models: Some(Arc::new(StubHandler)),
2963 github_token: Some("secret-token".into()),
2964 ..Default::default()
2965 };
2966 let debug = format!("{opts:?}");
2967 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
2968 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
2969 assert!(!debug.contains("secret-token"));
2970 }
2971
2972 #[tokio::test]
2973 async fn list_models_uses_on_list_models_handler_when_set() {
2974 use std::sync::atomic::{AtomicUsize, Ordering};
2975
2976 struct CountingHandler {
2977 calls: Arc<AtomicUsize>,
2978 models: Vec<Model>,
2979 }
2980 #[async_trait]
2981 impl ListModelsHandler for CountingHandler {
2982 async fn list_models(&self) -> Result<Vec<Model>> {
2983 self.calls.fetch_add(1, Ordering::SeqCst);
2984 Ok(self.models.clone())
2985 }
2986 }
2987
2988 let calls = Arc::new(AtomicUsize::new(0));
2989 let model = Model {
2990 id: "byok-gpt-4".into(),
2991 name: "BYOK GPT-4".into(),
2992 ..Default::default()
2993 };
2994 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
2995 calls: Arc::clone(&calls),
2996 models: vec![model.clone()],
2997 });
2998
2999 let client = client_with_list_models_handler(handler);
3000
3001 let result = client.list_models().await.unwrap();
3002 assert_eq!(result.len(), 1);
3003 assert_eq!(result[0].id, "byok-gpt-4");
3004 assert_eq!(calls.load(Ordering::SeqCst), 1);
3005 }
3006
3007 #[tokio::test]
3008 async fn list_models_serializes_concurrent_cache_misses() {
3009 use std::sync::atomic::{AtomicUsize, Ordering};
3010
3011 struct SlowCountingHandler {
3012 calls: Arc<AtomicUsize>,
3013 models: Vec<Model>,
3014 }
3015 #[async_trait]
3016 impl ListModelsHandler for SlowCountingHandler {
3017 async fn list_models(&self) -> Result<Vec<Model>> {
3018 self.calls.fetch_add(1, Ordering::SeqCst);
3019 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3020 Ok(self.models.clone())
3021 }
3022 }
3023
3024 let calls = Arc::new(AtomicUsize::new(0));
3025 let model = Model {
3026 id: "single-flight-model".into(),
3027 name: "Single Flight Model".into(),
3028 ..Default::default()
3029 };
3030 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3031 calls: Arc::clone(&calls),
3032 models: vec![model],
3033 });
3034 let client = client_with_list_models_handler(handler);
3035
3036 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3037 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3038 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3039 assert_eq!(calls.load(Ordering::SeqCst), 1);
3040 }
3041
3042 #[tokio::test]
3043 async fn cancelled_resume_session_unregisters_pending_session() {
3044 let (client_write, _server_read) = tokio::io::duplex(8192);
3045 let (_server_write, client_read) = tokio::io::duplex(8192);
3046 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3047 let session_id = SessionId::new("resume-cancel-test");
3048 let handle = tokio::spawn({
3049 let client = client.clone();
3050 async move {
3051 client
3052 .resume_session(ResumeSessionConfig::new(session_id))
3053 .await
3054 }
3055 });
3056
3057 wait_for_pending_session_registration(&client).await;
3058 handle.abort();
3059 let _ = handle.await;
3060
3061 assert!(client.inner.router.session_ids().is_empty());
3062 client.force_stop();
3063 }
3064
3065 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3066 Client {
3067 inner: Arc::new(ClientInner {
3068 child: parking_lot::Mutex::new(None),
3069 #[cfg(feature = "bundled-in-process")]
3070 ffi_host: parking_lot::Mutex::new(None),
3071 rpc: {
3072 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3073 let (notif_tx, _notif_rx) = broadcast::channel(16);
3074 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3075 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3076 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3077 },
3078 cwd: PathBuf::from("."),
3079 request_rx: parking_lot::Mutex::new(None),
3080 notification_tx: broadcast::channel(16).0,
3081 router: router::SessionRouter::new(),
3082 negotiated_protocol_version: OnceLock::new(),
3083 state: parking_lot::Mutex::new(ConnectionState::Connected),
3084 lifecycle_tx: broadcast::channel(16).0,
3085 on_list_models: Some(handler),
3086 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3087 session_fs_configured: false,
3088 session_fs_sqlite_declared: false,
3089 llm_inference: OnceLock::new(),
3090 on_github_telemetry: None,
3091 on_get_trace_context: None,
3092 effective_connection_token: None,
3093 mode: ClientMode::default(),
3094 }),
3095 }
3096 }
3097
3098 async fn wait_for_pending_session_registration(client: &Client) {
3099 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3100 while client.inner.router.session_ids().is_empty() {
3101 assert!(
3102 tokio::time::Instant::now() < deadline,
3103 "session was not registered"
3104 );
3105 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3106 }
3107 }
3108}