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 github_token;
27pub mod handler;
29pub mod hooks;
31mod jsonrpc;
32pub mod permission;
34pub mod provider_token;
36mod provider_token_dispatch;
37pub(crate) mod resolve;
39mod router;
40pub mod session;
42pub mod session_fs;
44mod session_fs_dispatch;
45pub mod startup_timings;
47pub mod subscription;
49pub mod tool;
51pub mod trace_context;
53pub mod transforms;
55pub mod types;
57mod wire;
58
59pub mod session_events;
61
62pub mod rpc;
65
66pub(crate) mod generated;
71
72pub mod mode;
75
76use std::ffi::OsString;
77use std::path::{Path, PathBuf};
78use std::process::Stdio;
79use std::sync::{Arc, OnceLock};
80use std::time::{Duration, Instant};
81
82use async_trait::async_trait;
83pub use github_token::{
84 GitHubToken, GitHubTokenProvider, GitHubTokenProviderArgs, GitHubTokenProviderResult,
85 GitHubTokenRequestReason,
86};
87pub use indexmap::IndexMap;
91pub(crate) use jsonrpc::{
94 JsonRpcClient, JsonRpcError, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, error_codes,
95};
96pub use mode::{BUILTIN_TOOLS_ISOLATED, ClientMode, ToolSet};
97pub use provider_token::{BearerTokenError, BearerTokenProvider, ProviderTokenArgs};
98
99#[cfg(feature = "test-support")]
101pub mod test_support {
102 pub use crate::jsonrpc::{
103 JsonRpcClient, JsonRpcMessage, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse,
104 error_codes,
105 };
106}
107use serde::{Deserialize, Serialize};
108use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, BufReader};
109use tokio::net::TcpStream;
110use tokio::process::{Child, Command};
111use tokio::sync::{broadcast, mpsc, oneshot};
112use tracing::{Instrument, debug, error, info, warn};
113pub use types::*;
114
115mod sdk_protocol_version;
116pub use sdk_protocol_version::{SDK_PROTOCOL_VERSION, get_sdk_protocol_version};
117pub use startup_timings::StartupTimings;
118pub use subscription::{EventSubscription, LifecycleSubscription};
119
120const MIN_PROTOCOL_VERSION: u32 = 3;
122const RUNTIME_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
123
124fn record_optional_millis(span: &tracing::Span, field: &'static str, value: Option<u64>) {
125 match value {
126 Some(value) => {
127 span.record(field, value);
128 }
129 None => {
130 span.record(field, "None");
131 }
132 }
133}
134
135#[derive(Debug, Default)]
137#[non_exhaustive]
138pub enum Transport {
139 #[default]
142 Default,
143 Stdio,
145 InProcess,
158 Tcp {
160 port: u16,
162 connection_token: Option<String>,
166 },
167 External {
169 host: String,
171 port: u16,
173 connection_token: Option<String>,
176 },
177}
178
179#[derive(Debug, Clone, Default)]
181pub enum CliProgram {
182 #[default]
185 Resolve,
186 Path(PathBuf),
188}
189
190impl From<PathBuf> for CliProgram {
191 fn from(path: PathBuf) -> Self {
192 Self::Path(path)
193 }
194}
195
196pub const HAS_BUNDLED_CLI: bool = cfg!(has_bundled_cli);
203
204pub fn install_bundled_cli() -> Option<PathBuf> {
228 #[cfg(feature = "bundled-cli")]
229 {
230 embeddedcli::path()
231 }
232 #[cfg(not(feature = "bundled-cli"))]
233 {
234 None
235 }
236}
237
238#[non_exhaustive]
248pub struct ClientOptions {
249 pub program: CliProgram,
251 pub prefix_args: Vec<OsString>,
253 pub working_directory: PathBuf,
257 pub env: Vec<(OsString, OsString)>,
259 pub env_remove: Vec<OsString>,
261 pub extra_args: Vec<String>,
263 pub builtin_plugin_directories: Vec<PathBuf>,
268 pub transport: Transport,
270 pub github_token: Option<String>,
275 pub use_logged_in_user: Option<bool>,
279 pub log_level: Option<LogLevel>,
283 pub session_idle_timeout_seconds: Option<u64>,
289 pub on_list_models: Option<Arc<dyn ListModelsHandler>>,
297 pub session_fs: Option<SessionFsConfig>,
305 pub request_handler: Option<Arc<dyn crate::copilot_request_handler::CopilotRequestHandler>>,
314 #[doc(hidden)]
322 pub on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
323 pub on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
333 pub telemetry: Option<TelemetryConfig>,
337 pub base_directory: Option<PathBuf>,
342 pub enable_remote_sessions: bool,
348 pub bundled_cli_extract_dir: Option<PathBuf>,
367 pub mode: ClientMode,
371}
372
373impl std::fmt::Debug for ClientOptions {
374 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
375 f.debug_struct("ClientOptions")
376 .field("program", &self.program)
377 .field("prefix_args", &self.prefix_args)
378 .field("working_directory", &self.working_directory)
379 .field("env", &self.env)
380 .field("env_remove", &self.env_remove)
381 .field("extra_args", &self.extra_args)
382 .field(
383 "builtin_plugin_directories",
384 &self.builtin_plugin_directories,
385 )
386 .field("transport", &self.transport)
387 .field(
388 "github_token",
389 &self.github_token.as_ref().map(|_| "<redacted>"),
390 )
391 .field("use_logged_in_user", &self.use_logged_in_user)
392 .field("log_level", &self.log_level)
393 .field(
394 "session_idle_timeout_seconds",
395 &self.session_idle_timeout_seconds,
396 )
397 .field(
398 "on_list_models",
399 &self.on_list_models.as_ref().map(|_| "<set>"),
400 )
401 .field("session_fs", &self.session_fs)
402 .field(
403 "request_handler",
404 &self.request_handler.as_ref().map(|_| "<set>"),
405 )
406 .field(
407 "on_github_telemetry",
408 &self.on_github_telemetry.as_ref().map(|_| "<set>"),
409 )
410 .field(
411 "on_get_trace_context",
412 &self.on_get_trace_context.as_ref().map(|_| "<set>"),
413 )
414 .field("telemetry", &self.telemetry)
415 .field("base_directory", &self.base_directory)
416 .field("enable_remote_sessions", &self.enable_remote_sessions)
417 .field("bundled_cli_extract_dir", &self.bundled_cli_extract_dir)
418 .finish()
419 }
420}
421
422#[async_trait]
431pub trait ListModelsHandler: Send + Sync + 'static {
432 async fn list_models(&self) -> Result<Vec<Model>>;
434}
435
436#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
438#[serde(rename_all = "lowercase")]
439pub enum LogLevel {
440 None,
442 Error,
444 Warning,
446 Info,
448 Debug,
450 All,
452}
453
454impl LogLevel {
455 pub fn as_str(self) -> &'static str {
457 match self {
458 Self::None => "none",
459 Self::Error => "error",
460 Self::Warning => "warning",
461 Self::Info => "info",
462 Self::Debug => "debug",
463 Self::All => "all",
464 }
465 }
466}
467
468impl std::fmt::Display for LogLevel {
469 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
470 f.write_str(self.as_str())
471 }
472}
473
474#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
479#[serde(rename_all = "kebab-case")]
480#[non_exhaustive]
481pub enum OtelExporterType {
482 OtlpHttp,
485 File,
488}
489
490impl OtelExporterType {
491 pub fn as_str(self) -> &'static str {
493 match self {
494 Self::OtlpHttp => "otlp-http",
495 Self::File => "file",
496 }
497 }
498}
499
500#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
506#[non_exhaustive]
507pub enum OtlpHttpProtocol {
508 #[serde(rename = "http/json")]
510 HttpJson,
511 #[serde(rename = "http/protobuf")]
513 HttpProtobuf,
514}
515
516impl OtlpHttpProtocol {
517 pub fn as_str(self) -> &'static str {
519 match self {
520 Self::HttpJson => "http/json",
521 Self::HttpProtobuf => "http/protobuf",
522 }
523 }
524}
525
526#[derive(Debug, Clone, Default)]
561#[non_exhaustive]
562pub struct TelemetryConfig {
563 pub otlp_endpoint: Option<String>,
565 pub otlp_protocol: Option<OtlpHttpProtocol>,
567 pub file_path: Option<PathBuf>,
569 pub exporter_type: Option<OtelExporterType>,
572 pub source_name: Option<String>,
576 pub capture_content: Option<bool>,
580}
581
582impl TelemetryConfig {
583 pub fn new() -> Self {
586 Self::default()
587 }
588
589 pub fn with_otlp_endpoint(mut self, endpoint: impl Into<String>) -> Self {
591 self.otlp_endpoint = Some(endpoint.into());
592 self
593 }
594
595 pub fn with_otlp_protocol(mut self, protocol: OtlpHttpProtocol) -> Self {
597 self.otlp_protocol = Some(protocol);
598 self
599 }
600
601 pub fn with_file_path(mut self, path: impl Into<PathBuf>) -> Self {
603 self.file_path = Some(path.into());
604 self
605 }
606
607 pub fn with_exporter_type(mut self, exporter_type: OtelExporterType) -> Self {
609 self.exporter_type = Some(exporter_type);
610 self
611 }
612
613 pub fn with_source_name(mut self, source_name: impl Into<String>) -> Self {
617 self.source_name = Some(source_name.into());
618 self
619 }
620
621 pub fn with_capture_content(mut self, capture: bool) -> Self {
625 self.capture_content = Some(capture);
626 self
627 }
628
629 pub fn is_empty(&self) -> bool {
632 self.otlp_endpoint.is_none()
633 && self.otlp_protocol.is_none()
634 && self.file_path.is_none()
635 && self.exporter_type.is_none()
636 && self.source_name.is_none()
637 && self.capture_content.is_none()
638 }
639}
640
641impl Default for ClientOptions {
642 fn default() -> Self {
643 Self {
644 program: CliProgram::Resolve,
645 prefix_args: Vec::new(),
646 working_directory: PathBuf::new(),
647 env: Vec::new(),
648 env_remove: Vec::new(),
649 extra_args: Vec::new(),
650 builtin_plugin_directories: Vec::new(),
651 transport: Transport::default(),
652 github_token: None,
653 use_logged_in_user: None,
654 log_level: None,
655 session_idle_timeout_seconds: None,
656 on_list_models: None,
657 session_fs: None,
658 request_handler: None,
659 on_github_telemetry: None,
660 on_get_trace_context: None,
661 telemetry: None,
662 base_directory: None,
663 enable_remote_sessions: false,
664 bundled_cli_extract_dir: None,
665 mode: ClientMode::default(),
666 }
667 }
668}
669
670impl ClientOptions {
671 pub fn new() -> Self {
687 Self::default()
688 }
689
690 pub fn with_program(mut self, program: impl Into<CliProgram>) -> Self {
692 self.program = program.into();
693 self
694 }
695
696 pub fn with_prefix_args<I, S>(mut self, args: I) -> Self
698 where
699 I: IntoIterator<Item = S>,
700 S: Into<OsString>,
701 {
702 self.prefix_args = args.into_iter().map(Into::into).collect();
703 self
704 }
705
706 pub fn with_cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
708 self.working_directory = cwd.into();
709 self
710 }
711
712 pub fn with_env<I, K, V>(mut self, env: I) -> Self
714 where
715 I: IntoIterator<Item = (K, V)>,
716 K: Into<OsString>,
717 V: Into<OsString>,
718 {
719 self.env = env.into_iter().map(|(k, v)| (k.into(), v.into())).collect();
720 self
721 }
722
723 pub fn with_env_remove<I, S>(mut self, names: I) -> Self
725 where
726 I: IntoIterator<Item = S>,
727 S: Into<OsString>,
728 {
729 self.env_remove = names.into_iter().map(Into::into).collect();
730 self
731 }
732
733 pub fn with_extra_args<I, S>(mut self, args: I) -> Self
735 where
736 I: IntoIterator<Item = S>,
737 S: Into<String>,
738 {
739 self.extra_args = args.into_iter().map(Into::into).collect();
740 self
741 }
742
743 pub fn with_builtin_plugin_directories<I, P>(mut self, paths: I) -> Self
748 where
749 I: IntoIterator<Item = P>,
750 P: Into<PathBuf>,
751 {
752 self.builtin_plugin_directories = paths.into_iter().map(Into::into).collect();
753 self
754 }
755
756 pub fn with_transport(mut self, transport: Transport) -> Self {
758 self.transport = transport;
759 self
760 }
761
762 pub fn with_github_token(mut self, token: impl Into<String>) -> Self {
765 self.github_token = Some(token.into());
766 self
767 }
768
769 pub fn with_use_logged_in_user(mut self, use_logged_in: bool) -> Self {
772 self.use_logged_in_user = Some(use_logged_in);
773 self
774 }
775
776 pub fn with_log_level(mut self, level: LogLevel) -> Self {
778 self.log_level = Some(level);
779 self
780 }
781
782 pub fn with_session_idle_timeout_seconds(mut self, seconds: u64) -> Self {
785 self.session_idle_timeout_seconds = Some(seconds);
786 self
787 }
788
789 pub fn with_list_models_handler<H>(mut self, handler: H) -> Self
792 where
793 H: ListModelsHandler + 'static,
794 {
795 self.on_list_models = Some(Arc::new(handler));
796 self
797 }
798
799 pub fn with_session_fs(mut self, config: SessionFsConfig) -> Self {
801 self.session_fs = Some(config);
802 self
803 }
804
805 pub fn with_request_handler<H>(mut self, handler: H) -> Self
810 where
811 H: crate::copilot_request_handler::CopilotRequestHandler,
812 {
813 self.request_handler = Some(Arc::new(handler));
814 self
815 }
816
817 #[doc(hidden)]
823 pub fn with_on_github_telemetry<F>(mut self, callback: F) -> Self
824 where
825 F: Fn(crate::github_telemetry::GitHubTelemetryNotification) + Send + Sync + 'static,
826 {
827 self.on_github_telemetry = Some(Arc::new(callback));
828 self
829 }
830
831 pub fn with_trace_context_provider<P>(mut self, provider: P) -> Self
835 where
836 P: TraceContextProvider + 'static,
837 {
838 self.on_get_trace_context = Some(Arc::new(provider));
839 self
840 }
841
842 pub fn with_telemetry(mut self, config: TelemetryConfig) -> Self {
844 self.telemetry = Some(config);
845 self
846 }
847
848 pub fn with_base_directory(mut self, dir: impl Into<PathBuf>) -> Self {
851 self.base_directory = Some(dir.into());
852 self
853 }
854
855 pub fn with_enable_remote_sessions(mut self, enabled: bool) -> Self {
858 self.enable_remote_sessions = enabled;
859 self
860 }
861
862 pub fn with_bundled_cli_extract_dir(mut self, dir: impl Into<PathBuf>) -> Self {
872 self.bundled_cli_extract_dir = Some(dir.into());
873 self
874 }
875
876 pub fn with_mode(mut self, mode: ClientMode) -> Self {
881 self.mode = mode;
882 self
883 }
884}
885
886fn validate_session_fs_config(cfg: &SessionFsConfig) -> Result<()> {
888 if cfg.initial_cwd.trim().is_empty() {
889 return Err(Error::with_message(
890 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
891 "invalid SessionFsConfig: initial_cwd must not be empty",
892 ));
893 }
894 if cfg.session_state_path.trim().is_empty() {
895 return Err(Error::with_message(
896 ErrorKind::Session(SessionErrorKind::InvalidSessionFsConfig),
897 "invalid SessionFsConfig: session_state_path must not be empty",
898 ));
899 }
900 Ok(())
901}
902
903fn generate_connection_token() -> String {
910 let mut bytes = [0u8; 16];
911 getrandom::getrandom(&mut bytes)
912 .expect("OS CSPRNG (getrandom) is unavailable; cannot generate connection token");
913 let mut hex = String::with_capacity(32);
914 for byte in bytes {
915 use std::fmt::Write;
916 let _ = write!(hex, "{byte:02x}");
917 }
918 hex
919}
920
921const DEFAULT_CONNECTION_ENV_VAR: &str = "COPILOT_SDK_DEFAULT_CONNECTION";
926
927fn resolve_default_transport(options: &ClientOptions) -> Result<Transport> {
929 let configured = options
930 .env
931 .iter()
932 .find(|(key, _)| {
933 key.to_string_lossy()
934 .eq_ignore_ascii_case(DEFAULT_CONNECTION_ENV_VAR)
935 })
936 .map(|(_, value)| value.to_string_lossy().into_owned());
937 let process = std::env::var(DEFAULT_CONNECTION_ENV_VAR).ok();
938 resolve_default_transport_value(configured.as_deref().or(process.as_deref()))
939}
940
941fn resolve_default_transport_value(value: Option<&str>) -> Result<Transport> {
942 match value {
943 None => Ok(Transport::Stdio),
944 Some(v) if v.is_empty() || v.eq_ignore_ascii_case("stdio") => Ok(Transport::Stdio),
945 Some(v) if v.eq_ignore_ascii_case("inprocess") => Ok(Transport::InProcess),
946 Some(v) => Err(Error::with_message(
947 ErrorKind::InvalidConfig,
948 format!(
949 "invalid {DEFAULT_CONNECTION_ENV_VAR} value '{v}'. \
950 Expected 'inprocess', 'stdio', or unset."
951 ),
952 )),
953 }
954}
955
956#[cfg(any(feature = "bundled-in-process", test))]
957fn validate_inprocess_options(options: &ClientOptions) -> Result<()> {
958 if !matches!(&options.program, CliProgram::Resolve) {
959 return Err(Error::with_message(
960 ErrorKind::InvalidConfig,
961 "ClientOptions::program is not supported with Transport::InProcess; \
962 set COPILOT_CLI_PATH only when using an externally provisioned runtime package",
963 ));
964 }
965 if !options.extra_args.is_empty() {
966 return Err(Error::with_message(
967 ErrorKind::InvalidConfig,
968 "ClientOptions::extra_args is not supported with Transport::InProcess; \
969 use typed client options instead",
970 ));
971 }
972
973 let unsupported = if !options.working_directory.as_os_str().is_empty() {
974 Some("working_directory")
975 } else if !options.env.is_empty() {
976 Some("env")
977 } else if !options.env_remove.is_empty() {
978 Some("env_remove")
979 } else if options.telemetry.is_some() {
980 Some("telemetry")
981 } else if !options.prefix_args.is_empty() {
982 Some("prefix_args")
983 } else {
984 None
985 };
986
987 if let Some(option) = unsupported {
988 return Err(Error::with_message(
989 ErrorKind::InvalidConfig,
990 format!(
991 "ClientOptions::{option} is not supported with Transport::InProcess; \
992 configure process-global settings on the host process instead"
993 ),
994 ));
995 }
996
997 Ok(())
998}
999
1000#[derive(Clone)]
1005pub struct Client {
1006 inner: Arc<ClientInner>,
1007}
1008
1009impl std::fmt::Debug for Client {
1010 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1011 f.debug_struct("Client")
1012 .field("working_directory", &self.inner.cwd)
1013 .field("pid", &self.pid())
1014 .finish()
1015 }
1016}
1017
1018struct ClientInner {
1019 child: parking_lot::Mutex<Option<Child>>,
1020 #[cfg(feature = "bundled-in-process")]
1021 ffi_host: parking_lot::Mutex<Option<Arc<crate::ffi::FfiShared>>>,
1024 rpc: JsonRpcClient,
1025 cwd: PathBuf,
1026 request_rx: parking_lot::Mutex<Option<mpsc::UnboundedReceiver<JsonRpcRequest>>>,
1027 notification_tx: broadcast::Sender<JsonRpcNotification>,
1028 router: router::SessionRouter,
1029 github_token_registry: Arc<github_token::GitHubTokenRegistry>,
1030 negotiated_protocol_version: OnceLock<u32>,
1031 state: parking_lot::Mutex<ConnectionState>,
1032 lifecycle_tx: broadcast::Sender<SessionLifecycleEvent>,
1033 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1034 models_cache: parking_lot::Mutex<Arc<tokio::sync::OnceCell<Vec<Model>>>>,
1035 session_fs_configured: bool,
1036 session_fs_sqlite_declared: bool,
1037 llm_inference: OnceLock<Arc<copilot_request_handler::CopilotRequestDispatcher>>,
1040 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1045 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1046 effective_connection_token: Option<String>,
1051 pub(crate) mode: ClientMode,
1054 startup_timings: OnceLock<StartupTimings>,
1058}
1059
1060impl Client {
1061 pub async fn start(options: ClientOptions) -> Result<Self> {
1074 let start_time = Instant::now();
1075 let mut timings = StartupTimings::default();
1076 let mut options = options;
1077 if matches!(options.transport, Transport::Default) {
1078 options.transport = resolve_default_transport(&options)?;
1079 }
1080 if matches!(options.transport, Transport::InProcess) {
1081 #[cfg(not(feature = "bundled-in-process"))]
1082 {
1083 return Err(Error::with_message(
1084 ErrorKind::InvalidConfig,
1085 "Transport::InProcess requires the `bundled-in-process` Cargo feature",
1086 ));
1087 }
1088 #[cfg(feature = "bundled-in-process")]
1089 validate_inprocess_options(&options)?;
1090 }
1091 if options.mode == ClientMode::Empty
1092 && options.base_directory.is_none()
1093 && options.session_fs.is_none()
1094 {
1095 return Err(Error::with_message(
1096 ErrorKind::InvalidConfig,
1097 "ClientMode::Empty requires either `base_directory` or \
1098 `session_fs` to be set (no implicit ~/.copilot fallback).",
1099 ));
1100 }
1101 if let Some(cfg) = &options.session_fs {
1102 validate_session_fs_config(cfg)?;
1103 }
1104 let builtin_plugin_directories = options
1105 .builtin_plugin_directories
1106 .iter()
1107 .map(|path| {
1108 if !path.is_absolute() {
1109 return Err(Error::with_message(
1110 ErrorKind::InvalidConfig,
1111 format!(
1112 "builtin_plugin_directories must contain only absolute paths: {}",
1113 path.display()
1114 ),
1115 ));
1116 }
1117 path.to_str().map(str::to_owned).ok_or_else(|| {
1118 Error::with_message(
1119 ErrorKind::InvalidConfig,
1120 format!(
1121 "builtin_plugin_directories must contain valid UTF-8 paths: {}",
1122 path.display()
1123 ),
1124 )
1125 })
1126 })
1127 .collect::<Result<Vec<_>>>()?;
1128 if matches!(options.transport, Transport::External { .. }) {
1131 if options.github_token.is_some() {
1132 return Err(Error::with_message(
1133 ErrorKind::InvalidConfig,
1134 "invalid client configuration: github_token cannot be used with \
1135 Transport::External (external server manages its own auth)",
1136 ));
1137 }
1138 if options.use_logged_in_user == Some(true) {
1139 return Err(Error::with_message(
1140 ErrorKind::InvalidConfig,
1141 "invalid client configuration: use_logged_in_user cannot be used with \
1142 Transport::External (external server manages its own auth)",
1143 ));
1144 }
1145 }
1146 match &options.transport {
1150 Transport::Tcp {
1151 connection_token: Some(t),
1152 ..
1153 }
1154 | Transport::External {
1155 connection_token: Some(t),
1156 ..
1157 } if t.is_empty() => {
1158 return Err(Error::with_message(
1159 ErrorKind::InvalidConfig,
1160 "invalid client configuration: connection_token must be a non-empty string",
1161 ));
1162 }
1163 _ => {}
1164 }
1165 let effective_connection_token: Option<String> = match &mut options.transport {
1170 Transport::Default => unreachable!("default transport resolved above"),
1171 Transport::Stdio | Transport::InProcess => None,
1172 Transport::Tcp {
1173 connection_token, ..
1174 } => Some(
1175 connection_token
1176 .get_or_insert_with(generate_connection_token)
1177 .clone(),
1178 ),
1179 Transport::External {
1180 connection_token, ..
1181 } => connection_token.clone(),
1182 };
1183 let session_fs_config = options.session_fs.clone();
1184 let request_handler = options.request_handler.clone();
1185 let session_fs_sqlite_declared = session_fs_config
1186 .as_ref()
1187 .and_then(|c| c.capabilities.as_ref())
1188 .is_some_and(|caps| caps.sqlite);
1189 let program = match &options.program {
1190 CliProgram::Path(path) => {
1191 info!(path = %path.display(), "using explicit copilot CLI path");
1192 path.clone()
1193 }
1194 CliProgram::Resolve => {
1195 let resolve_start = Instant::now();
1196 let resolved = resolve::copilot_binary_with_extract_dir(
1197 options.bundled_cli_extract_dir.as_deref(),
1198 )?;
1199 let resolve_elapsed = resolve_start.elapsed();
1200 timings.program_resolve_ms = Some(StartupTimings::millis(resolve_elapsed));
1201 debug!(
1202 elapsed_ms = resolve_elapsed.as_millis(),
1203 "Client::start CLI program resolution complete"
1204 );
1205 info!(path = %resolved.display(), "resolved copilot CLI");
1206 #[cfg(windows)]
1207 {
1208 if let Some(ext) = resolved.extension().and_then(|e| e.to_str()).filter(|ext| {
1209 ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat")
1210 }) {
1211 warn!(
1212 path = %resolved.display(),
1213 ext = %ext,
1214 "resolved copilot CLI is a .cmd/.bat wrapper; \
1215 this may cause console window flashes on Windows"
1216 );
1217 }
1218 }
1219 resolved
1220 }
1221 };
1222 let working_directory = {
1223 let cwd = options.working_directory.clone();
1224 if cwd.as_os_str().is_empty() {
1225 std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
1226 } else {
1227 cwd
1228 }
1229 };
1230
1231 let transport_setup_start = Instant::now();
1232 let client = match options.transport {
1233 Transport::Default => unreachable!("default transport resolved above"),
1234 Transport::External {
1235 ref host,
1236 port,
1237 connection_token: _,
1238 } => {
1239 info!(host = %host, port = %port, "connecting to external CLI server");
1240 let connect_start = Instant::now();
1241 let stream = TcpStream::connect((host.as_str(), port)).await?;
1242 debug!(
1243 elapsed_ms = connect_start.elapsed().as_millis(),
1244 host = %host,
1245 port,
1246 "Client::start TCP connect complete"
1247 );
1248 let (reader, writer) = tokio::io::split(stream);
1249 Self::from_transport(
1250 reader,
1251 writer,
1252 None,
1253 working_directory,
1254 options.on_list_models,
1255 session_fs_config.is_some(),
1256 session_fs_sqlite_declared,
1257 options.on_get_trace_context,
1258 options.on_github_telemetry,
1259 effective_connection_token.clone(),
1260 options.mode,
1261 )?
1262 }
1263 Transport::Tcp {
1264 port,
1265 connection_token: _,
1266 } => {
1267 let (mut child, actual_port, spawn_elapsed, port_wait_elapsed) =
1268 Self::spawn_tcp(&program, &options, &working_directory, port).await?;
1269 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1270 timings.port_wait_ms = Some(StartupTimings::millis(port_wait_elapsed));
1271 let connect_start = Instant::now();
1272 let stream = TcpStream::connect(("127.0.0.1", actual_port)).await?;
1273 debug!(
1274 elapsed_ms = connect_start.elapsed().as_millis(),
1275 port = actual_port,
1276 "Client::start TCP connect complete"
1277 );
1278 let (reader, writer) = tokio::io::split(stream);
1279 Self::drain_stderr(&mut child);
1280 Self::from_transport(
1281 reader,
1282 writer,
1283 Some(child),
1284 working_directory,
1285 options.on_list_models,
1286 session_fs_config.is_some(),
1287 session_fs_sqlite_declared,
1288 options.on_get_trace_context,
1289 options.on_github_telemetry,
1290 effective_connection_token.clone(),
1291 options.mode,
1292 )?
1293 }
1294 Transport::Stdio => {
1295 let (mut child, spawn_elapsed) =
1296 Self::spawn_stdio(&program, &options, &working_directory)?;
1297 timings.process_spawn_ms = Some(StartupTimings::millis(spawn_elapsed));
1298 let stdin = child.stdin.take().expect("stdin is piped");
1299 let stdout = child.stdout.take().expect("stdout is piped");
1300 Self::drain_stderr(&mut child);
1301 Self::from_transport(
1302 stdout,
1303 stdin,
1304 Some(child),
1305 working_directory,
1306 options.on_list_models,
1307 session_fs_config.is_some(),
1308 session_fs_sqlite_declared,
1309 options.on_get_trace_context,
1310 options.on_github_telemetry,
1311 effective_connection_token.clone(),
1312 options.mode,
1313 )?
1314 }
1315 Transport::InProcess => {
1316 #[cfg(feature = "bundled-in-process")]
1317 {
1318 info!(runtime_path = %program.display(), "hosting copilot runtime in-process (FFI)");
1319 let mut environment = Vec::new();
1320 if let Some(base_directory) = &options.base_directory {
1321 let value = base_directory.to_str().ok_or_else(|| {
1322 Error::with_message(
1323 ErrorKind::InvalidConfig,
1324 "base_directory must be valid UTF-8 for Transport::InProcess",
1325 )
1326 })?;
1327 environment.push(("COPILOT_HOME".to_string(), value.to_string()));
1328 }
1329 if options.mode == ClientMode::Empty {
1330 environment.push(("COPILOT_DISABLE_KEYTAR".to_string(), "1".to_string()));
1331 }
1332 if let Some(github_token) = &options.github_token {
1333 environment
1334 .push(("COPILOT_SDK_AUTH_TOKEN".to_string(), github_token.clone()));
1335 }
1336 let mut args = Vec::new();
1337 args.extend(
1338 Self::log_level_args(&options)
1339 .into_iter()
1340 .map(str::to_string),
1341 );
1342 args.extend(Self::session_idle_timeout_args(&options));
1343 args.extend(Self::remote_args(&options));
1344 if options.github_token.is_some() {
1345 args.extend([
1346 "--auth-token-env".to_string(),
1347 "COPILOT_SDK_AUTH_TOKEN".to_string(),
1348 ]);
1349 }
1350 let use_logged_in_user = options
1351 .use_logged_in_user
1352 .unwrap_or(options.github_token.is_none());
1353 if !use_logged_in_user {
1354 args.push("--no-auto-login".to_string());
1355 }
1356 let host = crate::ffi::FfiHost::create(&program, environment, args)?;
1357 let (reader, writer, shared) = host.start().await?;
1358 let client = Self::from_transport(
1359 reader,
1360 writer,
1361 None,
1362 working_directory,
1363 options.on_list_models,
1364 session_fs_config.is_some(),
1365 session_fs_sqlite_declared,
1366 options.on_get_trace_context,
1367 options.on_github_telemetry,
1368 effective_connection_token.clone(),
1369 options.mode,
1370 )?;
1371 *client.inner.ffi_host.lock() = Some(shared);
1372 client
1373 }
1374 #[cfg(not(feature = "bundled-in-process"))]
1375 unreachable!("in-process feature validation returned above")
1376 }
1377 };
1378 timings.transport_setup_ms = StartupTimings::millis(transport_setup_start.elapsed());
1379 debug!(
1380 elapsed_ms = start_time.elapsed().as_millis(),
1381 "Client::start transport setup complete"
1382 );
1383 let handshake_start = Instant::now();
1384 client.verify_protocol_version().await?;
1385 timings.handshake_ms = StartupTimings::millis(handshake_start.elapsed());
1386 debug!(
1387 elapsed_ms = start_time.elapsed().as_millis(),
1388 "Client::start protocol verification complete"
1389 );
1390 if !builtin_plugin_directories.is_empty() {
1391 client
1392 .call(
1393 "plugins.builtin.set",
1394 Some(serde_json::json!({ "paths": builtin_plugin_directories })),
1395 )
1396 .await?;
1397 }
1398 if let Some(cfg) = session_fs_config {
1399 let session_fs_start = Instant::now();
1400 let capabilities = cfg.capabilities.as_ref().map(|c| {
1401 crate::generated::api_types::SessionFsSetProviderCapabilities {
1402 sqlite: Some(c.sqlite),
1403 }
1404 });
1405 let request = crate::generated::api_types::SessionFsSetProviderRequest {
1406 capabilities,
1407 conventions: cfg.conventions.into_wire(),
1408 initial_cwd: cfg.initial_cwd,
1409 session_state_path: cfg.session_state_path,
1410 };
1411 client.rpc().session_fs().set_provider(request).await?;
1412 let session_fs_elapsed = session_fs_start.elapsed();
1413 timings.session_fs_ms = Some(StartupTimings::millis(session_fs_elapsed));
1414 debug!(
1415 elapsed_ms = session_fs_elapsed.as_millis(),
1416 "Client::start session filesystem setup complete"
1417 );
1418 }
1419 if let Some(handler) = request_handler {
1420 let llm_inference_start = Instant::now();
1421 let dispatcher = Arc::new(copilot_request_handler::CopilotRequestDispatcher::new(
1422 handler,
1423 ));
1424 dispatcher.set_client(Arc::downgrade(&client.inner));
1425 let _ = client.inner.llm_inference.set(dispatcher.clone());
1426 client.inner.router.ensure_started(
1429 &client.inner.notification_tx,
1430 &client.inner.request_rx,
1431 Some(dispatcher.clone()),
1432 client.inner.on_github_telemetry.clone(),
1433 client.inner.github_token_registry.clone(),
1434 );
1435 client.rpc().llm_inference().set_provider().await?;
1436 let llm_inference_elapsed = llm_inference_start.elapsed();
1437 timings.llm_handler_ms = Some(StartupTimings::millis(llm_inference_elapsed));
1438 debug!(
1439 elapsed_ms = llm_inference_elapsed.as_millis(),
1440 "Client::start Copilot request handler registration complete"
1441 );
1442 }
1443 timings.total_ms = StartupTimings::millis(start_time.elapsed());
1444 let timings_span = tracing::debug_span!(
1447 "Client::start timings",
1448 program_resolve_ms = tracing::field::Empty,
1449 process_spawn_ms = tracing::field::Empty,
1450 port_wait_ms = tracing::field::Empty,
1451 transport_setup_ms = timings.transport_setup_ms,
1452 handshake_ms = timings.handshake_ms,
1453 session_fs_ms = tracing::field::Empty,
1454 llm_handler_ms = tracing::field::Empty,
1455 total_ms = timings.total_ms,
1456 );
1457 record_optional_millis(
1458 &timings_span,
1459 "program_resolve_ms",
1460 timings.program_resolve_ms,
1461 );
1462 record_optional_millis(&timings_span, "process_spawn_ms", timings.process_spawn_ms);
1463 record_optional_millis(&timings_span, "port_wait_ms", timings.port_wait_ms);
1464 record_optional_millis(&timings_span, "session_fs_ms", timings.session_fs_ms);
1465 record_optional_millis(&timings_span, "llm_handler_ms", timings.llm_handler_ms);
1466 timings_span.in_scope(|| debug!("Client::start timings"));
1467 let _ = client.inner.startup_timings.set(timings);
1468 debug!(
1469 elapsed_ms = start_time.elapsed().as_millis(),
1470 "Client::start complete"
1471 );
1472 Ok(client)
1473 }
1474
1475 pub fn from_streams(
1479 reader: impl AsyncRead + Unpin + Send + 'static,
1480 writer: impl AsyncWrite + Unpin + Send + 'static,
1481 cwd: PathBuf,
1482 ) -> Result<Self> {
1483 Self::from_transport(
1484 reader,
1485 writer,
1486 None,
1487 cwd,
1488 None,
1489 false,
1490 false,
1491 None,
1492 None,
1493 None,
1494 ClientMode::default(),
1495 )
1496 }
1497
1498 #[cfg(any(test, feature = "test-support"))]
1506 pub fn from_streams_with_trace_provider(
1507 reader: impl AsyncRead + Unpin + Send + 'static,
1508 writer: impl AsyncWrite + Unpin + Send + 'static,
1509 cwd: PathBuf,
1510 provider: Arc<dyn TraceContextProvider>,
1511 ) -> Result<Self> {
1512 Self::from_transport(
1513 reader,
1514 writer,
1515 None,
1516 cwd,
1517 None,
1518 false,
1519 false,
1520 Some(provider),
1521 None,
1522 None,
1523 ClientMode::default(),
1524 )
1525 }
1526
1527 #[cfg(any(test, feature = "test-support"))]
1531 pub fn from_streams_with_connection_token(
1532 reader: impl AsyncRead + Unpin + Send + 'static,
1533 writer: impl AsyncWrite + Unpin + Send + 'static,
1534 cwd: PathBuf,
1535 token: Option<String>,
1536 ) -> Result<Self> {
1537 Self::from_transport(
1538 reader,
1539 writer,
1540 None,
1541 cwd,
1542 None,
1543 false,
1544 false,
1545 None,
1546 None,
1547 token,
1548 ClientMode::default(),
1549 )
1550 }
1551
1552 #[doc(hidden)]
1555 #[cfg(any(test, feature = "test-support"))]
1556 pub fn from_streams_with_github_telemetry(
1557 reader: impl AsyncRead + Unpin + Send + 'static,
1558 writer: impl AsyncWrite + Unpin + Send + 'static,
1559 cwd: PathBuf,
1560 on_github_telemetry: crate::github_telemetry::GitHubTelemetryCallback,
1561 ) -> Result<Self> {
1562 Self::from_transport(
1563 reader,
1564 writer,
1565 None,
1566 cwd,
1567 None,
1568 false,
1569 false,
1570 None,
1571 Some(on_github_telemetry),
1572 None,
1573 ClientMode::default(),
1574 )
1575 }
1576
1577 #[cfg(any(test, feature = "test-support"))]
1583 pub fn generate_connection_token_for_test() -> String {
1584 generate_connection_token()
1585 }
1586
1587 #[allow(clippy::too_many_arguments)]
1588 fn from_transport(
1589 reader: impl AsyncRead + Unpin + Send + 'static,
1590 writer: impl AsyncWrite + Unpin + Send + 'static,
1591 child: Option<Child>,
1592 cwd: PathBuf,
1593 on_list_models: Option<Arc<dyn ListModelsHandler>>,
1594 session_fs_configured: bool,
1595 session_fs_sqlite_declared: bool,
1596 on_get_trace_context: Option<Arc<dyn TraceContextProvider>>,
1597 on_github_telemetry: Option<crate::github_telemetry::GitHubTelemetryCallback>,
1598 effective_connection_token: Option<String>,
1599 mode: ClientMode,
1600 ) -> Result<Self> {
1601 let setup_start = Instant::now();
1602 let (request_tx, request_rx) = mpsc::unbounded_channel::<JsonRpcRequest>();
1603 let (notification_broadcast_tx, _) = broadcast::channel::<JsonRpcNotification>(1024);
1604 let rpc = JsonRpcClient::new(
1605 writer,
1606 reader,
1607 notification_broadcast_tx.clone(),
1608 request_tx,
1609 );
1610
1611 let pid = child.as_ref().and_then(|c| c.id());
1612 info!(pid = ?pid, "copilot CLI client ready");
1613
1614 let github_token_registry = Arc::new(github_token::GitHubTokenRegistry::new());
1615 let client = Self {
1616 inner: Arc::new(ClientInner {
1617 child: parking_lot::Mutex::new(child),
1618 #[cfg(feature = "bundled-in-process")]
1619 ffi_host: parking_lot::Mutex::new(None),
1620 rpc,
1621 cwd,
1622 request_rx: parking_lot::Mutex::new(Some(request_rx)),
1623 notification_tx: notification_broadcast_tx,
1624 router: router::SessionRouter::new(),
1625 github_token_registry: github_token_registry.clone(),
1626 negotiated_protocol_version: OnceLock::new(),
1627 state: parking_lot::Mutex::new(ConnectionState::Connected),
1628 lifecycle_tx: broadcast::channel(256).0,
1629 on_list_models,
1630 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
1631 session_fs_configured,
1632 session_fs_sqlite_declared,
1633 llm_inference: OnceLock::new(),
1634 on_github_telemetry,
1635 on_get_trace_context,
1636 effective_connection_token,
1637 mode,
1638 startup_timings: OnceLock::new(),
1639 }),
1640 };
1641 github_token_registry.set_client(Arc::downgrade(&client.inner));
1642 client.spawn_lifecycle_dispatcher();
1643 debug!(
1644 elapsed_ms = setup_start.elapsed().as_millis(),
1645 pid = ?pid,
1646 "Client::from_transport setup complete"
1647 );
1648 Ok(client)
1649 }
1650
1651 fn spawn_lifecycle_dispatcher(&self) {
1655 let mut notif_rx = self.inner.notification_tx.subscribe();
1656 let lifecycle_tx = self.inner.lifecycle_tx.clone();
1657 tokio::spawn(async move {
1658 loop {
1659 match notif_rx.recv().await {
1660 Ok(notification) => {
1661 if notification.method != "session.lifecycle" {
1662 continue;
1663 }
1664 let Some(params) = notification.params.as_ref() else {
1665 continue;
1666 };
1667 let event: SessionLifecycleEvent =
1668 match serde_json::from_value(params.clone()) {
1669 Ok(e) => e,
1670 Err(e) => {
1671 warn!(
1672 error = %e,
1673 "failed to deserialize session.lifecycle notification"
1674 );
1675 continue;
1676 }
1677 };
1678 let _ = lifecycle_tx.send(event);
1681 }
1682 Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
1683 warn!(missed = n, "lifecycle dispatcher lagged");
1684 }
1685 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
1686 }
1687 }
1688 });
1689 }
1690
1691 fn build_command(program: &Path, options: &ClientOptions, working_directory: &Path) -> Command {
1692 let mut command = Command::new(program);
1693 command.kill_on_drop(true);
1694 for arg in &options.prefix_args {
1695 command.arg(arg);
1696 }
1697 if let Some(token) = &options.github_token {
1700 command.env("COPILOT_SDK_AUTH_TOKEN", token);
1701 }
1702 if let Some(telemetry) = &options.telemetry {
1705 command.env("COPILOT_OTEL_ENABLED", "true");
1706 if let Some(endpoint) = &telemetry.otlp_endpoint {
1707 command.env("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
1708 }
1709 if let Some(protocol) = telemetry.otlp_protocol {
1710 command.env("OTEL_EXPORTER_OTLP_PROTOCOL", protocol.as_str());
1711 }
1712 if let Some(path) = &telemetry.file_path {
1713 command.env("COPILOT_OTEL_FILE_EXPORTER_PATH", path);
1714 }
1715 if let Some(exporter) = telemetry.exporter_type {
1716 command.env("COPILOT_OTEL_EXPORTER_TYPE", exporter.as_str());
1717 }
1718 if let Some(source) = &telemetry.source_name {
1719 command.env("COPILOT_OTEL_SOURCE_NAME", source);
1720 }
1721 if let Some(capture) = telemetry.capture_content {
1722 command.env(
1723 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
1724 if capture { "true" } else { "false" },
1725 );
1726 }
1727 }
1728 if let Some(dir) = &options.base_directory {
1729 command.env("COPILOT_HOME", dir);
1730 }
1731 if options.mode == ClientMode::Empty {
1734 command.env("COPILOT_DISABLE_KEYTAR", "1");
1735 }
1736 if let Transport::Tcp {
1737 connection_token: Some(token),
1738 ..
1739 } = &options.transport
1740 {
1741 command.env("COPILOT_CONNECTION_TOKEN", token);
1742 }
1743 for (key, value) in &options.env {
1744 command.env(key, value);
1745 }
1746 for key in &options.env_remove {
1747 command.env_remove(key);
1748 }
1749 command
1750 .current_dir(working_directory)
1751 .stdout(Stdio::piped())
1752 .stderr(Stdio::piped());
1753
1754 #[cfg(windows)]
1755 {
1756 use std::os::windows::process::CommandExt;
1757 const CREATE_NO_WINDOW: u32 = 0x08000000;
1758 command.as_std_mut().creation_flags(CREATE_NO_WINDOW);
1759 }
1760
1761 command
1762 }
1763
1764 fn auth_args(options: &ClientOptions) -> Vec<&'static str> {
1772 let mut args: Vec<&'static str> = Vec::new();
1773 if options.github_token.is_some() {
1774 args.push("--auth-token-env");
1775 args.push("COPILOT_SDK_AUTH_TOKEN");
1776 }
1777 let use_logged_in = options
1778 .use_logged_in_user
1779 .unwrap_or(options.github_token.is_none());
1780 if !use_logged_in {
1781 args.push("--no-auto-login");
1782 }
1783 args
1784 }
1785
1786 fn session_idle_timeout_args(options: &ClientOptions) -> Vec<String> {
1790 match options.session_idle_timeout_seconds {
1791 Some(secs) if secs > 0 => {
1792 vec!["--session-idle-timeout".to_string(), secs.to_string()]
1793 }
1794 _ => Vec::new(),
1795 }
1796 }
1797
1798 fn remote_args(options: &ClientOptions) -> Vec<String> {
1799 if options.enable_remote_sessions {
1800 vec!["--remote".to_string()]
1801 } else {
1802 Vec::new()
1803 }
1804 }
1805
1806 fn log_level_args(options: &ClientOptions) -> Vec<&'static str> {
1807 match options.log_level {
1808 Some(level) => vec!["--log-level", level.as_str()],
1809 None => Vec::new(),
1810 }
1811 }
1812
1813 fn spawn_stdio(
1814 program: &Path,
1815 options: &ClientOptions,
1816 working_directory: &Path,
1817 ) -> Result<(Child, Duration)> {
1818 info!(cwd = ?working_directory, program = %program.display(), "spawning copilot CLI (stdio)");
1819 let mut command = Self::build_command(program, options, working_directory);
1820 command
1821 .args(["--server", "--stdio", "--no-auto-update"])
1822 .args(Self::log_level_args(options))
1823 .args(Self::auth_args(options))
1824 .args(Self::session_idle_timeout_args(options))
1825 .args(Self::remote_args(options))
1826 .args(&options.extra_args)
1827 .stdin(Stdio::piped());
1828 let spawn_start = Instant::now();
1829 let child = command.spawn()?;
1830 let spawn_elapsed = spawn_start.elapsed();
1831 debug!(
1832 elapsed_ms = spawn_elapsed.as_millis(),
1833 "Client::spawn_stdio subprocess spawned"
1834 );
1835 Ok((child, spawn_elapsed))
1836 }
1837
1838 async fn spawn_tcp(
1839 program: &Path,
1840 options: &ClientOptions,
1841 working_directory: &Path,
1842 port: u16,
1843 ) -> Result<(Child, u16, Duration, Duration)> {
1844 info!(cwd = ?working_directory, program = %program.display(), port = %port, "spawning copilot CLI (tcp)");
1845 let mut command = Self::build_command(program, options, working_directory);
1846 command
1847 .args(["--server", "--port", &port.to_string(), "--no-auto-update"])
1848 .args(Self::log_level_args(options))
1849 .args(Self::auth_args(options))
1850 .args(Self::session_idle_timeout_args(options))
1851 .args(Self::remote_args(options))
1852 .args(&options.extra_args)
1853 .stdin(Stdio::null());
1854 let spawn_start = Instant::now();
1855 let mut child = command.spawn()?;
1856 let spawn_elapsed = spawn_start.elapsed();
1857 debug!(
1858 elapsed_ms = spawn_elapsed.as_millis(),
1859 "Client::spawn_tcp subprocess spawned"
1860 );
1861 let stdout = child.stdout.take().expect("stdout is piped");
1862
1863 let (port_tx, port_rx) = oneshot::channel::<u16>();
1864 let span = tracing::error_span!("copilot_cli_port_scan");
1865 tokio::spawn(
1866 async move {
1867 let port_re = regex::Regex::new(r"listening on port (\d+)").expect("valid regex");
1869 let mut lines = BufReader::new(stdout).lines();
1870 let mut port_tx = Some(port_tx);
1871 while let Ok(Some(line)) = lines.next_line().await {
1872 debug!(line = %line, "CLI stdout");
1873 if let Some(tx) = port_tx.take() {
1874 if let Some(caps) = port_re.captures(&line)
1875 && let Some(p) =
1876 caps.get(1).and_then(|m| m.as_str().parse::<u16>().ok())
1877 {
1878 let _ = tx.send(p);
1879 continue;
1880 }
1881 port_tx = Some(tx);
1883 }
1884 }
1885 }
1886 .instrument(span),
1887 );
1888
1889 let port_wait_start = Instant::now();
1890 let actual_port = tokio::time::timeout(std::time::Duration::from_secs(10), port_rx)
1891 .await
1892 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout)))?
1893 .map_err(|_| Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupFailed)))?;
1894
1895 let port_wait_elapsed = port_wait_start.elapsed();
1896 debug!(
1897 elapsed_ms = port_wait_elapsed.as_millis(),
1898 port = actual_port,
1899 "Client::spawn_tcp TCP port wait complete"
1900 );
1901 info!(port = %actual_port, "CLI server listening");
1902 Ok((child, actual_port, spawn_elapsed, port_wait_elapsed))
1903 }
1904
1905 fn drain_stderr(child: &mut Child) {
1906 if let Some(stderr) = child.stderr.take() {
1907 let span = tracing::error_span!("copilot_cli");
1908 tokio::spawn(
1909 async move {
1910 let mut reader = BufReader::new(stderr).lines();
1911 while let Ok(Some(line)) = reader.next_line().await {
1912 warn!(line = %line, "CLI stderr");
1913 }
1914 }
1915 .instrument(span),
1916 );
1917 }
1918 }
1919
1920 pub fn cwd(&self) -> &PathBuf {
1922 &self.inner.cwd
1923 }
1924
1925 pub fn mode(&self) -> ClientMode {
1927 self.inner.mode
1928 }
1929
1930 pub fn rpc(&self) -> crate::generated::rpc::ClientRpc<'_> {
1941 crate::generated::rpc::ClientRpc { client: self }
1942 }
1943
1944 #[allow(dead_code, reason = "convenience for future internal use")]
1946 pub(crate) async fn send_request(
1947 &self,
1948 method: &str,
1949 params: Option<serde_json::Value>,
1950 ) -> Result<JsonRpcResponse> {
1951 self.inner.rpc.send_request(method, params).await
1952 }
1953
1954 pub async fn call(
1974 &self,
1975 method: &str,
1976 params: Option<serde_json::Value>,
1977 ) -> Result<serde_json::Value> {
1978 self.call_with_inline_callback(method, params, None).await
1979 }
1980
1981 pub(crate) async fn call_with_inline_callback(
1996 &self,
1997 method: &str,
1998 params: Option<serde_json::Value>,
1999 inline_callback: Option<crate::jsonrpc::InlineResponseCallback>,
2000 ) -> Result<serde_json::Value> {
2001 let session_id: Option<SessionId> = params
2002 .as_ref()
2003 .and_then(|p| p.get("sessionId"))
2004 .and_then(|v| v.as_str())
2005 .map(SessionId::from);
2006 let response = self
2007 .inner
2008 .rpc
2009 .send_request_with_inline_callback(method, params, inline_callback)
2010 .await?;
2011 if let Some(err) = response.error {
2012 if err.message.contains("Session not found") {
2013 return Err(ErrorKind::Session(SessionErrorKind::NotFound(
2014 session_id.unwrap_or_else(|| "unknown".into()),
2015 ))
2016 .into());
2017 }
2018 return Err(Error::with_message(
2019 ErrorKind::Rpc { code: err.code },
2020 err.message,
2021 ));
2022 }
2023 Ok(response.result.unwrap_or(serde_json::Value::Null))
2024 }
2025
2026 pub(crate) async fn send_response(&self, response: &JsonRpcResponse) -> Result<()> {
2028 self.inner.rpc.write(response).await
2029 }
2030
2031 pub(crate) fn from_inner(inner: Arc<ClientInner>) -> Self {
2033 Self { inner }
2034 }
2035
2036 #[expect(dead_code, reason = "reserved for future pub(crate) use")]
2040 pub(crate) fn take_request_rx(&self) -> Option<mpsc::UnboundedReceiver<JsonRpcRequest>> {
2041 self.inner.request_rx.lock().take()
2042 }
2043
2044 pub(crate) fn register_session(
2052 &self,
2053 session_id: &SessionId,
2054 ) -> crate::router::SessionChannels {
2055 self.inner.router.ensure_started(
2056 &self.inner.notification_tx,
2057 &self.inner.request_rx,
2058 self.inner.llm_inference.get().cloned(),
2059 self.inner.on_github_telemetry.clone(),
2060 self.inner.github_token_registry.clone(),
2061 );
2062 self.inner.router.register(session_id)
2063 }
2064
2065 pub(crate) fn unregister_session(&self, session_id: &SessionId) {
2067 self.inner.router.unregister(session_id);
2068 }
2069
2070 pub(crate) fn register_github_token_provider(
2071 &self,
2072 provider: Arc<dyn GitHubTokenProvider>,
2073 ) -> github_token::GitHubTokenRegistration {
2074 self.inner.router.ensure_started(
2075 &self.inner.notification_tx,
2076 &self.inner.request_rx,
2077 self.inner.llm_inference.get().cloned(),
2078 self.inner.on_github_telemetry.clone(),
2079 self.inner.github_token_registry.clone(),
2080 );
2081 let id = self.inner.github_token_registry.register(provider);
2082 github_token::GitHubTokenRegistration::new(self.inner.github_token_registry.clone(), id)
2083 }
2084
2085 pub(crate) fn retire_github_token_provider(&self, session_id: &SessionId) {
2086 self.inner.github_token_registry.retire_session(session_id);
2087 }
2088
2089 pub fn protocol_version(&self) -> Option<u32> {
2096 self.inner.negotiated_protocol_version.get().copied()
2097 }
2098
2099 pub fn startup_timings(&self) -> Option<StartupTimings> {
2106 self.inner.startup_timings.get().cloned()
2107 }
2108
2109 pub async fn verify_protocol_version(&self) -> Result<()> {
2133 let handshake_start = Instant::now();
2134 let mut used_fallback_ping = false;
2135 let server_version = match self.connect_handshake().await {
2139 Ok(v) => v,
2140 Err(ref e) if e.rpc_code() == Some(error_codes::METHOD_NOT_FOUND) => {
2141 used_fallback_ping = true;
2142 self.ping(None).await?.protocol_version
2143 }
2144 Err(e) => return Err(e),
2145 };
2146
2147 match server_version {
2148 None => {
2149 warn!("CLI server did not report protocolVersion; skipping version check");
2150 }
2151 Some(v) if !(MIN_PROTOCOL_VERSION..=SDK_PROTOCOL_VERSION).contains(&v) => {
2152 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionMismatch {
2153 server: v,
2154 min: MIN_PROTOCOL_VERSION,
2155 max: SDK_PROTOCOL_VERSION,
2156 })
2157 .into());
2158 }
2159 Some(v) => {
2160 if let Some(&existing) = self.inner.negotiated_protocol_version.get() {
2161 if existing != v {
2162 return Err(ErrorKind::Protocol(ProtocolErrorKind::VersionChanged {
2163 previous: existing,
2164 current: v,
2165 })
2166 .into());
2167 }
2168 } else {
2169 let _ = self.inner.negotiated_protocol_version.set(v);
2170 }
2171 }
2172 }
2173
2174 debug!(
2175 elapsed_ms = handshake_start.elapsed().as_millis(),
2176 protocol_version = ?server_version,
2177 used_fallback_ping,
2178 "Client::verify_protocol_version protocol handshake complete"
2179 );
2180 Ok(())
2181 }
2182
2183 async fn connect_handshake(&self) -> Result<Option<u32>> {
2190 let params = crate::generated::api_types::ConnectRequest {
2191 token: self.inner.effective_connection_token.clone(),
2192 enable_git_hub_telemetry_forwarding: self
2193 .inner
2194 .on_github_telemetry
2195 .is_some()
2196 .then_some(true),
2197 ..Default::default()
2198 };
2199 let value = self
2200 .call(
2201 crate::generated::api_types::rpc_methods::CONNECT,
2202 Some(serde_json::to_value(params)?),
2203 )
2204 .await?;
2205 let result: crate::generated::api_types::ConnectResult = serde_json::from_value(value)?;
2206 Ok(Some(u32::try_from(result.protocol_version).map_err(
2207 |_| ProtocolErrorKind::InvalidProtocolVersion {
2208 server: result.protocol_version,
2209 },
2210 )?))
2211 }
2212
2213 pub async fn ping(&self, message: Option<&str>) -> Result<crate::types::PingResponse> {
2221 let params = match message {
2222 Some(m) => serde_json::json!({ "message": m }),
2223 None => serde_json::json!({}),
2224 };
2225 let value = self
2226 .call(generated::api_types::rpc_methods::PING, Some(params))
2227 .await?;
2228 Ok(serde_json::from_value(value)?)
2229 }
2230
2231 pub async fn list_sessions(
2234 &self,
2235 filter: Option<SessionListFilter>,
2236 ) -> Result<Vec<SessionMetadata>> {
2237 let params = match filter {
2238 Some(f) => serde_json::json!({ "filter": f }),
2239 None => serde_json::json!({}),
2240 };
2241 let result = self.call("session.list", Some(params)).await?;
2242 let response: ListSessionsResponse = serde_json::from_value(result)?;
2243 Ok(response.sessions)
2244 }
2245
2246 pub async fn get_session_metadata(
2264 &self,
2265 session_id: &SessionId,
2266 ) -> Result<Option<SessionMetadata>> {
2267 let result = self
2268 .call(
2269 "session.getMetadata",
2270 Some(serde_json::json!({ "sessionId": session_id })),
2271 )
2272 .await?;
2273 let response: GetSessionMetadataResponse = serde_json::from_value(result)?;
2274 Ok(response.session)
2275 }
2276
2277 pub async fn delete_session(&self, session_id: &SessionId) -> Result<()> {
2279 self.call(
2280 "session.delete",
2281 Some(serde_json::json!({ "sessionId": session_id })),
2282 )
2283 .await?;
2284 self.retire_github_token_provider(session_id);
2285 Ok(())
2286 }
2287
2288 #[cfg(feature = "test-support")]
2291 #[doc(hidden)]
2292 pub fn start_router_for_test(&self) {
2293 self.inner.router.ensure_started(
2294 &self.inner.notification_tx,
2295 &self.inner.request_rx,
2296 self.inner.llm_inference.get().cloned(),
2297 self.inner.on_github_telemetry.clone(),
2298 self.inner.github_token_registry.clone(),
2299 );
2300 }
2301
2302 #[cfg(feature = "test-support")]
2303 #[doc(hidden)]
2304 pub async fn cleanup_sessions_for_test(&self) -> Result<()> {
2307 let mut first_error = None;
2308
2309 for session_id in self.inner.router.session_ids() {
2310 if let Err(error) = self
2311 .call(
2312 "session.destroy",
2313 Some(serde_json::json!({ "sessionId": session_id })),
2314 )
2315 .await
2316 && first_error.is_none()
2317 {
2318 first_error = Some(error);
2319 }
2320 self.inner.router.unregister(&session_id);
2321 }
2322 self.inner.github_token_registry.clear();
2323
2324 match self.list_sessions(None).await {
2325 Ok(sessions) => {
2326 for session in sessions {
2327 if let Err(error) = self.delete_session(&session.session_id).await
2328 && first_error.is_none()
2329 {
2330 first_error = Some(error);
2331 }
2332 }
2333 }
2334 Err(error) if first_error.is_none() => first_error = Some(error),
2335 Err(_) => {}
2336 }
2337
2338 match first_error {
2339 Some(error) => Err(error),
2340 None => Ok(()),
2341 }
2342 }
2343
2344 pub async fn get_last_session_id(&self) -> Result<Option<SessionId>> {
2360 let result = self
2361 .call("session.getLastId", Some(serde_json::json!({})))
2362 .await?;
2363 let response: GetLastSessionIdResponse = serde_json::from_value(result)?;
2364 Ok(response.session_id)
2365 }
2366
2367 pub async fn get_foreground_session_id(&self) -> Result<Option<SessionId>> {
2372 let result = self
2373 .call("session.getForeground", Some(serde_json::json!({})))
2374 .await?;
2375 let response: GetForegroundSessionResponse = serde_json::from_value(result)?;
2376 Ok(response.session_id)
2377 }
2378
2379 pub async fn set_foreground_session_id(&self, session_id: &SessionId) -> Result<()> {
2384 self.call(
2385 "session.setForeground",
2386 Some(serde_json::json!({ "sessionId": session_id })),
2387 )
2388 .await?;
2389 Ok(())
2390 }
2391
2392 pub async fn get_status(&self) -> Result<GetStatusResponse> {
2394 let result = self.call("status.get", Some(serde_json::json!({}))).await?;
2395 Ok(serde_json::from_value(result)?)
2396 }
2397
2398 pub async fn get_auth_status(&self) -> Result<GetAuthStatusResponse> {
2400 let result = self
2401 .call("auth.getStatus", Some(serde_json::json!({})))
2402 .await?;
2403 Ok(serde_json::from_value(result)?)
2404 }
2405
2406 pub async fn list_models(&self) -> Result<Vec<Model>> {
2411 let cache = self.inner.models_cache.lock().clone();
2412 let models = cache
2413 .get_or_try_init(|| async {
2414 if let Some(handler) = &self.inner.on_list_models {
2415 handler.list_models().await
2416 } else {
2417 Ok(self.rpc().models().list().await?.models)
2418 }
2419 })
2420 .await?;
2421 Ok(models.clone())
2422 }
2423
2424 pub(crate) async fn resolve_trace_context(&self) -> TraceContext {
2427 if let Some(provider) = &self.inner.on_get_trace_context {
2428 provider.get_trace_context().await
2429 } else {
2430 TraceContext::default()
2431 }
2432 }
2433
2434 pub fn pid(&self) -> Option<u32> {
2436 self.inner.child.lock().as_ref().and_then(|c| c.id())
2437 }
2438
2439 pub async fn stop(&self) -> std::result::Result<(), StopErrors> {
2466 let pid = self.pid();
2467 info!(pid = ?pid, "stopping CLI process");
2468 let mut errors: Vec<Error> = Vec::new();
2469
2470 for session_id in self.inner.router.session_ids() {
2473 match self
2474 .call(
2475 "session.destroy",
2476 Some(serde_json::json!({ "sessionId": session_id })),
2477 )
2478 .await
2479 {
2480 Ok(_) => {}
2481 Err(e) => {
2482 warn!(
2483 session_id = %session_id,
2484 error = %e,
2485 "session.destroy failed during Client::stop",
2486 );
2487 errors.push(e);
2488 }
2489 }
2490 self.inner.router.unregister(&session_id);
2491 }
2492 self.inner.github_token_registry.clear();
2493
2494 let should_shutdown_runtime = self.inner.child.lock().is_some();
2495 #[cfg(feature = "bundled-in-process")]
2496 let should_shutdown_runtime =
2497 should_shutdown_runtime || self.inner.ffi_host.lock().is_some();
2498 if should_shutdown_runtime {
2499 let runtime_shutdown_start = Instant::now();
2500 match tokio::time::timeout(RUNTIME_SHUTDOWN_TIMEOUT, self.rpc().runtime().shutdown())
2501 .await
2502 {
2503 Ok(Ok(())) => {
2504 debug!(
2505 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2506 "Client::stop runtime shutdown complete"
2507 );
2508 }
2509 Ok(Err(e)) => {
2510 warn!(
2511 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2512 error = %e,
2513 "runtime.shutdown failed during Client::stop",
2514 );
2515 errors.push(e);
2516 }
2517 Err(_) => {
2518 let e = std::io::Error::new(
2519 std::io::ErrorKind::TimedOut,
2520 "runtime.shutdown timed out during Client::stop",
2521 );
2522 warn!(
2523 elapsed_ms = runtime_shutdown_start.elapsed().as_millis(),
2524 timeout = ?RUNTIME_SHUTDOWN_TIMEOUT,
2525 error = %e,
2526 "runtime.shutdown timed out during Client::stop",
2527 );
2528 errors.push(e.into());
2529 }
2530 }
2531 }
2532
2533 let child = self.inner.child.lock().take();
2534 *self.inner.state.lock() = ConnectionState::Disconnected;
2535 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2536 if let Some(mut child) = child {
2537 match child.try_wait() {
2538 Ok(Some(_status)) => {}
2539 Ok(None) => {
2540 if let Err(e) = child.kill().await {
2547 errors.push(e.into());
2548 }
2549 }
2550 Err(e) => errors.push(e.into()),
2551 }
2552 }
2553
2554 #[cfg(feature = "bundled-in-process")]
2557 {
2558 if let Some(host) = self.inner.ffi_host.lock().take() {
2559 self.inner.rpc.force_close();
2560 host.close();
2561 }
2562 }
2563
2564 info!(pid = ?pid, errors = errors.len(), "CLI process stopped");
2565 if errors.is_empty() {
2566 Ok(())
2567 } else {
2568 Err(StopErrors(errors))
2569 }
2570 }
2571
2572 pub fn force_stop(&self) {
2602 let pid = self.pid();
2603 info!(pid = ?pid, "force-stopping CLI process");
2604 if let Some(mut child) = self.inner.child.lock().take()
2605 && let Err(e) = child.start_kill()
2606 {
2607 error!(pid = ?pid, error = %e, "failed to send kill signal");
2608 }
2609 self.inner.rpc.force_close();
2610 #[cfg(feature = "bundled-in-process")]
2611 {
2612 if let Some(host) = self.inner.ffi_host.lock().take() {
2613 host.close();
2614 }
2615 }
2616 self.inner.router.clear();
2619 self.inner.github_token_registry.clear();
2620 *self.inner.state.lock() = ConnectionState::Disconnected;
2621 *self.inner.models_cache.lock() = Arc::new(tokio::sync::OnceCell::new());
2622 }
2623
2624 pub fn subscribe_lifecycle(&self) -> LifecycleSubscription {
2659 LifecycleSubscription::new(self.inner.lifecycle_tx.subscribe())
2660 }
2661}
2662
2663impl Drop for ClientInner {
2664 fn drop(&mut self) {
2665 if let Some(ref mut child) = *self.child.lock() {
2666 let pid = child.id();
2667 if let Err(e) = child.start_kill() {
2668 error!(pid = ?pid, error = %e, "failed to kill CLI process on drop");
2669 } else {
2670 info!(pid = ?pid, "kill signal sent for CLI process on drop");
2671 }
2672 }
2673 #[cfg(feature = "bundled-in-process")]
2674 {
2675 if let Some(host) = self.ffi_host.lock().take() {
2676 self.rpc.force_close();
2677 host.close();
2678 }
2679 }
2680 }
2681}
2682
2683#[cfg(test)]
2684mod tests {
2685 use super::*;
2686
2687 #[test]
2688 fn is_transport_failure_matches_request_cancelled() {
2689 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::RequestCancelled));
2690 assert!(err.is_transport_failure());
2691 }
2692
2693 #[test]
2694 fn is_transport_failure_matches_io_error() {
2695 let err = Error::from(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "gone"));
2696 assert!(err.is_transport_failure());
2697 }
2698
2699 #[test]
2700 fn is_transport_failure_rejects_rpc_error() {
2701 let err = Error::with_message(ErrorKind::Rpc { code: -1 }, "bad");
2702 assert!(!err.is_transport_failure());
2703 }
2704
2705 #[test]
2706 fn is_transport_failure_rejects_session_error() {
2707 let err = Error::from(ErrorKind::Session(SessionErrorKind::NotFound("s1".into())));
2708 assert!(!err.is_transport_failure());
2709 }
2710
2711 #[test]
2712 fn client_options_builder_composes() {
2713 let opts = ClientOptions::new()
2714 .with_program(CliProgram::Path(PathBuf::from("/usr/local/bin/copilot")))
2715 .with_prefix_args(["node"])
2716 .with_cwd(PathBuf::from("/tmp"))
2717 .with_env([("KEY", "value")])
2718 .with_env_remove(["UNWANTED"])
2719 .with_extra_args(["--quiet"])
2720 .with_github_token("ghp_test")
2721 .with_use_logged_in_user(false)
2722 .with_log_level(LogLevel::Debug)
2723 .with_session_idle_timeout_seconds(120)
2724 .with_enable_remote_sessions(true);
2725 assert!(matches!(opts.program, CliProgram::Path(_)));
2726 assert_eq!(opts.prefix_args, vec![std::ffi::OsString::from("node")]);
2727 assert_eq!(opts.working_directory, PathBuf::from("/tmp"));
2728 assert_eq!(
2729 opts.env,
2730 vec![(
2731 std::ffi::OsString::from("KEY"),
2732 std::ffi::OsString::from("value")
2733 )]
2734 );
2735 assert_eq!(opts.env_remove, vec![std::ffi::OsString::from("UNWANTED")]);
2736 assert_eq!(opts.extra_args, vec!["--quiet".to_string()]);
2737 assert_eq!(opts.github_token.as_deref(), Some("ghp_test"));
2738 assert_eq!(opts.use_logged_in_user, Some(false));
2739 assert!(matches!(opts.log_level, Some(LogLevel::Debug)));
2740 assert_eq!(opts.session_idle_timeout_seconds, Some(120));
2741 assert!(opts.enable_remote_sessions);
2742 }
2743
2744 #[test]
2745 fn default_transport_values_resolve_without_process_state() {
2746 assert!(matches!(
2747 resolve_default_transport_value(None).unwrap(),
2748 Transport::Stdio
2749 ));
2750 assert!(matches!(
2751 resolve_default_transport_value(Some("stdio")).unwrap(),
2752 Transport::Stdio
2753 ));
2754 assert!(matches!(
2755 resolve_default_transport_value(Some("INPROCESS")).unwrap(),
2756 Transport::InProcess
2757 ));
2758 assert!(resolve_default_transport_value(Some("tcp")).is_err());
2759 }
2760
2761 #[test]
2762 fn inprocess_rejects_process_scoped_options() {
2763 let invalid = [
2764 ClientOptions::new().with_cwd("."),
2765 ClientOptions::new().with_env([("KEY", "value")]),
2766 ClientOptions::new().with_env_remove(["KEY"]),
2767 ClientOptions::new().with_telemetry(TelemetryConfig::default()),
2768 ClientOptions::new().with_prefix_args(["index.js"]),
2769 ClientOptions::new().with_program(CliProgram::Path("copilot".into())),
2770 ClientOptions::new().with_extra_args(["--verbose"]),
2771 ];
2772
2773 for options in invalid {
2774 assert!(validate_inprocess_options(&options).is_err());
2775 }
2776 }
2777
2778 #[test]
2779 fn inprocess_allows_typed_runtime_options() {
2780 let options = ClientOptions::new()
2781 .with_base_directory("state")
2782 .with_log_level(LogLevel::Debug)
2783 .with_session_idle_timeout_seconds(10)
2784 .with_github_token("token")
2785 .with_use_logged_in_user(false)
2786 .with_enable_remote_sessions(true);
2787
2788 assert!(validate_inprocess_options(&options).is_ok());
2789 }
2790
2791 #[cfg(not(feature = "bundled-in-process"))]
2792 #[tokio::test]
2793 async fn inprocess_requires_cargo_feature() {
2794 let error = Client::start(ClientOptions::new().with_transport(Transport::InProcess))
2795 .await
2796 .unwrap_err();
2797
2798 assert!(error.to_string().contains("bundled-in-process"));
2799 }
2800
2801 #[test]
2802 fn is_transport_failure_rejects_other_protocol_errors() {
2803 let err = Error::from(ErrorKind::Protocol(ProtocolErrorKind::CliStartupTimeout));
2804 assert!(!err.is_transport_failure());
2805 }
2806
2807 #[test]
2808 fn build_command_lets_env_remove_strip_injected_token() {
2809 let opts = ClientOptions {
2810 github_token: Some("secret".to_string()),
2811 env_remove: vec![std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN")],
2812 ..Default::default()
2813 };
2814 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2815 let action = cmd
2817 .as_std()
2818 .get_envs()
2819 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2820 .map(|(_, v)| v);
2821 assert_eq!(
2822 action,
2823 Some(None),
2824 "env_remove should win over github_token"
2825 );
2826 }
2827
2828 #[test]
2829 fn build_command_lets_env_override_injected_token() {
2830 let opts = ClientOptions {
2831 github_token: Some("from-options".to_string()),
2832 env: vec![(
2833 std::ffi::OsString::from("COPILOT_SDK_AUTH_TOKEN"),
2834 std::ffi::OsString::from("from-env"),
2835 )],
2836 ..Default::default()
2837 };
2838 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2839 let value = cmd
2840 .as_std()
2841 .get_envs()
2842 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2843 .and_then(|(_, v)| v);
2844 assert_eq!(value, Some(std::ffi::OsStr::new("from-env")));
2845 }
2846
2847 #[test]
2848 fn build_command_injects_github_token_by_default() {
2849 let opts = ClientOptions {
2850 github_token: Some("just-the-token".to_string()),
2851 ..Default::default()
2852 };
2853 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2854 let value = cmd
2855 .as_std()
2856 .get_envs()
2857 .find(|(k, _)| *k == std::ffi::OsStr::new("COPILOT_SDK_AUTH_TOKEN"))
2858 .and_then(|(_, v)| v);
2859 assert_eq!(value, Some(std::ffi::OsStr::new("just-the-token")));
2860 }
2861
2862 fn env_value<'a>(cmd: &'a tokio::process::Command, key: &str) -> Option<&'a std::ffi::OsStr> {
2863 cmd.as_std()
2864 .get_envs()
2865 .find(|(k, _)| *k == std::ffi::OsStr::new(key))
2866 .and_then(|(_, v)| v)
2867 }
2868
2869 #[test]
2870 fn telemetry_config_builder_composes() {
2871 let cfg = TelemetryConfig::new()
2872 .with_otlp_endpoint("http://collector:4318")
2873 .with_otlp_protocol(OtlpHttpProtocol::HttpProtobuf)
2874 .with_file_path(PathBuf::from("/var/log/copilot.jsonl"))
2875 .with_exporter_type(OtelExporterType::OtlpHttp)
2876 .with_source_name("my-app")
2877 .with_capture_content(true);
2878
2879 assert_eq!(cfg.otlp_endpoint.as_deref(), Some("http://collector:4318"));
2880 assert_eq!(cfg.otlp_protocol, Some(OtlpHttpProtocol::HttpProtobuf));
2881 assert_eq!(
2882 cfg.file_path.as_deref(),
2883 Some(Path::new("/var/log/copilot.jsonl")),
2884 );
2885 assert_eq!(cfg.exporter_type, Some(OtelExporterType::OtlpHttp));
2886 assert_eq!(cfg.source_name.as_deref(), Some("my-app"));
2887 assert_eq!(cfg.capture_content, Some(true));
2888 assert!(!cfg.is_empty());
2889 assert!(TelemetryConfig::new().is_empty());
2890 }
2891
2892 #[test]
2893 fn otlp_http_protocol_serde_matches_env_value() {
2894 for (protocol, wire) in [
2895 (OtlpHttpProtocol::HttpJson, "http/json"),
2896 (OtlpHttpProtocol::HttpProtobuf, "http/protobuf"),
2897 ] {
2898 assert_eq!(protocol.as_str(), wire);
2899
2900 let serialized = serde_json::to_string(&protocol).unwrap();
2901 assert_eq!(serialized, format!("\"{wire}\""));
2902
2903 let deserialized: OtlpHttpProtocol = serde_json::from_str(&serialized).unwrap();
2904 assert_eq!(deserialized, protocol);
2905 }
2906 }
2907
2908 #[test]
2909 fn build_command_sets_otel_env_when_telemetry_enabled() {
2910 let opts = ClientOptions {
2911 telemetry: Some(TelemetryConfig {
2912 otlp_endpoint: Some("http://collector:4318".to_string()),
2913 otlp_protocol: Some(OtlpHttpProtocol::HttpProtobuf),
2914 file_path: Some(PathBuf::from("/var/log/copilot.jsonl")),
2915 exporter_type: Some(OtelExporterType::OtlpHttp),
2916 source_name: Some("my-app".to_string()),
2917 capture_content: Some(true),
2918 }),
2919 ..Default::default()
2920 };
2921 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2922 assert_eq!(
2923 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2924 Some(std::ffi::OsStr::new("true")),
2925 );
2926 assert_eq!(
2927 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2928 Some(std::ffi::OsStr::new("http://collector:4318")),
2929 );
2930 assert_eq!(
2931 env_value(&cmd, "OTEL_EXPORTER_OTLP_PROTOCOL"),
2932 Some(std::ffi::OsStr::new("http/protobuf")),
2933 );
2934 assert_eq!(
2935 env_value(&cmd, "COPILOT_OTEL_FILE_EXPORTER_PATH"),
2936 Some(std::ffi::OsStr::new("/var/log/copilot.jsonl")),
2937 );
2938 assert_eq!(
2939 env_value(&cmd, "COPILOT_OTEL_EXPORTER_TYPE"),
2940 Some(std::ffi::OsStr::new("otlp-http")),
2941 );
2942 assert_eq!(
2943 env_value(&cmd, "COPILOT_OTEL_SOURCE_NAME"),
2944 Some(std::ffi::OsStr::new("my-app")),
2945 );
2946 assert_eq!(
2947 env_value(&cmd, "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
2948 Some(std::ffi::OsStr::new("true")),
2949 );
2950 }
2951
2952 #[test]
2953 fn build_command_omits_otel_env_when_telemetry_none() {
2954 let opts = ClientOptions::default();
2955 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2956 for key in [
2957 "COPILOT_OTEL_ENABLED",
2958 "OTEL_EXPORTER_OTLP_ENDPOINT",
2959 "OTEL_EXPORTER_OTLP_PROTOCOL",
2960 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2961 "COPILOT_OTEL_EXPORTER_TYPE",
2962 "COPILOT_OTEL_SOURCE_NAME",
2963 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2964 ] {
2965 assert!(
2966 env_value(&cmd, key).is_none(),
2967 "expected {key} to be unset when telemetry is None",
2968 );
2969 }
2970 }
2971
2972 #[test]
2973 fn build_command_omits_unset_telemetry_fields() {
2974 let opts = ClientOptions {
2975 telemetry: Some(TelemetryConfig {
2976 otlp_endpoint: Some("http://collector:4318".to_string()),
2977 ..Default::default()
2978 }),
2979 ..Default::default()
2980 };
2981 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
2982 assert_eq!(
2984 env_value(&cmd, "COPILOT_OTEL_ENABLED"),
2985 Some(std::ffi::OsStr::new("true")),
2986 );
2987 assert_eq!(
2988 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
2989 Some(std::ffi::OsStr::new("http://collector:4318")),
2990 );
2991 for key in [
2993 "OTEL_EXPORTER_OTLP_PROTOCOL",
2994 "COPILOT_OTEL_FILE_EXPORTER_PATH",
2995 "COPILOT_OTEL_EXPORTER_TYPE",
2996 "COPILOT_OTEL_SOURCE_NAME",
2997 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
2998 ] {
2999 assert!(env_value(&cmd, key).is_none(), "{key} should be unset");
3000 }
3001 }
3002
3003 #[test]
3004 fn build_command_lets_user_env_override_telemetry() {
3005 let opts = ClientOptions {
3006 telemetry: Some(TelemetryConfig {
3007 otlp_endpoint: Some("http://from-config:4318".to_string()),
3008 ..Default::default()
3009 }),
3010 env: vec![(
3011 std::ffi::OsString::from("OTEL_EXPORTER_OTLP_ENDPOINT"),
3012 std::ffi::OsString::from("http://from-user-env:4318"),
3013 )],
3014 ..Default::default()
3015 };
3016 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3017 assert_eq!(
3018 env_value(&cmd, "OTEL_EXPORTER_OTLP_ENDPOINT"),
3019 Some(std::ffi::OsStr::new("http://from-user-env:4318")),
3020 "user-supplied options.env should override telemetry config",
3021 );
3022 }
3023
3024 #[test]
3025 fn build_command_sets_copilot_home_env_when_configured() {
3026 let opts = ClientOptions::new().with_base_directory(PathBuf::from("/custom/copilot"));
3027 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3028 assert_eq!(
3029 env_value(&cmd, "COPILOT_HOME"),
3030 Some(std::ffi::OsStr::new("/custom/copilot")),
3031 );
3032
3033 let opts = ClientOptions::default();
3034 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3035 assert!(env_value(&cmd, "COPILOT_HOME").is_none());
3036 }
3037
3038 #[test]
3039 fn build_command_sets_connection_token_env_when_configured() {
3040 let opts = ClientOptions::new().with_transport(Transport::Tcp {
3041 port: 0,
3042 connection_token: Some("secret-token".to_string()),
3043 });
3044 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3045 assert_eq!(
3046 env_value(&cmd, "COPILOT_CONNECTION_TOKEN"),
3047 Some(std::ffi::OsStr::new("secret-token")),
3048 );
3049
3050 let opts = ClientOptions::default();
3051 let cmd = Client::build_command(Path::new("/bin/echo"), &opts, Path::new("/tmp"));
3052 assert!(env_value(&cmd, "COPILOT_CONNECTION_TOKEN").is_none());
3053 }
3054
3055 #[tokio::test]
3056 async fn start_rejects_empty_connection_token() {
3057 let opts = ClientOptions::new()
3058 .with_transport(Transport::Tcp {
3059 port: 0,
3060 connection_token: Some(String::new()),
3061 })
3062 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3063 let err = Client::start(opts).await.unwrap_err();
3064 assert!(
3065 matches!(err.kind(), ErrorKind::InvalidConfig),
3066 "got {err:?}"
3067 );
3068 }
3069
3070 #[tokio::test]
3071 async fn start_rejects_empty_external_connection_token() {
3072 let opts = ClientOptions::new()
3073 .with_transport(Transport::External {
3074 host: "127.0.0.1".to_string(),
3075 port: 1,
3076 connection_token: Some(String::new()),
3077 })
3078 .with_program(CliProgram::Path(PathBuf::from("/bin/echo")));
3079 let err = Client::start(opts).await.unwrap_err();
3080 assert!(
3081 matches!(err.kind(), ErrorKind::InvalidConfig),
3082 "got {err:?}"
3083 );
3084 }
3085
3086 #[test]
3087 fn telemetry_config_capture_content_serializes_as_lowercase_bool() {
3088 let opts_true = ClientOptions {
3089 telemetry: Some(TelemetryConfig {
3090 capture_content: Some(true),
3091 ..Default::default()
3092 }),
3093 ..Default::default()
3094 };
3095 let opts_false = ClientOptions {
3096 telemetry: Some(TelemetryConfig {
3097 capture_content: Some(false),
3098 ..Default::default()
3099 }),
3100 ..Default::default()
3101 };
3102 let cmd_true = Client::build_command(Path::new("/bin/echo"), &opts_true, Path::new("/tmp"));
3103 let cmd_false =
3104 Client::build_command(Path::new("/bin/echo"), &opts_false, Path::new("/tmp"));
3105 assert_eq!(
3106 env_value(
3107 &cmd_true,
3108 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3109 ),
3110 Some(std::ffi::OsStr::new("true")),
3111 );
3112 assert_eq!(
3113 env_value(
3114 &cmd_false,
3115 "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
3116 ),
3117 Some(std::ffi::OsStr::new("false")),
3118 );
3119 }
3120
3121 #[test]
3122 fn session_idle_timeout_args_are_omitted_by_default() {
3123 let opts = ClientOptions::default();
3124 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3125 }
3126
3127 #[test]
3128 fn session_idle_timeout_args_omitted_for_zero() {
3129 let opts = ClientOptions {
3130 session_idle_timeout_seconds: Some(0),
3131 ..Default::default()
3132 };
3133 assert!(Client::session_idle_timeout_args(&opts).is_empty());
3134 }
3135
3136 #[test]
3137 fn session_idle_timeout_args_emit_flag_for_positive_value() {
3138 let opts = ClientOptions {
3139 session_idle_timeout_seconds: Some(300),
3140 ..Default::default()
3141 };
3142 assert_eq!(
3143 Client::session_idle_timeout_args(&opts),
3144 vec!["--session-idle-timeout".to_string(), "300".to_string()]
3145 );
3146 }
3147
3148 #[test]
3149 fn remote_args_omitted_by_default() {
3150 let opts = ClientOptions::default();
3151 assert!(Client::remote_args(&opts).is_empty());
3152 }
3153
3154 #[test]
3155 fn remote_args_emit_flag_when_enabled() {
3156 let opts = ClientOptions {
3157 enable_remote_sessions: true,
3158 ..Default::default()
3159 };
3160 assert_eq!(Client::remote_args(&opts), vec!["--remote".to_string()]);
3161 }
3162
3163 #[test]
3164 fn log_level_args_omitted_when_unset() {
3165 let opts = ClientOptions::default();
3166 assert!(opts.log_level.is_none());
3167 assert!(
3168 Client::log_level_args(&opts).is_empty(),
3169 "with no caller-supplied log_level the SDK must not pass --log-level"
3170 );
3171 }
3172
3173 #[test]
3174 fn log_level_args_emit_flag_when_set() {
3175 let opts = ClientOptions::default().with_log_level(LogLevel::Debug);
3176 assert_eq!(Client::log_level_args(&opts), vec!["--log-level", "debug"]);
3177 }
3178
3179 #[test]
3180 fn log_level_str_round_trips() {
3181 for level in [
3182 LogLevel::None,
3183 LogLevel::Error,
3184 LogLevel::Warning,
3185 LogLevel::Info,
3186 LogLevel::Debug,
3187 LogLevel::All,
3188 ] {
3189 let s = level.as_str();
3190 let json = serde_json::to_string(&level).unwrap();
3191 assert_eq!(json, format!("\"{s}\""));
3192 let parsed: LogLevel = serde_json::from_str(&json).unwrap();
3193 assert_eq!(parsed, level);
3194 }
3195 }
3196
3197 #[test]
3198 fn client_options_debug_redacts_handler() {
3199 struct StubHandler;
3200 #[async_trait]
3201 impl ListModelsHandler for StubHandler {
3202 async fn list_models(&self) -> Result<Vec<Model>> {
3203 Ok(vec![])
3204 }
3205 }
3206 let opts = ClientOptions {
3207 on_list_models: Some(Arc::new(StubHandler)),
3208 github_token: Some("secret-token".into()),
3209 ..Default::default()
3210 };
3211 let debug = format!("{opts:?}");
3212 assert!(debug.contains("on_list_models: Some(\"<set>\")"));
3213 assert!(debug.contains("github_token: Some(\"<redacted>\")"));
3214 assert!(!debug.contains("secret-token"));
3215 }
3216
3217 #[tokio::test]
3218 async fn list_models_uses_on_list_models_handler_when_set() {
3219 use std::sync::atomic::{AtomicUsize, Ordering};
3220
3221 struct CountingHandler {
3222 calls: Arc<AtomicUsize>,
3223 models: Vec<Model>,
3224 }
3225 #[async_trait]
3226 impl ListModelsHandler for CountingHandler {
3227 async fn list_models(&self) -> Result<Vec<Model>> {
3228 self.calls.fetch_add(1, Ordering::SeqCst);
3229 Ok(self.models.clone())
3230 }
3231 }
3232
3233 let calls = Arc::new(AtomicUsize::new(0));
3234 let model = Model {
3235 id: "byok-gpt-4".into(),
3236 name: "BYOK GPT-4".into(),
3237 ..Default::default()
3238 };
3239 let handler: Arc<dyn ListModelsHandler> = Arc::new(CountingHandler {
3240 calls: Arc::clone(&calls),
3241 models: vec![model.clone()],
3242 });
3243
3244 let client = client_with_list_models_handler(handler);
3245
3246 let result = client.list_models().await.unwrap();
3247 assert_eq!(result.len(), 1);
3248 assert_eq!(result[0].id, "byok-gpt-4");
3249 assert_eq!(calls.load(Ordering::SeqCst), 1);
3250 }
3251
3252 #[tokio::test]
3253 async fn list_models_serializes_concurrent_cache_misses() {
3254 use std::sync::atomic::{AtomicUsize, Ordering};
3255
3256 struct SlowCountingHandler {
3257 calls: Arc<AtomicUsize>,
3258 models: Vec<Model>,
3259 }
3260 #[async_trait]
3261 impl ListModelsHandler for SlowCountingHandler {
3262 async fn list_models(&self) -> Result<Vec<Model>> {
3263 self.calls.fetch_add(1, Ordering::SeqCst);
3264 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
3265 Ok(self.models.clone())
3266 }
3267 }
3268
3269 let calls = Arc::new(AtomicUsize::new(0));
3270 let model = Model {
3271 id: "single-flight-model".into(),
3272 name: "Single Flight Model".into(),
3273 ..Default::default()
3274 };
3275 let handler: Arc<dyn ListModelsHandler> = Arc::new(SlowCountingHandler {
3276 calls: Arc::clone(&calls),
3277 models: vec![model],
3278 });
3279 let client = client_with_list_models_handler(handler);
3280
3281 let (first, second) = tokio::join!(client.list_models(), client.list_models());
3282 assert_eq!(first.unwrap()[0].id, "single-flight-model");
3283 assert_eq!(second.unwrap()[0].id, "single-flight-model");
3284 assert_eq!(calls.load(Ordering::SeqCst), 1);
3285 }
3286
3287 #[tokio::test]
3288 async fn cancelled_resume_session_unregisters_pending_session() {
3289 let (client_write, _server_read) = tokio::io::duplex(8192);
3290 let (_server_write, client_read) = tokio::io::duplex(8192);
3291 let client = Client::from_streams(client_read, client_write, std::env::temp_dir()).unwrap();
3292 assert!(client.startup_timings().is_none());
3293 let session_id = SessionId::new("resume-cancel-test");
3294 let handle = tokio::spawn({
3295 let client = client.clone();
3296 async move {
3297 client
3298 .resume_session(ResumeSessionConfig::new(session_id))
3299 .await
3300 }
3301 });
3302
3303 wait_for_pending_session_registration(&client).await;
3304 handle.abort();
3305 let _ = handle.await;
3306
3307 assert!(client.inner.router.session_ids().is_empty());
3308 client.force_stop();
3309 }
3310
3311 #[cfg(any(unix, windows))]
3312 #[tokio::test]
3313 async fn dropping_last_client_kills_spawned_cli() {
3314 let temp = tempfile::tempdir().unwrap();
3315 let ready = temp.path().join("ready");
3316 let survived = temp.path().join("survived");
3317 let child = test_child_command(temp.path(), &ready, &survived)
3318 .spawn()
3319 .unwrap();
3320 let (client_write, _server_read) = tokio::io::duplex(64);
3321 let (_server_write, client_read) = tokio::io::duplex(64);
3322 let client = Client::from_transport(
3323 client_read,
3324 client_write,
3325 Some(child),
3326 temp.path().to_path_buf(),
3327 None,
3328 false,
3329 false,
3330 None,
3331 None,
3332 None,
3333 ClientMode::default(),
3334 )
3335 .unwrap();
3336
3337 wait_for_test_child(&ready).await;
3338 drop(client);
3339
3340 assert_test_child_killed(&survived).await;
3341 }
3342
3343 #[cfg(any(unix, windows))]
3344 #[tokio::test]
3345 async fn spawned_child_is_killed_when_dropped() {
3346 let temp = tempfile::tempdir().unwrap();
3347 let ready = temp.path().join("ready");
3348 let survived = temp.path().join("survived");
3349 let child = test_child_command(temp.path(), &ready, &survived)
3350 .spawn()
3351 .unwrap();
3352
3353 wait_for_test_child(&ready).await;
3354 drop(child);
3355
3356 assert_test_child_killed(&survived).await;
3357 }
3358
3359 #[cfg(any(unix, windows))]
3360 fn test_child_command(temp: &Path, ready: &Path, survived: &Path) -> Command {
3361 #[cfg(unix)]
3362 let mut command = {
3363 let mut command =
3364 Client::build_command(Path::new("sh"), &ClientOptions::default(), temp);
3365 command.args([
3366 "-c",
3367 "printf ready > \"$READY\"; sleep 1; printf survived > \"$SURVIVED\"",
3368 ]);
3369 command
3370 };
3371 #[cfg(windows)]
3372 let mut command = {
3373 let mut command =
3374 Client::build_command(Path::new("powershell.exe"), &ClientOptions::default(), temp);
3375 command.args([
3376 "-NoLogo",
3377 "-NoProfile",
3378 "-NonInteractive",
3379 "-Command",
3380 "Set-Content -LiteralPath $env:READY ready; Start-Sleep -Seconds 1; Set-Content -LiteralPath $env:SURVIVED survived",
3381 ]);
3382 command
3383 };
3384 command.env("READY", ready).env("SURVIVED", survived);
3385 command
3386 }
3387
3388 #[cfg(any(unix, windows))]
3389 async fn wait_for_test_child(ready: &Path) {
3390 let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
3391 while !ready.exists() {
3392 assert!(
3393 tokio::time::Instant::now() < deadline,
3394 "child did not report readiness"
3395 );
3396 tokio::time::sleep(Duration::from_millis(10)).await;
3397 }
3398 }
3399
3400 #[cfg(any(unix, windows))]
3401 async fn assert_test_child_killed(survived: &Path) {
3402 tokio::time::sleep(Duration::from_millis(1500)).await;
3403
3404 assert!(
3405 !survived.exists(),
3406 "child survived after its owner was dropped"
3407 );
3408 }
3409
3410 fn client_with_list_models_handler(handler: Arc<dyn ListModelsHandler>) -> Client {
3411 Client {
3412 inner: Arc::new(ClientInner {
3413 child: parking_lot::Mutex::new(None),
3414 #[cfg(feature = "bundled-in-process")]
3415 ffi_host: parking_lot::Mutex::new(None),
3416 rpc: {
3417 let (req_tx, _req_rx) = mpsc::unbounded_channel();
3418 let (notif_tx, _notif_rx) = broadcast::channel(16);
3419 let (read_pipe, _write_pipe) = tokio::io::duplex(64);
3420 let (_unused_read, write_pipe) = tokio::io::duplex(64);
3421 JsonRpcClient::new(write_pipe, read_pipe, notif_tx, req_tx)
3422 },
3423 cwd: PathBuf::from("."),
3424 request_rx: parking_lot::Mutex::new(None),
3425 notification_tx: broadcast::channel(16).0,
3426 router: router::SessionRouter::new(),
3427 github_token_registry: Arc::new(github_token::GitHubTokenRegistry::new()),
3428 negotiated_protocol_version: OnceLock::new(),
3429 state: parking_lot::Mutex::new(ConnectionState::Connected),
3430 lifecycle_tx: broadcast::channel(16).0,
3431 on_list_models: Some(handler),
3432 models_cache: parking_lot::Mutex::new(Arc::new(tokio::sync::OnceCell::new())),
3433 session_fs_configured: false,
3434 session_fs_sqlite_declared: false,
3435 llm_inference: OnceLock::new(),
3436 on_github_telemetry: None,
3437 on_get_trace_context: None,
3438 effective_connection_token: None,
3439 mode: ClientMode::default(),
3440 startup_timings: OnceLock::new(),
3441 }),
3442 }
3443 }
3444
3445 async fn wait_for_pending_session_registration(client: &Client) {
3446 let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
3447 while client.inner.router.session_ids().is_empty() {
3448 assert!(
3449 tokio::time::Instant::now() < deadline,
3450 "session was not registered"
3451 );
3452 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
3453 }
3454 }
3455}