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 #[cfg(feature = "test-support")]
2202 #[doc(hidden)]
2203 pub fn start_router_for_test(&self) {
2204 self.inner.router.ensure_started(
2205 &self.inner.notification_tx,
2206 &self.inner.request_rx,
2207 self.inner.llm_inference.get().cloned(),
2208 self.inner.on_github_telemetry.clone(),
2209 );
2210 }
2211
2212 #[cfg(feature = "test-support")]
2213 #[doc(hidden)]
2214 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2217 let mut first_error = None;
2218
2219 for session_id in self.inner.router.session_ids() {
2220 if let Err(error) = self
2221 .call(
2222 "session.destroy",
2223 Some(serde_json::json!({ "sessionId": session_id })),
2224 )
2225 .await
2226 && first_error.is_none()
2227 {
2228 first_error = Some(error);
2229 }
2230 self.inner.router.unregister(&session_id);
2231 }
2232
2233 match self.list_sessions(None).await {
2234 Ok(sessions) => {
2235 for session in sessions {
2236 if let Err(error) = self.delete_session(&session.session_id).await
2237 && first_error.is_none()
2238 {
2239 first_error = Some(error);
2240 }
2241 }
2242 }
2243 Err(error) if first_error.is_none() => first_error = Some(error),
2244 Err(_) => {}
2245 }
2246
2247 match first_error {
2248 Some(error) => Err(error),
2249 None => Ok(()),
2250 }
2251 }
2252
2253 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2269 let result = self
2270 .call("session.getLastId", Some(serde_json::json!({})))
2271 .await?;
2272 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2273 Ok(response.session_id)
2274 }
2275
2276 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2281 let result = self
2282 .call("session.getForeground", Some(serde_json::json!({})))
2283 .await?;
2284 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2285 Ok(response.session_id)
2286 }
2287
2288 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2293 self.call(
2294 "session.setForeground",
2295 Some(serde_json::json!({ "sessionId": session_id })),
2296 )
2297 .await?;
2298 Ok(())
2299 }
2300
2301 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2303 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2304 Ok(serde_json::from_value(result)?)
2305 }
2306
2307 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2309 let result = self
2310 .call("auth.getStatus", Some(serde_json::json!({})))
2311 .await?;
2312 Ok(serde_json::from_value(result)?)
2313 }
2314
2315 pub async fn list_models(&self) -> Result<Vec<Model>> {
2320 let cache = self.inner.models_cache.lock().clone();
2321 let models = cache
2322 .get_or_try_init(|| async {
2323 if let Some(handler) = &self.inner.on_list_models {
2324 handler.list_models().await
2325 } else {
2326 Ok(self.rpc().models().list().await?.models)
2327 }
2328 })
2329 .await?;
2330 Ok(models.clone())
2331 }
2332
2333 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2336 if let Some(provider) = &self.inner.on_get_trace_context {
2337 provider.get_trace_context().await
2338 } else {
2339 TraceContext::default()
2340 }
2341 }
2342
2343 pub fn pid(&self) -> Option<u32> {
2345 self.inner.child.lock().as_ref().and_then(|c| c.id())
2346 }
2347
2348 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2375 let pid = self.pid();
2376 info!(pid = ?pid, "stopping CLI process");
2377 let mut errors: Vec<Error> = Vec::new();
2378
2379 for session_id in self.inner.router.session_ids() {
2382 match self
2383 .call(
2384 "session.destroy",
2385 Some(serde_json::json!({ "sessionId": session_id })),
2386 )
2387 .await
2388 {
2389 Ok(_) => {}
2390 Err(e) => {
2391 warn!(
2392 session_id = %session_id,
2393 error = %e,
2394 "session.destroy failed during Client::stop",
2395 );
2396 errors.push(e);
2397 }
2398 }
2399 self.inner.router.unregister(&session_id);
2400 }
2401
2402 let should_shutdown_runtime = self.inner.child.lock().is_some();
2403 #[cfg(feature = "bundled-in-process")]
2404 let should_shutdown_runtime =
2405 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2406 if should_shutdown_runtime {
2407 let runtime_shutdown_start = Instant::now();
2408 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2409 .await
2410 {
2411 Ok(Ok(())) => {
2412 debug!(
2413 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2414 "Client::stop runtime shutdown complete"
2415 );
2416 }
2417 Ok(Err(e)) => {
2418 warn!(
2419 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2420 error = %e,
2421 "runtime.shutdown failed during Client::stop",
2422 );
2423 errors.push(e);
2424 }
2425 Err(_) => {
2426 let e = std::io::Error::new(
2427 std::io::ErrorKind::TimedOut,
2428 "runtime.shutdown timed out during Client::stop",
2429 );
2430 warn!(
2431 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2432 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2433 error = %e,
2434 "runtime.shutdown timed out during Client::stop",
2435 );
2436 errors.push(e.into());
2437 }
2438 }
2439 }
2440
2441 let child = self.inner.child.lock().take();
2442 *self.inner.state.lock() = ConnectionState::Disconnected;
2443 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2444 if let Some(mut child) = child {
2445 match child.try_wait() {
2446 Ok(Some(_status)) => {}
2447 Ok(None) => {
2448 if let Err(e) = child.kill().await {
2455 errors.push(e.into());
2456 }
2457 }
2458 Err(e) => errors.push(e.into()),
2459 }
2460 }
2461
2462 #[cfg(feature = "bundled-in-process")]
2465 {
2466 if let Some(host) = self.inner.ffi_host.lock().take() {
2467 self.inner.rpc.force_close();
2468 host.close();
2469 }
2470 }
2471
2472 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2473 if errors.is_empty() {
2474 Ok(())
2475 } else {
2476 Err(StopErrors(errors))
2477 }
2478 }
2479
2480 pub fn force_stop(&self) {
2510 let pid = self.pid();
2511 info!(pid = ?pid, "force-stopping CLI process");
2512 if let Some(mut child) = self.inner.child.lock().take()
2513 && let Err(e) = child.start_kill()
2514 {
2515 error!(pid = ?pid, error = %e, "failed to send kill signal");
2516 }
2517 self.inner.rpc.force_close();
2518 #[cfg(feature = "bundled-in-process")]
2519 {
2520 if let Some(host) = self.inner.ffi_host.lock().take() {
2521 host.close();
2522 }
2523 }
2524 self.inner.router.clear();
2527 *self.inner.state.lock() = ConnectionState::Disconnected;
2528 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2529 }
2530
2531 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2566 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2567 }
2568}
2569
2570impl Drop for ClientInner {
2571 fn drop(&mut self) {
2572 if let Some(ref mut child) = *self.child.lock() {
2573 let pid = child.id();
2574 if let Err(e) = child.start_kill() {
2575 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2576 } else {
2577 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2578 }
2579 }
2580 #[cfg(feature = "bundled-in-process")]
2581 {
2582 if let Some(host) = self.ffi_host.lock().take() {
2583 self.rpc.force_close();
2584 host.close();
2585 }
2586 }
2587 }
2588}
2589
2590#[cfg(test)]
2591mod tests {
2592 use super::*;
2593
2594 #[test]
2595 fn is_transport_failure_matches_request_cancelled() {
2596 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2597 assert!(err.is_transport_failure());
2598 }
2599
2600 #[test]
2601 fn is_transport_failure_matches_io_error() {
2602 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2603 assert!(err.is_transport_failure());
2604 }
2605
2606 #[test]
2607 fn is_transport_failure_rejects_rpc_error() {
2608 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2609 assert!(!err.is_transport_failure());
2610 }
2611
2612 #[test]
2613 fn is_transport_failure_rejects_session_error() {
2614 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2615 assert!(!err.is_transport_failure());
2616 }
2617
2618 #[test]
2619 fn client_options_builder_composes() {
2620 let opts = ClientOptions::new()
2621 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2622 .with_prefix_args(["node"])
2623 .with_cwd(PathBuf::from("/tmp"))
2624 .with_env([("KEY", "value")])
2625 .with_env_remove(["UNWANTED"])
2626 .with_extra_args(["--quiet"])
2627 .with_github_token("ghp_test")
2628 .with_use_logged_in_user(false)
2629 .with_log_level(LogLevel::Debug)
2630 .with_session_idle_timeout_seconds(120)
2631 .with_enable_remote_sessions(true);
2632 assert!(matches!(opts.program, CliProgram::Path(_)));
2633 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2634 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2635 assert_eq!(
2636 opts.env,
2637 vec![(
2638 std::ffi::OsString::from("KEY"),
2639 std::ffi::OsString::from("value")
2640 )]
2641 );
2642 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2643 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2644 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2645 assert_eq!(opts.use_logged_in_user, Some(false));
2646 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2647 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2648 assert!(opts.enable_remote_sessions);
2649 }
2650
2651 #[test]
2652 fn default_transport_values_resolve_without_process_state() {
2653 assert!(matches!(
2654 resolve_default_transport_value(None).unwrap(),
2655 Transport::Stdio
2656 ));
2657 assert!(matches!(
2658 resolve_default_transport_value(Some("stdio")).unwrap(),
2659 Transport::Stdio
2660 ));
2661 assert!(matches!(
2662 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2663 Transport::InProcess
2664 ));
2665 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2666 }
2667
2668 #[test]
2669 fn inprocess_rejects_process_scoped_options() {
2670 let invalid = [
2671 ClientOptions::new().with_cwd("."),
2672 ClientOptions::new().with_env([("KEY", "value")]),
2673 ClientOptions::new().with_env_remove(["KEY"]),
2674 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2675 ClientOptions::new().with_prefix_args(["index.js"]),
2676 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
2677 ClientOptions::new().with_extra_args(["--verbose"]),
2678 ];
2679
2680 for options in invalid {
2681 assert!(validate_inprocess_options(&options).is_err());
2682 }
2683 }
2684
2685 #[test]
2686 fn inprocess_allows_typed_runtime_options() {
2687 let options = ClientOptions::new()
2688 .with_base_directory("state")
2689 .with_log_level(LogLevel::Debug)
2690 .with_session_idle_timeout_seconds(10)
2691 .with_github_token("token")
2692 .with_use_logged_in_user(false)
2693 .with_enable_remote_sessions(true);
2694
2695 assert!(validate_inprocess_options(&options).is_ok());
2696 }
2697
2698 #[cfg(not(feature = "bundled-in-process"))]
2699 #[tokio::test]
2700 async fn inprocess_requires_cargo_feature() {
2701 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
2702 .await
2703 .unwrap_err();
2704
2705 assert!(error.to_string().contains("bundled-in-process"));
2706 }
2707
2708 #[test]
2709 fn is_transport_failure_rejects_other_protocol_errors() {
2710 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2711 assert!(!err.is_transport_failure());
2712 }
2713
2714 #[test]
2715 fn build_command_lets_env_remove_strip_injected_token() {
2716 let opts = ClientOptions {
2717 github_token: Some("secret".to_string()),
2718 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2719 ..Default::default()
2720 };
2721 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2722 let action = cmd
2724 .as_std()
2725 .get_envs()
2726 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2727 .map(|(_, v)| v);
2728 assert_eq!(
2729 action,
2730 Some(None),
2731 "env_remove should win over github_token"
2732 );
2733 }
2734
2735 #[test]
2736 fn build_command_lets_env_override_injected_token() {
2737 let opts = ClientOptions {
2738 github_token: Some("from-options".to_string()),
2739 env: vec![(
2740 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2741 std::ffi::OsString::from("from-env"),
2742 )],
2743 ..Default::default()
2744 };
2745 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2746 let value = cmd
2747 .as_std()
2748 .get_envs()
2749 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2750 .and_then(|(_, v)| v);
2751 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2752 }
2753
2754 #[test]
2755 fn build_command_injects_github_token_by_default() {
2756 let opts = ClientOptions {
2757 github_token: Some("just-the-token".to_string()),
2758 ..Default::default()
2759 };
2760 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2761 let value = cmd
2762 .as_std()
2763 .get_envs()
2764 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2765 .and_then(|(_, v)| v);
2766 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2767 }
2768
2769 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2770 cmd.as_std()
2771 .get_envs()
2772 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2773 .and_then(|(_, v)| v)
2774 }
2775
2776 #[test]
2777 fn telemetry_config_builder_composes() {
2778 let cfg = TelemetryConfig::new()
2779 .with_otlp_endpoint("http://collector:4318")
2780 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2781 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2782 .with_exporter_type(OtelExporterType::OtlpHttp)
2783 .with_source_name("my-app")
2784 .with_capture_content(true);
2785
2786 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
2787 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
2788 assert_eq!(
2789 cfg.file_path.as_deref(),
2790 Some(Path::new("/var/log/copilot.jsonl")),
2791 );
2792 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
2793 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
2794 assert_eq!(cfg.capture_content, Some(true));
2795 assert!(!cfg.is_empty());
2796 assert!(TelemetryConfig::new().is_empty());
2797 }
2798
2799 #[test]
2800 fn otlp_http_protocol_serde_matches_env_value() {
2801 for (protocol, wire) in [
2802 (OtlpHttpProtocol::HttpJson, "http/json"),
2803 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
2804 ] {
2805 assert_eq!(protocol.as_str(), wire);
2806
2807 let serialized = serde_json::to_string(&protocol).unwrap();
2808 assert_eq!(serialized, format!("\"{wire}\""));
2809
2810 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
2811 assert_eq!(deserialized, protocol);
2812 }
2813 }
2814
2815 #[test]
2816 fn build_command_sets_otel_env_when_telemetry_enabled() {
2817 let opts = ClientOptions {
2818 telemetry: Some(TelemetryConfig {
2819 otlp_endpoint: Some("http://collector:4318".to_string()),
2820 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
2821 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
2822 exporter_type: Some(OtelExporterType::OtlpHttp),
2823 source_name: Some("my-app".to_string()),
2824 capture_content: Some(true),
2825 }),
2826 ..Default::default()
2827 };
2828 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2829 assert_eq!(
2830 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2831 Some(std::ffi::OsStr::new("true")),
2832 );
2833 assert_eq!(
2834 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2835 Some(std::ffi::OsStr::new("http://collector:4318")),
2836 );
2837 assert_eq!(
2838 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
2839 Some(std::ffi::OsStr::new("http/protobuf")),
2840 );
2841 assert_eq!(
2842 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
2843 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
2844 );
2845 assert_eq!(
2846 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
2847 Some(std::ffi::OsStr::new("otlp-http")),
2848 );
2849 assert_eq!(
2850 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
2851 Some(std::ffi::OsStr::new("my-app")),
2852 );
2853 assert_eq!(
2854 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
2855 Some(std::ffi::OsStr::new("true")),
2856 );
2857 }
2858
2859 #[test]
2860 fn build_command_omits_otel_env_when_telemetry_none() {
2861 let opts = ClientOptions::default();
2862 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2863 for key in [
2864 "COPILOT_OTEL_ENABLED",
2865 "OTEL_EXPORTER_OTLP_ENDPOINT",
2866 "OTEL_EXPORTER_OTLP_PROTOCOL",
2867 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2868 "COPILOT_OTEL_EXPORTER_TYPE",
2869 "COPILOT_OTEL_SOURCE_NAME",
2870 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2871 ] {
2872 assert!(
2873 env_value(&cmd, key).is_none(),
2874 "expected {key} to be unset when telemetry is None",
2875 );
2876 }
2877 }
2878
2879 #[test]
2880 fn build_command_omits_unset_telemetry_fields() {
2881 let opts = ClientOptions {
2882 telemetry: Some(TelemetryConfig {
2883 otlp_endpoint: Some("http://collector:4318".to_string()),
2884 ..Default::default()
2885 }),
2886 ..Default::default()
2887 };
2888 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2889 assert_eq!(
2891 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2892 Some(std::ffi::OsStr::new("true")),
2893 );
2894 assert_eq!(
2895 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2896 Some(std::ffi::OsStr::new("http://collector:4318")),
2897 );
2898 for key in [
2900 "OTEL_EXPORTER_OTLP_PROTOCOL",
2901 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2902 "COPILOT_OTEL_EXPORTER_TYPE",
2903 "COPILOT_OTEL_SOURCE_NAME",
2904 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2905 ] {
2906 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
2907 }
2908 }
2909
2910 #[test]
2911 fn build_command_lets_user_env_override_telemetry() {
2912 let opts = ClientOptions {
2913 telemetry: Some(TelemetryConfig {
2914 otlp_endpoint: Some("http://from-config:4318".to_string()),
2915 ..Default::default()
2916 }),
2917 env: vec![(
2918 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
2919 std::ffi::OsString::from("http://from-user-env:4318"),
2920 )],
2921 ..Default::default()
2922 };
2923 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2924 assert_eq!(
2925 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2926 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
2927 "user-supplied options.env should override telemetry config",
2928 );
2929 }
2930
2931 #[test]
2932 fn build_command_sets_copilot_home_env_when_configured() {
2933 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
2934 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2935 assert_eq!(
2936 env_value(&cmd, "COPILOT_HOME"),
2937 Some(std::ffi::OsStr::new("/custom/copilot")),
2938 );
2939
2940 let opts = ClientOptions::default();
2941 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2942 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
2943 }
2944
2945 #[test]
2946 fn build_command_sets_connection_token_env_when_configured() {
2947 let opts = ClientOptions::new().with_transport(Transport::Tcp {
2948 port: 0,
2949 connection_token: Some("secret-token".to_string()),
2950 });
2951 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2952 assert_eq!(
2953 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
2954 Some(std::ffi::OsStr::new("secret-token")),
2955 );
2956
2957 let opts = ClientOptions::default();
2958 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2959 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
2960 }
2961
2962 #[tokio::test]
2963 async fn start_rejects_empty_connection_token() {
2964 let opts = ClientOptions::new()
2965 .with_transport(Transport::Tcp {
2966 port: 0,
2967 connection_token: Some(String::new()),
2968 })
2969 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
2970 let err = Client::start(opts).await.unwrap_err();
2971 assert!(
2972 matches!(err.kind(), ErrorKind::InvalidConfig),
2973 "got {err:?}"
2974 );
2975 }
2976
2977 #[tokio::test]
2978 async fn start_rejects_empty_external_connection_token() {
2979 let opts = ClientOptions::new()
2980 .with_transport(Transport::External {
2981 host: "127.0.0.1".to_string(),
2982 port: 1,
2983 connection_token: Some(String::new()),
2984 })
2985 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
2986 let err = Client::start(opts).await.unwrap_err();
2987 assert!(
2988 matches!(err.kind(), ErrorKind::InvalidConfig),
2989 "got {err:?}"
2990 );
2991 }
2992
2993 #[test]
2994 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
2995 let opts_true = ClientOptions {
2996 telemetry: Some(TelemetryConfig {
2997 capture_content: Some(true),
2998 ..Default::default()
2999 }),
3000 ..Default::default()
3001 };
3002 let opts_false = ClientOptions {
3003 telemetry: Some(TelemetryConfig {
3004 capture_content: Some(false),
3005 ..Default::default()
3006 }),
3007 ..Default::default()
3008 };
3009 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3010 let cmd_false =
3011 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3012 assert_eq!(
3013 env_value(
3014 &cmd_true,
3015 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3016 ),
3017 Some(std::ffi::OsStr::new("true")),
3018 );
3019 assert_eq!(
3020 env_value(
3021 &cmd_false,
3022 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3023 ),
3024 Some(std::ffi::OsStr::new("false")),
3025 );
3026 }
3027
3028 #[test]
3029 fn session_idle_timeout_args_are_omitted_by_default() {
3030 let opts = ClientOptions::default();
3031 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3032 }
3033
3034 #[test]
3035 fn session_idle_timeout_args_omitted_for_zero() {
3036 let opts = ClientOptions {
3037 session_idle_timeout_seconds: Some(0),
3038 ..Default::default()
3039 };
3040 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3041 }
3042
3043 #[test]
3044 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3045 let opts = ClientOptions {
3046 session_idle_timeout_seconds: Some(300),
3047 ..Default::default()
3048 };
3049 assert_eq!(
3050 Client::session_idle_timeout_args(&opts),
3051 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3052 );
3053 }
3054
3055 #[test]
3056 fn remote_args_omitted_by_default() {
3057 let opts = ClientOptions::default();
3058 assert!(Client::remote_args(&opts).is_empty());
3059 }
3060
3061 #[test]
3062 fn remote_args_emit_flag_when_enabled() {
3063 let opts = ClientOptions {
3064 enable_remote_sessions: true,
3065 ..Default::default()
3066 };
3067 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3068 }
3069
3070 #[test]
3071 fn log_level_args_omitted_when_unset() {
3072 let opts = ClientOptions::default();
3073 assert!(opts.log_level.is_none());
3074 assert!(
3075 Client::log_level_args(&opts).is_empty(),
3076 "with no caller-supplied log_level the SDK must not pass --log-level"
3077 );
3078 }
3079
3080 #[test]
3081 fn log_level_args_emit_flag_when_set() {
3082 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3083 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3084 }
3085
3086 #[test]
3087 fn log_level_str_round_trips() {
3088 for level in [
3089 LogLevel::None,
3090 LogLevel::Error,
3091 LogLevel::Warning,
3092 LogLevel::Info,
3093 LogLevel::Debug,
3094 LogLevel::All,
3095 ] {
3096 let s = level.as_str();
3097 let json = serde_json::to_string(&level).unwrap();
3098 assert_eq!(json, format!("\"{s}\""));
3099 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3100 assert_eq!(parsed, level);
3101 }
3102 }
3103
3104 #[test]
3105 fn client_options_debug_redacts_handler() {
3106 struct StubHandler;
3107 #[async_trait]
3108 impl ListModelsHandler for StubHandler {
3109 async fn list_models(&self) -> Result<Vec<Model>> {
3110 Ok(vec![])
3111 }
3112 }
3113 let opts = ClientOptions {
3114 on_list_models: Some(Arc::new(StubHandler)),
3115 github_token: Some("secret-token".into()),
3116 ..Default::default()
3117 };
3118 let debug = format!("{opts:?}");
3119 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3120 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3121 assert!(!debug.contains("secret-token"));
3122 }
3123
3124 #[tokio::test]
3125 async fn list_models_uses_on_list_models_handler_when_set() {
3126 use std::sync::atomic::{AtomicUsize, Ordering};
3127
3128 struct CountingHandler {
3129 calls: Arc<AtomicUsize>,
3130 models: Vec<Model>,
3131 }
3132 #[async_trait]
3133 impl ListModelsHandler for CountingHandler {
3134 async fn list_models(&self) -> Result<Vec<Model>> {
3135 self.calls.fetch_add(1, Ordering::SeqCst);
3136 Ok(self.models.clone())
3137 }
3138 }
3139
3140 let calls = Arc::new(AtomicUsize::new(0));
3141 let model = Model {
3142 id: "byok-gpt-4".into(),
3143 name: "BYOK GPT-4".into(),
3144 ..Default::default()
3145 };
3146 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3147 calls: Arc::clone(&calls),
3148 models: vec![model.clone()],
3149 });
3150
3151 let client = client_with_list_models_handler(handler);
3152
3153 let result = client.list_models().await.unwrap();
3154 assert_eq!(result.len(), 1);
3155 assert_eq!(result[0].id, "byok-gpt-4");
3156 assert_eq!(calls.load(Ordering::SeqCst), 1);
3157 }
3158
3159 #[tokio::test]
3160 async fn list_models_serializes_concurrent_cache_misses() {
3161 use std::sync::atomic::{AtomicUsize, Ordering};
3162
3163 struct SlowCountingHandler {
3164 calls: Arc<AtomicUsize>,
3165 models: Vec<Model>,
3166 }
3167 #[async_trait]
3168 impl ListModelsHandler for SlowCountingHandler {
3169 async fn list_models(&self) -> Result<Vec<Model>> {
3170 self.calls.fetch_add(1, Ordering::SeqCst);
3171 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3172 Ok(self.models.clone())
3173 }
3174 }
3175
3176 let calls = Arc::new(AtomicUsize::new(0));
3177 let model = Model {
3178 id: "single-flight-model".into(),
3179 name: "Single Flight Model".into(),
3180 ..Default::default()
3181 };
3182 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3183 calls: Arc::clone(&calls),
3184 models: vec![model],
3185 });
3186 let client = client_with_list_models_handler(handler);
3187
3188 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3189 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3190 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3191 assert_eq!(calls.load(Ordering::SeqCst), 1);
3192 }
3193
3194 #[tokio::test]
3195 async fn cancelled_resume_session_unregisters_pending_session() {
3196 let (client_write, _server_read) = tokio::io::duplex(8192);
3197 let (_server_write, client_read) = tokio::io::duplex(8192);
3198 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3199 assert!(client.startup_timings().is_none());
3200 let session_id = SessionId::new("resume-cancel-test");
3201 let handle = tokio::spawn({
3202 let client = client.clone();
3203 async move {
3204 client
3205 .resume_session(ResumeSessionConfig::new(session_id))
3206 .await
3207 }
3208 });
3209
3210 wait_for_pending_session_registration(&client).await;
3211 handle.abort();
3212 let _ = handle.await;
3213
3214 assert!(client.inner.router.session_ids().is_empty());
3215 client.force_stop();
3216 }
3217
3218 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3219 Client {
3220 inner: Arc::new(ClientInner {
3221 child: parking_lot::Mutex::new(None),
3222 #[cfg(feature = "bundled-in-process")]
3223 ffi_host: parking_lot::Mutex::new(None),
3224 rpc: {
3225 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3226 let (notif_tx, _notif_rx) = broadcast::channel(16);
3227 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3228 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3229 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3230 },
3231 cwd: PathBuf::from("."),
3232 request_rx: parking_lot::Mutex::new(None),
3233 notification_tx: broadcast::channel(16).0,
3234 router: router::SessionRouter::new(),
3235 negotiated_protocol_version: OnceLock::new(),
3236 state: parking_lot::Mutex::new(ConnectionState::Connected),
3237 lifecycle_tx: broadcast::channel(16).0,
3238 on_list_models: Some(handler),
3239 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3240 session_fs_configured: false,
3241 session_fs_sqlite_declared: false,
3242 llm_inference: OnceLock::new(),
3243 on_github_telemetry: None,
3244 on_get_trace_context: None,
3245 effective_connection_token: None,
3246 mode: ClientMode::default(),
3247 startup_timings: OnceLock::new(),
3248 }),
3249 }
3250 }
3251
3252 async fn wait_for_pending_session_registration(client: &Client) {
3253 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3254 while client.inner.router.session_ids().is_empty() {
3255 assert!(
3256 tokio::time::Instant::now() < deadline,
3257 "session was not registered"
3258 );
3259 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3260 }
3261 }
3262}