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 transport: Transport,
259 pub github_token: Option<String>,
264 pub use_logged_in_user: Option<bool>,
268 pub log_level: Option<LogLevel>,
272 pub session_idle_timeout_seconds: Option<u64>,
278 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
286 pub session_fs: Option<SessionFsConfig>,
294 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
303 #[doc(hidden)]
311 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
312 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
322 pub telemetry: Option<TelemetryConfig>,
326 pub base_directory: Option<PathBuf>,
331 pub enable_remote_sessions: bool,
337 pub bundled_cli_extract_dir: Option<PathBuf>,
356 pub mode: ClientMode,
360}
361
362impl std::fmt::Debug for ClientOptions {
363 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
364 f.debug_struct("ClientOptions")
365 .field("program", &self.program)
366 .field("prefix_args", &self.prefix_args)
367 .field("working_directory", &self.working_directory)
368 .field("env", &self.env)
369 .field("env_remove", &self.env_remove)
370 .field("extra_args", &self.extra_args)
371 .field("transport", &self.transport)
372 .field(
373 "github_token",
374 &self.github_token.as_ref().map(|_| "<redacted>"),
375 )
376 .field("use_logged_in_user", &self.use_logged_in_user)
377 .field("log_level", &self.log_level)
378 .field(
379 "session_idle_timeout_seconds",
380 &self.session_idle_timeout_seconds,
381 )
382 .field(
383 "on_list_models",
384 &self.on_list_models.as_ref().map(|_| "<set>"),
385 )
386 .field("session_fs", &self.session_fs)
387 .field(
388 "request_handler",
389 &self.request_handler.as_ref().map(|_| "<set>"),
390 )
391 .field(
392 "on_github_telemetry",
393 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
394 )
395 .field(
396 "on_get_trace_context",
397 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
398 )
399 .field("telemetry", &self.telemetry)
400 .field("base_directory", &self.base_directory)
401 .field("enable_remote_sessions", &self.enable_remote_sessions)
402 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
403 .finish()
404 }
405}
406
407#[async_trait]
416pub trait ListModelsHandler: Send + Sync + 'static {
417 async fn list_models(&self) -> Result<Vec<Model>>;
419}
420
421#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
423#[serde(rename_all = "lowercase")]
424pub enum LogLevel {
425 None,
427 Error,
429 Warning,
431 Info,
433 Debug,
435 All,
437}
438
439impl LogLevel {
440 pub fn as_str(self) -> &'static str {
442 match self {
443 Self::None => "none",
444 Self::Error => "error",
445 Self::Warning => "warning",
446 Self::Info => "info",
447 Self::Debug => "debug",
448 Self::All => "all",
449 }
450 }
451}
452
453impl std::fmt::Display for LogLevel {
454 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
455 f.write_str(self.as_str())
456 }
457}
458
459#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
464#[serde(rename_all = "kebab-case")]
465#[non_exhaustive]
466pub enum OtelExporterType {
467 OtlpHttp,
470 File,
473}
474
475impl OtelExporterType {
476 pub fn as_str(self) -> &'static str {
478 match self {
479 Self::OtlpHttp => "otlp-http",
480 Self::File => "file",
481 }
482 }
483}
484
485#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
491#[non_exhaustive]
492pub enum OtlpHttpProtocol {
493 #[serde(rename = "http/json")]
495 HttpJson,
496 #[serde(rename = "http/protobuf")]
498 HttpProtobuf,
499}
500
501impl OtlpHttpProtocol {
502 pub fn as_str(self) -> &'static str {
504 match self {
505 Self::HttpJson => "http/json",
506 Self::HttpProtobuf => "http/protobuf",
507 }
508 }
509}
510
511#[derive(Debug, Clone, Default)]
546#[non_exhaustive]
547pub struct TelemetryConfig {
548 pub otlp_endpoint: Option<String>,
550 pub otlp_protocol: Option<OtlpHttpProtocol>,
552 pub file_path: Option<PathBuf>,
554 pub exporter_type: Option<OtelExporterType>,
557 pub source_name: Option<String>,
561 pub capture_content: Option<bool>,
565}
566
567impl TelemetryConfig {
568 pub fn new() -> Self {
571 Self::default()
572 }
573
574 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
576 self.otlp_endpoint = Some(endpoint.into());
577 self
578 }
579
580 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
582 self.otlp_protocol = Some(protocol);
583 self
584 }
585
586 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
588 self.file_path = Some(path.into());
589 self
590 }
591
592 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
594 self.exporter_type = Some(exporter_type);
595 self
596 }
597
598 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
602 self.source_name = Some(source_name.into());
603 self
604 }
605
606 pub fn with_capture_content(mut self, capture: bool) -> Self {
610 self.capture_content = Some(capture);
611 self
612 }
613
614 pub fn is_empty(&self) -> bool {
617 self.otlp_endpoint.is_none()
618 && self.otlp_protocol.is_none()
619 && self.file_path.is_none()
620 && self.exporter_type.is_none()
621 && self.source_name.is_none()
622 && self.capture_content.is_none()
623 }
624}
625
626impl Default for ClientOptions {
627 fn default() -> Self {
628 Self {
629 program: CliProgram::Resolve,
630 prefix_args: Vec::new(),
631 working_directory: PathBuf::new(),
632 env: Vec::new(),
633 env_remove: Vec::new(),
634 extra_args: Vec::new(),
635 transport: Transport::default(),
636 github_token: None,
637 use_logged_in_user: None,
638 log_level: None,
639 session_idle_timeout_seconds: None,
640 on_list_models: None,
641 session_fs: None,
642 request_handler: None,
643 on_github_telemetry: None,
644 on_get_trace_context: None,
645 telemetry: None,
646 base_directory: None,
647 enable_remote_sessions: false,
648 bundled_cli_extract_dir: None,
649 mode: ClientMode::default(),
650 }
651 }
652}
653
654impl ClientOptions {
655 pub fn new() -> Self {
671 Self::default()
672 }
673
674 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
676 self.program = program.into();
677 self
678 }
679
680 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
682 where
683 I: IntoIterator<Item = S>,
684 S: Into<OsString>,
685 {
686 self.prefix_args = args.into_iter().map(Into::into).collect();
687 self
688 }
689
690 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
692 self.working_directory = cwd.into();
693 self
694 }
695
696 pub fn with_env<I, K, V>(mut self, env: I) -> Self
698 where
699 I: IntoIterator<Item = (K, V)>,
700 K: Into<OsString>,
701 V: Into<OsString>,
702 {
703 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
704 self
705 }
706
707 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
709 where
710 I: IntoIterator<Item = S>,
711 S: Into<OsString>,
712 {
713 self.env_remove = names.into_iter().map(Into::into).collect();
714 self
715 }
716
717 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
719 where
720 I: IntoIterator<Item = S>,
721 S: Into<String>,
722 {
723 self.extra_args = args.into_iter().map(Into::into).collect();
724 self
725 }
726
727 pub fn with_transport(mut self, transport: Transport) -> Self {
729 self.transport = transport;
730 self
731 }
732
733 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
736 self.github_token = Some(token.into());
737 self
738 }
739
740 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
743 self.use_logged_in_user = Some(use_logged_in);
744 self
745 }
746
747 pub fn with_log_level(mut self, level: LogLevel) -> Self {
749 self.log_level = Some(level);
750 self
751 }
752
753 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
756 self.session_idle_timeout_seconds = Some(seconds);
757 self
758 }
759
760 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
763 where
764 H: ListModelsHandler + 'static,
765 {
766 self.on_list_models = Some(Arc::new(handler));
767 self
768 }
769
770 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
772 self.session_fs = Some(config);
773 self
774 }
775
776 pub fn with_request_handler<H>(mut self, handler: H) -> Self
781 where
782 H: crate::copilot_request_handler::CopilotRequestHandler,
783 {
784 self.request_handler = Some(Arc::new(handler));
785 self
786 }
787
788 #[doc(hidden)]
794 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
795 where
796 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
797 {
798 self.on_github_telemetry = Some(Arc::new(callback));
799 self
800 }
801
802 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
806 where
807 P: TraceContextProvider + 'static,
808 {
809 self.on_get_trace_context = Some(Arc::new(provider));
810 self
811 }
812
813 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
815 self.telemetry = Some(config);
816 self
817 }
818
819 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
822 self.base_directory = Some(dir.into());
823 self
824 }
825
826 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
829 self.enable_remote_sessions = enabled;
830 self
831 }
832
833 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
843 self.bundled_cli_extract_dir = Some(dir.into());
844 self
845 }
846
847 pub fn with_mode(mut self, mode: ClientMode) -> Self {
852 self.mode = mode;
853 self
854 }
855}
856
857fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
859 if cfg.initial_cwd.trim().is_empty() {
860 return Err(Error::with_message(
861 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
862 "invalid SessionFsConfig: initial_cwd must not be empty",
863 ));
864 }
865 if cfg.session_state_path.trim().is_empty() {
866 return Err(Error::with_message(
867 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
868 "invalid SessionFsConfig: session_state_path must not be empty",
869 ));
870 }
871 Ok(())
872}
873
874fn generate_connection_token() -> String {
881 let mut bytes = [0u8; 16];
882 getrandom::getrandom(&mut bytes)
883 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
884 let mut hex = String::with_capacity(32);
885 for byte in bytes {
886 use std::fmt::Write;
887 let _ = write!(hex, "{byte:02x}");
888 }
889 hex
890}
891
892const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
897
898fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
900 let configured = options
901 .env
902 .iter()
903 .find(|(key, _)| {
904 key.to_string_lossy()
905 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
906 })
907 .map(|(_, value)| value.to_string_lossy().into_owned());
908 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
909 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
910}
911
912fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
913 match value {
914 None => Ok(Transport::Stdio),
915 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
916 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
917 Some(v) => Err(Error::with_message(
918 ErrorKind::InvalidConfig,
919 format!(
920 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
921 Expected 'inprocess', 'stdio', or unset."
922 ),
923 )),
924 }
925}
926
927#[cfg(any(feature = "bundled-in-process", test))]
928fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
929 if !matches!(&options.program, CliProgram::Resolve) {
930 return Err(Error::with_message(
931 ErrorKind::InvalidConfig,
932 "ClientOptions::program is not supported with Transport::InProcess; \
933 set COPILOT_CLI_PATH only when using an externally provisioned runtime package",
934 ));
935 }
936 if !options.extra_args.is_empty() {
937 return Err(Error::with_message(
938 ErrorKind::InvalidConfig,
939 "ClientOptions::extra_args is not supported with Transport::InProcess; \
940 use typed client options instead",
941 ));
942 }
943
944 let unsupported = if !options.working_directory.as_os_str().is_empty() {
945 Some("working_directory")
946 } else if !options.env.is_empty() {
947 Some("env")
948 } else if !options.env_remove.is_empty() {
949 Some("env_remove")
950 } else if options.telemetry.is_some() {
951 Some("telemetry")
952 } else if !options.prefix_args.is_empty() {
953 Some("prefix_args")
954 } else {
955 None
956 };
957
958 if let Some(option) = unsupported {
959 return Err(Error::with_message(
960 ErrorKind::InvalidConfig,
961 format!(
962 "ClientOptions::{option} is not supported with Transport::InProcess; \
963 configure process-global settings on the host process instead"
964 ),
965 ));
966 }
967
968 Ok(())
969}
970
971#[derive(Clone)]
976pub struct Client {
977 inner: Arc<ClientInner>,
978}
979
980impl std::fmt::Debug for Client {
981 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
982 f.debug_struct("Client")
983 .field("working_directory", &self.inner.cwd)
984 .field("pid", &self.pid())
985 .finish()
986 }
987}
988
989struct ClientInner {
990 child: parking_lot::Mutex<Option<Child>>,
991 #[cfg(feature = "bundled-in-process")]
992 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
995 rpc: JsonRpcClient,
996 cwd: PathBuf,
997 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
998 notification_tx: broadcast::Sender<JsonRpcNotification>,
999 router: router::SessionRouter,
1000 negotiated_protocol_version: OnceLock<u32>,
1001 state: parking_lot::Mutex<ConnectionState>,
1002 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
1003 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1004 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
1005 session_fs_configured: bool,
1006 session_fs_sqlite_declared: bool,
1007 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
1010 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1015 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1016 effective_connection_token: Option<String>,
1021 pub(crate) mode: ClientMode,
1024 startup_timings: OnceLock<StartupTimings>,
1028}
1029
1030impl Client {
1031 pub async fn start(options: ClientOptions) -> Result<Self> {
1044 let start_time = Instant::now();
1045 let mut timings = StartupTimings::default();
1046 let mut options = options;
1047 if matches!(options.transport, Transport::Default) {
1048 options.transport = resolve_default_transport(&options)?;
1049 }
1050 if matches!(options.transport, Transport::InProcess) {
1051 #[cfg(not(feature = "bundled-in-process"))]
1052 {
1053 return Err(Error::with_message(
1054 ErrorKind::InvalidConfig,
1055 "Transport::InProcess requires the `bundled-in-process` Cargo feature",
1056 ));
1057 }
1058 #[cfg(feature = "bundled-in-process")]
1059 validate_inprocess_options(&options)?;
1060 }
1061 if options.mode == ClientMode::Empty
1062 && options.base_directory.is_none()
1063 && options.session_fs.is_none()
1064 {
1065 return Err(Error::with_message(
1066 ErrorKind::InvalidConfig,
1067 "ClientMode::Empty requires either `base_directory` or \
1068 `session_fs` to be set (no implicit ~/.copilot fallback).",
1069 ));
1070 }
1071 if let Some(cfg) = &options.session_fs {
1072 validate_session_fs_config(cfg)?;
1073 }
1074 if matches!(options.transport, Transport::External { .. }) {
1077 if options.github_token.is_some() {
1078 return Err(Error::with_message(
1079 ErrorKind::InvalidConfig,
1080 "invalid client configuration: github_token cannot be used with \
1081 Transport::External (external server manages its own auth)",
1082 ));
1083 }
1084 if options.use_logged_in_user == Some(true) {
1085 return Err(Error::with_message(
1086 ErrorKind::InvalidConfig,
1087 "invalid client configuration: use_logged_in_user cannot be used with \
1088 Transport::External (external server manages its own auth)",
1089 ));
1090 }
1091 }
1092 match &options.transport {
1096 Transport::Tcp {
1097 connection_token: Some(t),
1098 ..
1099 }
1100 | Transport::External {
1101 connection_token: Some(t),
1102 ..
1103 } if t.is_empty() => {
1104 return Err(Error::with_message(
1105 ErrorKind::InvalidConfig,
1106 "invalid client configuration: connection_token must be a non-empty string",
1107 ));
1108 }
1109 _ => {}
1110 }
1111 let effective_connection_token: Option<String> = match &mut options.transport {
1116 Transport::Default => unreachable!("default transport resolved above"),
1117 Transport::Stdio | Transport::InProcess => None,
1118 Transport::Tcp {
1119 connection_token, ..
1120 } => Some(
1121 connection_token
1122 .get_or_insert_with(generate_connection_token)
1123 .clone(),
1124 ),
1125 Transport::External {
1126 connection_token, ..
1127 } => connection_token.clone(),
1128 };
1129 let session_fs_config = options.session_fs.clone();
1130 let request_handler = options.request_handler.clone();
1131 let session_fs_sqlite_declared = session_fs_config
1132 .as_ref()
1133 .and_then(|c| c.capabilities.as_ref())
1134 .is_some_and(|caps| caps.sqlite);
1135 let program = match &options.program {
1136 CliProgram::Path(path) => {
1137 info!(path = %path.display(), "using explicit copilot CLI path");
1138 path.clone()
1139 }
1140 CliProgram::Resolve => {
1141 let resolve_start = Instant::now();
1142 let resolved = resolve::copilot_binary_with_extract_dir(
1143 options.bundled_cli_extract_dir.as_deref(),
1144 )?;
1145 let resolve_elapsed = resolve_start.elapsed();
1146 timings.program_resolve_ms = Some(StartupTimings::millis(resolve_elapsed));
1147 debug!(
1148 elapsed_ms = resolve_elapsed.as_millis(),
1149 "Client::start CLI program resolution complete"
1150 );
1151 info!(path = %resolved.display(), "resolved copilot CLI");
1152 #[cfg(windows)]
1153 {
1154 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1155 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1156 }) {
1157 warn!(
1158 path = %resolved.display(),
1159 ext = %ext,
1160 "resolved copilot CLI is a .cmd/.bat wrapper; \
1161 this may cause console window flashes on Windows"
1162 );
1163 }
1164 }
1165 resolved
1166 }
1167 };
1168 let working_directory = {
1169 let cwd = options.working_directory.clone();
1170 if cwd.as_os_str().is_empty() {
1171 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1172 } else {
1173 cwd
1174 }
1175 };
1176
1177 let transport_setup_start = Instant::now();
1178 let client = match options.transport {
1179 Transport::Default => unreachable!("default transport resolved above"),
1180 Transport::External {
1181 ref host,
1182 port,
1183 connection_token: _,
1184 } => {
1185 info!(host = %host, port = %port, "connecting to external CLI server");
1186 let connect_start = Instant::now();
1187 let stream = TcpStream::connect((host.as_str(), port)).await?;
1188 debug!(
1189 elapsed_ms = connect_start.elapsed().as_millis(),
1190 host = %host,
1191 port,
1192 "Client::start TCP connect complete"
1193 );
1194 let (reader, writer) = tokio::io::split(stream);
1195 Self::from_transport(
1196 reader,
1197 writer,
1198 None,
1199 working_directory,
1200 options.on_list_models,
1201 session_fs_config.is_some(),
1202 session_fs_sqlite_declared,
1203 options.on_get_trace_context,
1204 options.on_github_telemetry,
1205 effective_connection_token.clone(),
1206 options.mode,
1207 )?
1208 }
1209 Transport::Tcp {
1210 port,
1211 connection_token: _,
1212 } => {
1213 let (mut child, actual_port, spawn_elapsed, port_wait_elapsed) =
1214 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1215 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1216 timings.port_wait_ms = Some(StartupTimings::millis(port_wait_elapsed));
1217 let connect_start = Instant::now();
1218 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1219 debug!(
1220 elapsed_ms = connect_start.elapsed().as_millis(),
1221 port = actual_port,
1222 "Client::start TCP connect complete"
1223 );
1224 let (reader, writer) = tokio::io::split(stream);
1225 Self::drain_stderr(&mut child);
1226 Self::from_transport(
1227 reader,
1228 writer,
1229 Some(child),
1230 working_directory,
1231 options.on_list_models,
1232 session_fs_config.is_some(),
1233 session_fs_sqlite_declared,
1234 options.on_get_trace_context,
1235 options.on_github_telemetry,
1236 effective_connection_token.clone(),
1237 options.mode,
1238 )?
1239 }
1240 Transport::Stdio => {
1241 let (mut child, spawn_elapsed) =
1242 Self::spawn_stdio(&program, &options, &working_directory)?;
1243 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1244 let stdin = child.stdin.take().expect("stdin is piped");
1245 let stdout = child.stdout.take().expect("stdout is piped");
1246 Self::drain_stderr(&mut child);
1247 Self::from_transport(
1248 stdout,
1249 stdin,
1250 Some(child),
1251 working_directory,
1252 options.on_list_models,
1253 session_fs_config.is_some(),
1254 session_fs_sqlite_declared,
1255 options.on_get_trace_context,
1256 options.on_github_telemetry,
1257 effective_connection_token.clone(),
1258 options.mode,
1259 )?
1260 }
1261 Transport::InProcess => {
1262 #[cfg(feature = "bundled-in-process")]
1263 {
1264 info!(runtime_path = %program.display(), "hosting copilot runtime in-process (FFI)");
1265 let mut environment = Vec::new();
1266 if let Some(base_directory) = &options.base_directory {
1267 let value = base_directory.to_str().ok_or_else(|| {
1268 Error::with_message(
1269 ErrorKind::InvalidConfig,
1270 "base_directory must be valid UTF-8 for Transport::InProcess",
1271 )
1272 })?;
1273 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1274 }
1275 if options.mode == ClientMode::Empty {
1276 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1277 }
1278 if let Some(github_token) = &options.github_token {
1279 environment
1280 .push(("COPILOT_SDK_AUTH_TOKEN".to_string(), github_token.clone()));
1281 }
1282 let mut args = Vec::new();
1283 args.extend(
1284 Self::log_level_args(&options)
1285 .into_iter()
1286 .map(str::to_string),
1287 );
1288 args.extend(Self::session_idle_timeout_args(&options));
1289 args.extend(Self::remote_args(&options));
1290 if options.github_token.is_some() {
1291 args.extend([
1292 "--auth-token-env".to_string(),
1293 "COPILOT_SDK_AUTH_TOKEN".to_string(),
1294 ]);
1295 }
1296 let use_logged_in_user = options
1297 .use_logged_in_user
1298 .unwrap_or(options.github_token.is_none());
1299 if !use_logged_in_user {
1300 args.push("--no-auto-login".to_string());
1301 }
1302 let host = crate::ffi::FfiHost::create(&program, environment, args)?;
1303 let (reader, writer, shared) = host.start().await?;
1304 let client = Self::from_transport(
1305 reader,
1306 writer,
1307 None,
1308 working_directory,
1309 options.on_list_models,
1310 session_fs_config.is_some(),
1311 session_fs_sqlite_declared,
1312 options.on_get_trace_context,
1313 options.on_github_telemetry,
1314 effective_connection_token.clone(),
1315 options.mode,
1316 )?;
1317 *client.inner.ffi_host.lock() = Some(shared);
1318 client
1319 }
1320 #[cfg(not(feature = "bundled-in-process"))]
1321 unreachable!("in-process feature validation returned above")
1322 }
1323 };
1324 timings.transport_setup_ms = StartupTimings::millis(transport_setup_start.elapsed());
1325 debug!(
1326 elapsed_ms = start_time.elapsed().as_millis(),
1327 "Client::start transport setup complete"
1328 );
1329 let handshake_start = Instant::now();
1330 client.verify_protocol_version().await?;
1331 timings.handshake_ms = StartupTimings::millis(handshake_start.elapsed());
1332 debug!(
1333 elapsed_ms = start_time.elapsed().as_millis(),
1334 "Client::start protocol verification complete"
1335 );
1336 if let Some(cfg) = session_fs_config {
1337 let session_fs_start = Instant::now();
1338 let capabilities = cfg.capabilities.as_ref().map(|c| {
1339 crate::generated::api_types::SessionFsSetProviderCapabilities {
1340 sqlite: Some(c.sqlite),
1341 }
1342 });
1343 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1344 capabilities,
1345 conventions: cfg.conventions.into_wire(),
1346 initial_cwd: cfg.initial_cwd,
1347 session_state_path: cfg.session_state_path,
1348 };
1349 client.rpc().session_fs().set_provider(request).await?;
1350 let session_fs_elapsed = session_fs_start.elapsed();
1351 timings.session_fs_ms = Some(StartupTimings::millis(session_fs_elapsed));
1352 debug!(
1353 elapsed_ms = session_fs_elapsed.as_millis(),
1354 "Client::start session filesystem setup complete"
1355 );
1356 }
1357 if let Some(handler) = request_handler {
1358 let llm_inference_start = Instant::now();
1359 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1360 handler,
1361 ));
1362 dispatcher.set_client(Arc::downgrade(&client.inner));
1363 let _ = client.inner.llm_inference.set(dispatcher.clone());
1364 client.inner.router.ensure_started(
1367 &client.inner.notification_tx,
1368 &client.inner.request_rx,
1369 Some(dispatcher.clone()),
1370 client.inner.on_github_telemetry.clone(),
1371 );
1372 client.rpc().llm_inference().set_provider().await?;
1373 let llm_inference_elapsed = llm_inference_start.elapsed();
1374 timings.llm_handler_ms = Some(StartupTimings::millis(llm_inference_elapsed));
1375 debug!(
1376 elapsed_ms = llm_inference_elapsed.as_millis(),
1377 "Client::start Copilot request handler registration complete"
1378 );
1379 }
1380 timings.total_ms = StartupTimings::millis(start_time.elapsed());
1381 let timings_span = tracing::debug_span!(
1384 "Client::start timings",
1385 program_resolve_ms = tracing::field::Empty,
1386 process_spawn_ms = tracing::field::Empty,
1387 port_wait_ms = tracing::field::Empty,
1388 transport_setup_ms = timings.transport_setup_ms,
1389 handshake_ms = timings.handshake_ms,
1390 session_fs_ms = tracing::field::Empty,
1391 llm_handler_ms = tracing::field::Empty,
1392 total_ms = timings.total_ms,
1393 );
1394 record_optional_millis(
1395 &timings_span,
1396 "program_resolve_ms",
1397 timings.program_resolve_ms,
1398 );
1399 record_optional_millis(&timings_span, "process_spawn_ms", timings.process_spawn_ms);
1400 record_optional_millis(&timings_span, "port_wait_ms", timings.port_wait_ms);
1401 record_optional_millis(&timings_span, "session_fs_ms", timings.session_fs_ms);
1402 record_optional_millis(&timings_span, "llm_handler_ms", timings.llm_handler_ms);
1403 timings_span.in_scope(|| debug!("Client::start timings"));
1404 let _ = client.inner.startup_timings.set(timings);
1405 debug!(
1406 elapsed_ms = start_time.elapsed().as_millis(),
1407 "Client::start complete"
1408 );
1409 Ok(client)
1410 }
1411
1412 pub fn from_streams(
1416 reader: impl AsyncRead + Unpin + Send + 'static,
1417 writer: impl AsyncWrite + Unpin + Send + 'static,
1418 cwd: PathBuf,
1419 ) -> Result<Self> {
1420 Self::from_transport(
1421 reader,
1422 writer,
1423 None,
1424 cwd,
1425 None,
1426 false,
1427 false,
1428 None,
1429 None,
1430 None,
1431 ClientMode::default(),
1432 )
1433 }
1434
1435 #[cfg(any(test, feature = "test-support"))]
1443 pub fn from_streams_with_trace_provider(
1444 reader: impl AsyncRead + Unpin + Send + 'static,
1445 writer: impl AsyncWrite + Unpin + Send + 'static,
1446 cwd: PathBuf,
1447 provider: Arc<dyn TraceContextProvider>,
1448 ) -> Result<Self> {
1449 Self::from_transport(
1450 reader,
1451 writer,
1452 None,
1453 cwd,
1454 None,
1455 false,
1456 false,
1457 Some(provider),
1458 None,
1459 None,
1460 ClientMode::default(),
1461 )
1462 }
1463
1464 #[cfg(any(test, feature = "test-support"))]
1468 pub fn from_streams_with_connection_token(
1469 reader: impl AsyncRead + Unpin + Send + 'static,
1470 writer: impl AsyncWrite + Unpin + Send + 'static,
1471 cwd: PathBuf,
1472 token: Option<String>,
1473 ) -> Result<Self> {
1474 Self::from_transport(
1475 reader,
1476 writer,
1477 None,
1478 cwd,
1479 None,
1480 false,
1481 false,
1482 None,
1483 None,
1484 token,
1485 ClientMode::default(),
1486 )
1487 }
1488
1489 #[doc(hidden)]
1492 #[cfg(any(test, feature = "test-support"))]
1493 pub fn from_streams_with_github_telemetry(
1494 reader: impl AsyncRead + Unpin + Send + 'static,
1495 writer: impl AsyncWrite + Unpin + Send + 'static,
1496 cwd: PathBuf,
1497 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1498 ) -> Result<Self> {
1499 Self::from_transport(
1500 reader,
1501 writer,
1502 None,
1503 cwd,
1504 None,
1505 false,
1506 false,
1507 None,
1508 Some(on_github_telemetry),
1509 None,
1510 ClientMode::default(),
1511 )
1512 }
1513
1514 #[cfg(any(test, feature = "test-support"))]
1520 pub fn generate_connection_token_for_test() -> String {
1521 generate_connection_token()
1522 }
1523
1524 #[allow(clippy::too_many_arguments)]
1525 fn from_transport(
1526 reader: impl AsyncRead + Unpin + Send + 'static,
1527 writer: impl AsyncWrite + Unpin + Send + 'static,
1528 child: Option<Child>,
1529 cwd: PathBuf,
1530 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1531 session_fs_configured: bool,
1532 session_fs_sqlite_declared: bool,
1533 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1534 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1535 effective_connection_token: Option<String>,
1536 mode: ClientMode,
1537 ) -> Result<Self> {
1538 let setup_start = Instant::now();
1539 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1540 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1541 let rpc = JsonRpcClient::new(
1542 writer,
1543 reader,
1544 notification_broadcast_tx.clone(),
1545 request_tx,
1546 );
1547
1548 let pid = child.as_ref().and_then(|c| c.id());
1549 info!(pid = ?pid, "copilot CLI client ready");
1550
1551 let client = Self {
1552 inner: Arc::new(ClientInner {
1553 child: parking_lot::Mutex::new(child),
1554 #[cfg(feature = "bundled-in-process")]
1555 ffi_host: parking_lot::Mutex::new(None),
1556 rpc,
1557 cwd,
1558 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1559 notification_tx: notification_broadcast_tx,
1560 router: router::SessionRouter::new(),
1561 negotiated_protocol_version: OnceLock::new(),
1562 state: parking_lot::Mutex::new(ConnectionState::Connected),
1563 lifecycle_tx: broadcast::channel(256).0,
1564 on_list_models,
1565 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1566 session_fs_configured,
1567 session_fs_sqlite_declared,
1568 llm_inference: OnceLock::new(),
1569 on_github_telemetry,
1570 on_get_trace_context,
1571 effective_connection_token,
1572 mode,
1573 startup_timings: OnceLock::new(),
1574 }),
1575 };
1576 client.spawn_lifecycle_dispatcher();
1577 debug!(
1578 elapsed_ms = setup_start.elapsed().as_millis(),
1579 pid = ?pid,
1580 "Client::from_transport setup complete"
1581 );
1582 Ok(client)
1583 }
1584
1585 fn spawn_lifecycle_dispatcher(&self) {
1589 let inner = Arc::clone(&self.inner);
1590 let mut notif_rx = inner.notification_tx.subscribe();
1591 tokio::spawn(async move {
1592 loop {
1593 match notif_rx.recv().await {
1594 Ok(notification) => {
1595 if notification.method != "session.lifecycle" {
1596 continue;
1597 }
1598 let Some(params) = notification.params.as_ref() else {
1599 continue;
1600 };
1601 let event: SessionLifecycleEvent =
1602 match serde_json::from_value(params.clone()) {
1603 Ok(e) => e,
1604 Err(e) => {
1605 warn!(
1606 error = %e,
1607 "failed to deserialize session.lifecycle notification"
1608 );
1609 continue;
1610 }
1611 };
1612 let _ = inner.lifecycle_tx.send(event);
1615 }
1616 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
1617 warn!(missed = n, "lifecycle dispatcher lagged");
1618 }
1619 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
1620 }
1621 }
1622 });
1623 }
1624
1625 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
1626 let mut command = Command::new(program);
1627 for arg in &options.prefix_args {
1628 command.arg(arg);
1629 }
1630 if let Some(token) = &options.github_token {
1633 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1634 }
1635 if let Some(telemetry) = &options.telemetry {
1638 command.env("COPILOT_OTEL_ENABLED", "true");
1639 if let Some(endpoint) = &telemetry.otlp_endpoint {
1640 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1641 }
1642 if let Some(protocol) = telemetry.otlp_protocol {
1643 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
1644 }
1645 if let Some(path) = &telemetry.file_path {
1646 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
1647 }
1648 if let Some(exporter) = telemetry.exporter_type {
1649 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
1650 }
1651 if let Some(source) = &telemetry.source_name {
1652 command.env("COPILOT_OTEL_SOURCE_NAME", source);
1653 }
1654 if let Some(capture) = telemetry.capture_content {
1655 command.env(
1656 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
1657 if capture { "true" } else { "false" },
1658 );
1659 }
1660 }
1661 if let Some(dir) = &options.base_directory {
1662 command.env("COPILOT_HOME", dir);
1663 }
1664 if options.mode == ClientMode::Empty {
1667 command.env("COPILOT_DISABLE_KEYTAR", "1");
1668 }
1669 if let Transport::Tcp {
1670 connection_token: Some(token),
1671 ..
1672 } = &options.transport
1673 {
1674 command.env("COPILOT_CONNECTION_TOKEN", token);
1675 }
1676 for (key, value) in &options.env {
1677 command.env(key, value);
1678 }
1679 for key in &options.env_remove {
1680 command.env_remove(key);
1681 }
1682 command
1683 .current_dir(working_directory)
1684 .stdout(Stdio::piped())
1685 .stderr(Stdio::piped());
1686
1687 #[cfg(windows)]
1688 {
1689 use std::os::windows::process::CommandExt;
1690 const CREATE_NO_WINDOW: u32 = 0x08000000;
1691 command.as_std_mut().creation_flags(CREATE_NO_WINDOW);
1692 }
1693
1694 command
1695 }
1696
1697 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
1705 let mut args: Vec<&'static str> = Vec::new();
1706 if options.github_token.is_some() {
1707 args.push("--auth-token-env");
1708 args.push("COPILOT_SDK_AUTH_TOKEN");
1709 }
1710 let use_logged_in = options
1711 .use_logged_in_user
1712 .unwrap_or(options.github_token.is_none());
1713 if !use_logged_in {
1714 args.push("--no-auto-login");
1715 }
1716 args
1717 }
1718
1719 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
1723 match options.session_idle_timeout_seconds {
1724 Some(secs) if secs > 0 => {
1725 vec!["--session-idle-timeout".to_string(), secs.to_string()]
1726 }
1727 _ => Vec::new(),
1728 }
1729 }
1730
1731 fn remote_args(options: &ClientOptions) -> Vec<String> {
1732 if options.enable_remote_sessions {
1733 vec!["--remote".to_string()]
1734 } else {
1735 Vec::new()
1736 }
1737 }
1738
1739 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
1740 match options.log_level {
1741 Some(level) => vec!["--log-level", level.as_str()],
1742 None => Vec::new(),
1743 }
1744 }
1745
1746 fn spawn_stdio(
1747 program: &Path,
1748 options: &ClientOptions,
1749 working_directory: &Path,
1750 ) -> Result<(Child, Duration)> {
1751 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
1752 let mut command = Self::build_command(program, options, working_directory);
1753 command
1754 .args(["--server", "--stdio", "--no-auto-update"])
1755 .args(Self::log_level_args(options))
1756 .args(Self::auth_args(options))
1757 .args(Self::session_idle_timeout_args(options))
1758 .args(Self::remote_args(options))
1759 .args(&options.extra_args)
1760 .stdin(Stdio::piped());
1761 let spawn_start = Instant::now();
1762 let child = command.spawn()?;
1763 let spawn_elapsed = spawn_start.elapsed();
1764 debug!(
1765 elapsed_ms = spawn_elapsed.as_millis(),
1766 "Client::spawn_stdio subprocess spawned"
1767 );
1768 Ok((child, spawn_elapsed))
1769 }
1770
1771 async fn spawn_tcp(
1772 program: &Path,
1773 options: &ClientOptions,
1774 working_directory: &Path,
1775 port: u16,
1776 ) -> Result<(Child, u16, Duration, Duration)> {
1777 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
1778 let mut command = Self::build_command(program, options, working_directory);
1779 command
1780 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
1781 .args(Self::log_level_args(options))
1782 .args(Self::auth_args(options))
1783 .args(Self::session_idle_timeout_args(options))
1784 .args(Self::remote_args(options))
1785 .args(&options.extra_args)
1786 .stdin(Stdio::null());
1787 let spawn_start = Instant::now();
1788 let mut child = command.spawn()?;
1789 let spawn_elapsed = spawn_start.elapsed();
1790 debug!(
1791 elapsed_ms = spawn_elapsed.as_millis(),
1792 "Client::spawn_tcp subprocess spawned"
1793 );
1794 let stdout = child.stdout.take().expect("stdout is piped");
1795
1796 let (port_tx, port_rx) = oneshot::channel::<u16>();
1797 let span = tracing::error_span!("copilot_cli_port_scan");
1798 tokio::spawn(
1799 async move {
1800 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
1802 let mut lines = BufReader::new(stdout).lines();
1803 let mut port_tx = Some(port_tx);
1804 while let Ok(Some(line)) = lines.next_line().await {
1805 debug!(line = %line, "CLI stdout");
1806 if let Some(tx) = port_tx.take() {
1807 if let Some(caps) = port_re.captures(&line)
1808 && let Some(p) =
1809 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
1810 {
1811 let _ = tx.send(p);
1812 continue;
1813 }
1814 port_tx = Some(tx);
1816 }
1817 }
1818 }
1819 .instrument(span),
1820 );
1821
1822 let port_wait_start = Instant::now();
1823 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
1824 .await
1825 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
1826 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
1827
1828 let port_wait_elapsed = port_wait_start.elapsed();
1829 debug!(
1830 elapsed_ms = port_wait_elapsed.as_millis(),
1831 port = actual_port,
1832 "Client::spawn_tcp TCP port wait complete"
1833 );
1834 info!(port = %actual_port, "CLI server listening");
1835 Ok((child, actual_port, spawn_elapsed, port_wait_elapsed))
1836 }
1837
1838 fn drain_stderr(child: &mut Child) {
1839 if let Some(stderr) = child.stderr.take() {
1840 let span = tracing::error_span!("copilot_cli");
1841 tokio::spawn(
1842 async move {
1843 let mut reader = BufReader::new(stderr).lines();
1844 while let Ok(Some(line)) = reader.next_line().await {
1845 warn!(line = %line, "CLI stderr");
1846 }
1847 }
1848 .instrument(span),
1849 );
1850 }
1851 }
1852
1853 pub fn cwd(&self) -> &PathBuf {
1855 &self.inner.cwd
1856 }
1857
1858 pub fn mode(&self) -> ClientMode {
1860 self.inner.mode
1861 }
1862
1863 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
1874 crate::generated::rpc::ClientRpc { client: self }
1875 }
1876
1877 #[allow(dead_code, reason = "convenience for future internal use")]
1879 pub(crate) async fn send_request(
1880 &self,
1881 method: &str,
1882 params: Option<serde_json::Value>,
1883 ) -> Result<JsonRpcResponse> {
1884 self.inner.rpc.send_request(method, params).await
1885 }
1886
1887 pub async fn call(
1907 &self,
1908 method: &str,
1909 params: Option<serde_json::Value>,
1910 ) -> Result<serde_json::Value> {
1911 self.call_with_inline_callback(method, params, None).await
1912 }
1913
1914 pub(crate) async fn call_with_inline_callback(
1929 &self,
1930 method: &str,
1931 params: Option<serde_json::Value>,
1932 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
1933 ) -> Result<serde_json::Value> {
1934 let session_id: Option<SessionId> = params
1935 .as_ref()
1936 .and_then(|p| p.get("sessionId"))
1937 .and_then(|v| v.as_str())
1938 .map(SessionId::from);
1939 let response = self
1940 .inner
1941 .rpc
1942 .send_request_with_inline_callback(method, params, inline_callback)
1943 .await?;
1944 if let Some(err) = response.error {
1945 if err.message.contains("Session not found") {
1946 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
1947 session_id.unwrap_or_else(|| "unknown".into()),
1948 ))
1949 .into());
1950 }
1951 return Err(Error::with_message(
1952 ErrorKind::Rpc { code: err.code },
1953 err.message,
1954 ));
1955 }
1956 Ok(response.result.unwrap_or(serde_json::Value::Null))
1957 }
1958
1959 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
1961 self.inner.rpc.write(response).await
1962 }
1963
1964 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
1966 Self { inner }
1967 }
1968
1969 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
1973 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
1974 self.inner.request_rx.lock().take()
1975 }
1976
1977 pub(crate) fn register_session(
1985 &self,
1986 session_id: &SessionId,
1987 ) -> crate::router::SessionChannels {
1988 self.inner.router.ensure_started(
1989 &self.inner.notification_tx,
1990 &self.inner.request_rx,
1991 self.inner.llm_inference.get().cloned(),
1992 self.inner.on_github_telemetry.clone(),
1993 );
1994 self.inner.router.register(session_id)
1995 }
1996
1997 pub(crate) fn unregister_session(&self, session_id: &SessionId) {
1999 self.inner.router.unregister(session_id);
2000 }
2001
2002 pub fn protocol_version(&self) -> Option<u32> {
2009 self.inner.negotiated_protocol_version.get().copied()
2010 }
2011
2012 pub fn startup_timings(&self) -> Option<StartupTimings> {
2019 self.inner.startup_timings.get().cloned()
2020 }
2021
2022 pub async fn verify_protocol_version(&self) -> Result<()> {
2046 let handshake_start = Instant::now();
2047 let mut used_fallback_ping = false;
2048 let server_version = match self.connect_handshake().await {
2052 Ok(v) => v,
2053 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2054 used_fallback_ping = true;
2055 self.ping(None).await?.protocol_version
2056 }
2057 Err(e) => return Err(e),
2058 };
2059
2060 match server_version {
2061 None => {
2062 warn!("CLI server did not report protocolVersion; skipping version check");
2063 }
2064 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2065 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2066 server: v,
2067 min: MIN_PROTOCOL_VERSION,
2068 max: SDK_PROTOCOL_VERSION,
2069 })
2070 .into());
2071 }
2072 Some(v) => {
2073 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2074 if existing != v {
2075 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2076 previous: existing,
2077 current: v,
2078 })
2079 .into());
2080 }
2081 } else {
2082 let _ = self.inner.negotiated_protocol_version.set(v);
2083 }
2084 }
2085 }
2086
2087 debug!(
2088 elapsed_ms = handshake_start.elapsed().as_millis(),
2089 protocol_version = ?server_version,
2090 used_fallback_ping,
2091 "Client::verify_protocol_version protocol handshake complete"
2092 );
2093 Ok(())
2094 }
2095
2096 async fn connect_handshake(&self) -> Result<Option<u32>> {
2103 let params = crate::generated::api_types::ConnectRequest {
2104 token: self.inner.effective_connection_token.clone(),
2105 enable_git_hub_telemetry_forwarding: self
2106 .inner
2107 .on_github_telemetry
2108 .is_some()
2109 .then_some(true),
2110 };
2111 let value = self
2112 .call(
2113 crate::generated::api_types::rpc_methods::CONNECT,
2114 Some(serde_json::to_value(params)?),
2115 )
2116 .await?;
2117 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2118 Ok(Some(u32::try_from(result.protocol_version).map_err(
2119 |_| ProtocolErrorKind::InvalidProtocolVersion {
2120 server: result.protocol_version,
2121 },
2122 )?))
2123 }
2124
2125 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2133 let params = match message {
2134 Some(m) => serde_json::json!({ "message": m }),
2135 None => serde_json::json!({}),
2136 };
2137 let value = self
2138 .call(generated::api_types::rpc_methods::PING, Some(params))
2139 .await?;
2140 Ok(serde_json::from_value(value)?)
2141 }
2142
2143 pub async fn list_sessions(
2146 &self,
2147 filter: Option<SessionListFilter>,
2148 ) -> Result<Vec<SessionMetadata>> {
2149 let params = match filter {
2150 Some(f) => serde_json::json!({ "filter": f }),
2151 None => serde_json::json!({}),
2152 };
2153 let result = self.call("session.list", Some(params)).await?;
2154 let response: ListSessionsResponse = serde_json::from_value(result)?;
2155 Ok(response.sessions)
2156 }
2157
2158 pub async fn get_session_metadata(
2176 &self,
2177 session_id: &SessionId,
2178 ) -> Result<Option<SessionMetadata>> {
2179 let result = self
2180 .call(
2181 "session.getMetadata",
2182 Some(serde_json::json!({ "sessionId": session_id })),
2183 )
2184 .await?;
2185 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2186 Ok(response.session)
2187 }
2188
2189 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2191 self.call(
2192 "session.delete",
2193 Some(serde_json::json!({ "sessionId": session_id })),
2194 )
2195 .await?;
2196 Ok(())
2197 }
2198
2199 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2215 let result = self
2216 .call("session.getLastId", Some(serde_json::json!({})))
2217 .await?;
2218 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2219 Ok(response.session_id)
2220 }
2221
2222 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2227 let result = self
2228 .call("session.getForeground", Some(serde_json::json!({})))
2229 .await?;
2230 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2231 Ok(response.session_id)
2232 }
2233
2234 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2239 self.call(
2240 "session.setForeground",
2241 Some(serde_json::json!({ "sessionId": session_id })),
2242 )
2243 .await?;
2244 Ok(())
2245 }
2246
2247 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2249 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2250 Ok(serde_json::from_value(result)?)
2251 }
2252
2253 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2255 let result = self
2256 .call("auth.getStatus", Some(serde_json::json!({})))
2257 .await?;
2258 Ok(serde_json::from_value(result)?)
2259 }
2260
2261 pub async fn list_models(&self) -> Result<Vec<Model>> {
2266 let cache = self.inner.models_cache.lock().clone();
2267 let models = cache
2268 .get_or_try_init(|| async {
2269 if let Some(handler) = &self.inner.on_list_models {
2270 handler.list_models().await
2271 } else {
2272 Ok(self.rpc().models().list().await?.models)
2273 }
2274 })
2275 .await?;
2276 Ok(models.clone())
2277 }
2278
2279 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2282 if let Some(provider) = &self.inner.on_get_trace_context {
2283 provider.get_trace_context().await
2284 } else {
2285 TraceContext::default()
2286 }
2287 }
2288
2289 pub fn pid(&self) -> Option<u32> {
2291 self.inner.child.lock().as_ref().and_then(|c| c.id())
2292 }
2293
2294 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2321 let pid = self.pid();
2322 info!(pid = ?pid, "stopping CLI process");
2323 let mut errors: Vec<Error> = Vec::new();
2324
2325 for session_id in self.inner.router.session_ids() {
2328 match self
2329 .call(
2330 "session.destroy",
2331 Some(serde_json::json!({ "sessionId": session_id })),
2332 )
2333 .await
2334 {
2335 Ok(_) => {}
2336 Err(e) => {
2337 warn!(
2338 session_id = %session_id,
2339 error = %e,
2340 "session.destroy failed during Client::stop",
2341 );
2342 errors.push(e);
2343 }
2344 }
2345 self.inner.router.unregister(&session_id);
2346 }
2347
2348 let should_shutdown_runtime = self.inner.child.lock().is_some();
2349 #[cfg(feature = "bundled-in-process")]
2350 let should_shutdown_runtime =
2351 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2352 if should_shutdown_runtime {
2353 let runtime_shutdown_start = Instant::now();
2354 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2355 .await
2356 {
2357 Ok(Ok(())) => {
2358 debug!(
2359 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2360 "Client::stop runtime shutdown complete"
2361 );
2362 }
2363 Ok(Err(e)) => {
2364 warn!(
2365 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2366 error = %e,
2367 "runtime.shutdown failed during Client::stop",
2368 );
2369 errors.push(e);
2370 }
2371 Err(_) => {
2372 let e = std::io::Error::new(
2373 std::io::ErrorKind::TimedOut,
2374 "runtime.shutdown timed out during Client::stop",
2375 );
2376 warn!(
2377 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2378 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2379 error = %e,
2380 "runtime.shutdown timed out during Client::stop",
2381 );
2382 errors.push(e.into());
2383 }
2384 }
2385 }
2386
2387 let child = self.inner.child.lock().take();
2388 *self.inner.state.lock() = ConnectionState::Disconnected;
2389 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2390 if let Some(mut child) = child {
2391 match child.try_wait() {
2392 Ok(Some(_status)) => {}
2393 Ok(None) => {
2394 if let Err(e) = child.kill().await {
2401 errors.push(e.into());
2402 }
2403 }
2404 Err(e) => errors.push(e.into()),
2405 }
2406 }
2407
2408 #[cfg(feature = "bundled-in-process")]
2411 {
2412 if let Some(host) = self.inner.ffi_host.lock().take() {
2413 self.inner.rpc.force_close();
2414 host.close();
2415 }
2416 }
2417
2418 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2419 if errors.is_empty() {
2420 Ok(())
2421 } else {
2422 Err(StopErrors(errors))
2423 }
2424 }
2425
2426 pub fn force_stop(&self) {
2456 let pid = self.pid();
2457 info!(pid = ?pid, "force-stopping CLI process");
2458 if let Some(mut child) = self.inner.child.lock().take()
2459 && let Err(e) = child.start_kill()
2460 {
2461 error!(pid = ?pid, error = %e, "failed to send kill signal");
2462 }
2463 self.inner.rpc.force_close();
2464 #[cfg(feature = "bundled-in-process")]
2465 {
2466 if let Some(host) = self.inner.ffi_host.lock().take() {
2467 host.close();
2468 }
2469 }
2470 self.inner.router.clear();
2473 *self.inner.state.lock() = ConnectionState::Disconnected;
2474 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2475 }
2476
2477 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2512 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2513 }
2514}
2515
2516impl Drop for ClientInner {
2517 fn drop(&mut self) {
2518 if let Some(ref mut child) = *self.child.lock() {
2519 let pid = child.id();
2520 if let Err(e) = child.start_kill() {
2521 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2522 } else {
2523 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2524 }
2525 }
2526 #[cfg(feature = "bundled-in-process")]
2527 {
2528 if let Some(host) = self.ffi_host.lock().take() {
2529 self.rpc.force_close();
2530 host.close();
2531 }
2532 }
2533 }
2534}
2535
2536#[cfg(test)]
2537mod tests {
2538 use super::*;
2539
2540 #[test]
2541 fn is_transport_failure_matches_request_cancelled() {
2542 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2543 assert!(err.is_transport_failure());
2544 }
2545
2546 #[test]
2547 fn is_transport_failure_matches_io_error() {
2548 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2549 assert!(err.is_transport_failure());
2550 }
2551
2552 #[test]
2553 fn is_transport_failure_rejects_rpc_error() {
2554 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2555 assert!(!err.is_transport_failure());
2556 }
2557
2558 #[test]
2559 fn is_transport_failure_rejects_session_error() {
2560 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2561 assert!(!err.is_transport_failure());
2562 }
2563
2564 #[test]
2565 fn client_options_builder_composes() {
2566 let opts = ClientOptions::new()
2567 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2568 .with_prefix_args(["node"])
2569 .with_cwd(PathBuf::from("/tmp"))
2570 .with_env([("KEY", "value")])
2571 .with_env_remove(["UNWANTED"])
2572 .with_extra_args(["--quiet"])
2573 .with_github_token("ghp_test")
2574 .with_use_logged_in_user(false)
2575 .with_log_level(LogLevel::Debug)
2576 .with_session_idle_timeout_seconds(120)
2577 .with_enable_remote_sessions(true);
2578 assert!(matches!(opts.program, CliProgram::Path(_)));
2579 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2580 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2581 assert_eq!(
2582 opts.env,
2583 vec![(
2584 std::ffi::OsString::from("KEY"),
2585 std::ffi::OsString::from("value")
2586 )]
2587 );
2588 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2589 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2590 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2591 assert_eq!(opts.use_logged_in_user, Some(false));
2592 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2593 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2594 assert!(opts.enable_remote_sessions);
2595 }
2596
2597 #[test]
2598 fn default_transport_values_resolve_without_process_state() {
2599 assert!(matches!(
2600 resolve_default_transport_value(None).unwrap(),
2601 Transport::Stdio
2602 ));
2603 assert!(matches!(
2604 resolve_default_transport_value(Some("stdio")).unwrap(),
2605 Transport::Stdio
2606 ));
2607 assert!(matches!(
2608 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2609 Transport::InProcess
2610 ));
2611 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2612 }
2613
2614 #[test]
2615 fn inprocess_rejects_process_scoped_options() {
2616 let invalid = [
2617 ClientOptions::new().with_cwd("."),
2618 ClientOptions::new().with_env([("KEY", "value")]),
2619 ClientOptions::new().with_env_remove(["KEY"]),
2620 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2621 ClientOptions::new().with_prefix_args(["index.js"]),
2622 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
2623 ClientOptions::new().with_extra_args(["--verbose"]),
2624 ];
2625
2626 for options in invalid {
2627 assert!(validate_inprocess_options(&options).is_err());
2628 }
2629 }
2630
2631 #[test]
2632 fn inprocess_allows_typed_runtime_options() {
2633 let options = ClientOptions::new()
2634 .with_base_directory("state")
2635 .with_log_level(LogLevel::Debug)
2636 .with_session_idle_timeout_seconds(10)
2637 .with_github_token("token")
2638 .with_use_logged_in_user(false)
2639 .with_enable_remote_sessions(true);
2640
2641 assert!(validate_inprocess_options(&options).is_ok());
2642 }
2643
2644 #[cfg(not(feature = "bundled-in-process"))]
2645 #[tokio::test]
2646 async fn inprocess_requires_cargo_feature() {
2647 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
2648 .await
2649 .unwrap_err();
2650
2651 assert!(error.to_string().contains("bundled-in-process"));
2652 }
2653
2654 #[test]
2655 fn is_transport_failure_rejects_other_protocol_errors() {
2656 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2657 assert!(!err.is_transport_failure());
2658 }
2659
2660 #[test]
2661 fn build_command_lets_env_remove_strip_injected_token() {
2662 let opts = ClientOptions {
2663 github_token: Some("secret".to_string()),
2664 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2665 ..Default::default()
2666 };
2667 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2668 let action = cmd
2670 .as_std()
2671 .get_envs()
2672 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2673 .map(|(_, v)| v);
2674 assert_eq!(
2675 action,
2676 Some(None),
2677 "env_remove should win over github_token"
2678 );
2679 }
2680
2681 #[test]
2682 fn build_command_lets_env_override_injected_token() {
2683 let opts = ClientOptions {
2684 github_token: Some("from-options".to_string()),
2685 env: vec![(
2686 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2687 std::ffi::OsString::from("from-env"),
2688 )],
2689 ..Default::default()
2690 };
2691 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2692 let value = cmd
2693 .as_std()
2694 .get_envs()
2695 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2696 .and_then(|(_, v)| v);
2697 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2698 }
2699
2700 #[test]
2701 fn build_command_injects_github_token_by_default() {
2702 let opts = ClientOptions {
2703 github_token: Some("just-the-token".to_string()),
2704 ..Default::default()
2705 };
2706 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2707 let value = cmd
2708 .as_std()
2709 .get_envs()
2710 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2711 .and_then(|(_, v)| v);
2712 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2713 }
2714
2715 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2716 cmd.as_std()
2717 .get_envs()
2718 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2719 .and_then(|(_, v)| v)
2720 }
2721
2722 #[test]
2723 fn telemetry_config_builder_composes() {
2724 let cfg = TelemetryConfig::new()
2725 .with_otlp_endpoint("http://collector:4318")
2726 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2727 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2728 .with_exporter_type(OtelExporterType::OtlpHttp)
2729 .with_source_name("my-app")
2730 .with_capture_content(true);
2731
2732 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
2733 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
2734 assert_eq!(
2735 cfg.file_path.as_deref(),
2736 Some(Path::new("/var/log/copilot.jsonl")),
2737 );
2738 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
2739 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
2740 assert_eq!(cfg.capture_content, Some(true));
2741 assert!(!cfg.is_empty());
2742 assert!(TelemetryConfig::new().is_empty());
2743 }
2744
2745 #[test]
2746 fn otlp_http_protocol_serde_matches_env_value() {
2747 for (protocol, wire) in [
2748 (OtlpHttpProtocol::HttpJson, "http/json"),
2749 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
2750 ] {
2751 assert_eq!(protocol.as_str(), wire);
2752
2753 let serialized = serde_json::to_string(&protocol).unwrap();
2754 assert_eq!(serialized, format!("\"{wire}\""));
2755
2756 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
2757 assert_eq!(deserialized, protocol);
2758 }
2759 }
2760
2761 #[test]
2762 fn build_command_sets_otel_env_when_telemetry_enabled() {
2763 let opts = ClientOptions {
2764 telemetry: Some(TelemetryConfig {
2765 otlp_endpoint: Some("http://collector:4318".to_string()),
2766 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
2767 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
2768 exporter_type: Some(OtelExporterType::OtlpHttp),
2769 source_name: Some("my-app".to_string()),
2770 capture_content: Some(true),
2771 }),
2772 ..Default::default()
2773 };
2774 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2775 assert_eq!(
2776 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2777 Some(std::ffi::OsStr::new("true")),
2778 );
2779 assert_eq!(
2780 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2781 Some(std::ffi::OsStr::new("http://collector:4318")),
2782 );
2783 assert_eq!(
2784 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
2785 Some(std::ffi::OsStr::new("http/protobuf")),
2786 );
2787 assert_eq!(
2788 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
2789 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
2790 );
2791 assert_eq!(
2792 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
2793 Some(std::ffi::OsStr::new("otlp-http")),
2794 );
2795 assert_eq!(
2796 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
2797 Some(std::ffi::OsStr::new("my-app")),
2798 );
2799 assert_eq!(
2800 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
2801 Some(std::ffi::OsStr::new("true")),
2802 );
2803 }
2804
2805 #[test]
2806 fn build_command_omits_otel_env_when_telemetry_none() {
2807 let opts = ClientOptions::default();
2808 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2809 for key in [
2810 "COPILOT_OTEL_ENABLED",
2811 "OTEL_EXPORTER_OTLP_ENDPOINT",
2812 "OTEL_EXPORTER_OTLP_PROTOCOL",
2813 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2814 "COPILOT_OTEL_EXPORTER_TYPE",
2815 "COPILOT_OTEL_SOURCE_NAME",
2816 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2817 ] {
2818 assert!(
2819 env_value(&cmd, key).is_none(),
2820 "expected {key} to be unset when telemetry is None",
2821 );
2822 }
2823 }
2824
2825 #[test]
2826 fn build_command_omits_unset_telemetry_fields() {
2827 let opts = ClientOptions {
2828 telemetry: Some(TelemetryConfig {
2829 otlp_endpoint: Some("http://collector:4318".to_string()),
2830 ..Default::default()
2831 }),
2832 ..Default::default()
2833 };
2834 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2835 assert_eq!(
2837 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2838 Some(std::ffi::OsStr::new("true")),
2839 );
2840 assert_eq!(
2841 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2842 Some(std::ffi::OsStr::new("http://collector:4318")),
2843 );
2844 for key in [
2846 "OTEL_EXPORTER_OTLP_PROTOCOL",
2847 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2848 "COPILOT_OTEL_EXPORTER_TYPE",
2849 "COPILOT_OTEL_SOURCE_NAME",
2850 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2851 ] {
2852 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
2853 }
2854 }
2855
2856 #[test]
2857 fn build_command_lets_user_env_override_telemetry() {
2858 let opts = ClientOptions {
2859 telemetry: Some(TelemetryConfig {
2860 otlp_endpoint: Some("http://from-config:4318".to_string()),
2861 ..Default::default()
2862 }),
2863 env: vec![(
2864 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
2865 std::ffi::OsString::from("http://from-user-env:4318"),
2866 )],
2867 ..Default::default()
2868 };
2869 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2870 assert_eq!(
2871 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2872 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
2873 "user-supplied options.env should override telemetry config",
2874 );
2875 }
2876
2877 #[test]
2878 fn build_command_sets_copilot_home_env_when_configured() {
2879 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
2880 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2881 assert_eq!(
2882 env_value(&cmd, "COPILOT_HOME"),
2883 Some(std::ffi::OsStr::new("/custom/copilot")),
2884 );
2885
2886 let opts = ClientOptions::default();
2887 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2888 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
2889 }
2890
2891 #[test]
2892 fn build_command_sets_connection_token_env_when_configured() {
2893 let opts = ClientOptions::new().with_transport(Transport::Tcp {
2894 port: 0,
2895 connection_token: Some("secret-token".to_string()),
2896 });
2897 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2898 assert_eq!(
2899 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
2900 Some(std::ffi::OsStr::new("secret-token")),
2901 );
2902
2903 let opts = ClientOptions::default();
2904 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2905 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
2906 }
2907
2908 #[tokio::test]
2909 async fn start_rejects_empty_connection_token() {
2910 let opts = ClientOptions::new()
2911 .with_transport(Transport::Tcp {
2912 port: 0,
2913 connection_token: Some(String::new()),
2914 })
2915 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
2916 let err = Client::start(opts).await.unwrap_err();
2917 assert!(
2918 matches!(err.kind(), ErrorKind::InvalidConfig),
2919 "got {err:?}"
2920 );
2921 }
2922
2923 #[tokio::test]
2924 async fn start_rejects_empty_external_connection_token() {
2925 let opts = ClientOptions::new()
2926 .with_transport(Transport::External {
2927 host: "127.0.0.1".to_string(),
2928 port: 1,
2929 connection_token: Some(String::new()),
2930 })
2931 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
2932 let err = Client::start(opts).await.unwrap_err();
2933 assert!(
2934 matches!(err.kind(), ErrorKind::InvalidConfig),
2935 "got {err:?}"
2936 );
2937 }
2938
2939 #[test]
2940 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
2941 let opts_true = ClientOptions {
2942 telemetry: Some(TelemetryConfig {
2943 capture_content: Some(true),
2944 ..Default::default()
2945 }),
2946 ..Default::default()
2947 };
2948 let opts_false = ClientOptions {
2949 telemetry: Some(TelemetryConfig {
2950 capture_content: Some(false),
2951 ..Default::default()
2952 }),
2953 ..Default::default()
2954 };
2955 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
2956 let cmd_false =
2957 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
2958 assert_eq!(
2959 env_value(
2960 &cmd_true,
2961 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
2962 ),
2963 Some(std::ffi::OsStr::new("true")),
2964 );
2965 assert_eq!(
2966 env_value(
2967 &cmd_false,
2968 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
2969 ),
2970 Some(std::ffi::OsStr::new("false")),
2971 );
2972 }
2973
2974 #[test]
2975 fn session_idle_timeout_args_are_omitted_by_default() {
2976 let opts = ClientOptions::default();
2977 assert!(Client::session_idle_timeout_args(&opts).is_empty());
2978 }
2979
2980 #[test]
2981 fn session_idle_timeout_args_omitted_for_zero() {
2982 let opts = ClientOptions {
2983 session_idle_timeout_seconds: Some(0),
2984 ..Default::default()
2985 };
2986 assert!(Client::session_idle_timeout_args(&opts).is_empty());
2987 }
2988
2989 #[test]
2990 fn session_idle_timeout_args_emit_flag_for_positive_value() {
2991 let opts = ClientOptions {
2992 session_idle_timeout_seconds: Some(300),
2993 ..Default::default()
2994 };
2995 assert_eq!(
2996 Client::session_idle_timeout_args(&opts),
2997 vec!["--session-idle-timeout".to_string(), "300".to_string()]
2998 );
2999 }
3000
3001 #[test]
3002 fn remote_args_omitted_by_default() {
3003 let opts = ClientOptions::default();
3004 assert!(Client::remote_args(&opts).is_empty());
3005 }
3006
3007 #[test]
3008 fn remote_args_emit_flag_when_enabled() {
3009 let opts = ClientOptions {
3010 enable_remote_sessions: true,
3011 ..Default::default()
3012 };
3013 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3014 }
3015
3016 #[test]
3017 fn log_level_args_omitted_when_unset() {
3018 let opts = ClientOptions::default();
3019 assert!(opts.log_level.is_none());
3020 assert!(
3021 Client::log_level_args(&opts).is_empty(),
3022 "with no caller-supplied log_level the SDK must not pass --log-level"
3023 );
3024 }
3025
3026 #[test]
3027 fn log_level_args_emit_flag_when_set() {
3028 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3029 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3030 }
3031
3032 #[test]
3033 fn log_level_str_round_trips() {
3034 for level in [
3035 LogLevel::None,
3036 LogLevel::Error,
3037 LogLevel::Warning,
3038 LogLevel::Info,
3039 LogLevel::Debug,
3040 LogLevel::All,
3041 ] {
3042 let s = level.as_str();
3043 let json = serde_json::to_string(&level).unwrap();
3044 assert_eq!(json, format!("\"{s}\""));
3045 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3046 assert_eq!(parsed, level);
3047 }
3048 }
3049
3050 #[test]
3051 fn client_options_debug_redacts_handler() {
3052 struct StubHandler;
3053 #[async_trait]
3054 impl ListModelsHandler for StubHandler {
3055 async fn list_models(&self) -> Result<Vec<Model>> {
3056 Ok(vec![])
3057 }
3058 }
3059 let opts = ClientOptions {
3060 on_list_models: Some(Arc::new(StubHandler)),
3061 github_token: Some("secret-token".into()),
3062 ..Default::default()
3063 };
3064 let debug = format!("{opts:?}");
3065 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3066 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3067 assert!(!debug.contains("secret-token"));
3068 }
3069
3070 #[tokio::test]
3071 async fn list_models_uses_on_list_models_handler_when_set() {
3072 use std::sync::atomic::{AtomicUsize, Ordering};
3073
3074 struct CountingHandler {
3075 calls: Arc<AtomicUsize>,
3076 models: Vec<Model>,
3077 }
3078 #[async_trait]
3079 impl ListModelsHandler for CountingHandler {
3080 async fn list_models(&self) -> Result<Vec<Model>> {
3081 self.calls.fetch_add(1, Ordering::SeqCst);
3082 Ok(self.models.clone())
3083 }
3084 }
3085
3086 let calls = Arc::new(AtomicUsize::new(0));
3087 let model = Model {
3088 id: "byok-gpt-4".into(),
3089 name: "BYOK GPT-4".into(),
3090 ..Default::default()
3091 };
3092 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3093 calls: Arc::clone(&calls),
3094 models: vec![model.clone()],
3095 });
3096
3097 let client = client_with_list_models_handler(handler);
3098
3099 let result = client.list_models().await.unwrap();
3100 assert_eq!(result.len(), 1);
3101 assert_eq!(result[0].id, "byok-gpt-4");
3102 assert_eq!(calls.load(Ordering::SeqCst), 1);
3103 }
3104
3105 #[tokio::test]
3106 async fn list_models_serializes_concurrent_cache_misses() {
3107 use std::sync::atomic::{AtomicUsize, Ordering};
3108
3109 struct SlowCountingHandler {
3110 calls: Arc<AtomicUsize>,
3111 models: Vec<Model>,
3112 }
3113 #[async_trait]
3114 impl ListModelsHandler for SlowCountingHandler {
3115 async fn list_models(&self) -> Result<Vec<Model>> {
3116 self.calls.fetch_add(1, Ordering::SeqCst);
3117 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3118 Ok(self.models.clone())
3119 }
3120 }
3121
3122 let calls = Arc::new(AtomicUsize::new(0));
3123 let model = Model {
3124 id: "single-flight-model".into(),
3125 name: "Single Flight Model".into(),
3126 ..Default::default()
3127 };
3128 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3129 calls: Arc::clone(&calls),
3130 models: vec![model],
3131 });
3132 let client = client_with_list_models_handler(handler);
3133
3134 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3135 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3136 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3137 assert_eq!(calls.load(Ordering::SeqCst), 1);
3138 }
3139
3140 #[tokio::test]
3141 async fn cancelled_resume_session_unregisters_pending_session() {
3142 let (client_write, _server_read) = tokio::io::duplex(8192);
3143 let (_server_write, client_read) = tokio::io::duplex(8192);
3144 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3145 assert!(client.startup_timings().is_none());
3146 let session_id = SessionId::new("resume-cancel-test");
3147 let handle = tokio::spawn({
3148 let client = client.clone();
3149 async move {
3150 client
3151 .resume_session(ResumeSessionConfig::new(session_id))
3152 .await
3153 }
3154 });
3155
3156 wait_for_pending_session_registration(&client).await;
3157 handle.abort();
3158 let _ = handle.await;
3159
3160 assert!(client.inner.router.session_ids().is_empty());
3161 client.force_stop();
3162 }
3163
3164 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3165 Client {
3166 inner: Arc::new(ClientInner {
3167 child: parking_lot::Mutex::new(None),
3168 #[cfg(feature = "bundled-in-process")]
3169 ffi_host: parking_lot::Mutex::new(None),
3170 rpc: {
3171 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3172 let (notif_tx, _notif_rx) = broadcast::channel(16);
3173 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3174 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3175 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3176 },
3177 cwd: PathBuf::from("."),
3178 request_rx: parking_lot::Mutex::new(None),
3179 notification_tx: broadcast::channel(16).0,
3180 router: router::SessionRouter::new(),
3181 negotiated_protocol_version: OnceLock::new(),
3182 state: parking_lot::Mutex::new(ConnectionState::Connected),
3183 lifecycle_tx: broadcast::channel(16).0,
3184 on_list_models: Some(handler),
3185 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3186 session_fs_configured: false,
3187 session_fs_sqlite_declared: false,
3188 llm_inference: OnceLock::new(),
3189 on_github_telemetry: None,
3190 on_get_trace_context: None,
3191 effective_connection_token: None,
3192 mode: ClientMode::default(),
3193 startup_timings: OnceLock::new(),
3194 }),
3195 }
3196 }
3197
3198 async fn wait_for_pending_session_registration(client: &Client) {
3199 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3200 while client.inner.router.session_ids().is_empty() {
3201 assert!(
3202 tokio::time::Instant::now() < deadline,
3203 "session was not registered"
3204 );
3205 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3206 }
3207 }
3208}