1use crate::cmd::CommandErrorContext;
10use anyhow::Context;
11use std::collections::HashMap;
12use tokio::process::Command;
13
14#[derive(Debug, Clone)]
16pub struct Session {
17 pub name: String,
18 pub task_id: String,
20 pub project: String,
22 pub created_at: Option<u64>, }
24
25#[derive(Clone)]
27pub struct TmuxManager {
28 prefix: String,
30}
31
32impl TmuxManager {
33 pub fn new() -> Self {
34 Self {
35 prefix: "orch-".to_string(),
36 }
37 }
38
39 pub fn session_name(&self, project: &str, task_id: &str) -> String {
44 let project_name = project
45 .rsplit('/')
46 .next()
47 .unwrap_or(project)
48 .trim_end_matches(".git");
49 let safe_task_id = task_id.replace(':', "-");
50 format!("{}{project_name}-{safe_task_id}", self.prefix)
51 }
52
53 pub async fn create_session(
68 &self,
69 repo: &str,
70 task_id: &str,
71 working_dir: &str,
72 command: &str,
73 env_vars: &[(&str, &str)],
74 ) -> anyhow::Result<String> {
75 let name = self.session_name(repo, task_id);
76
77 let mut cmd = Command::new("tmux");
78 cmd.args(["new-session", "-d", "-s", &name, "-c", working_dir]);
79
80 cmd.arg("-e").arg("DISABLE_AUTO_UPDATE=true");
82
83 for (key, value) in env_vars {
87 cmd.arg("-e").arg(format!("{}={}", key, value));
89 }
90
91 cmd.arg(command);
92
93 let output = cmd
94 .output_with_context()
95 .await
96 .context("spawning tmux session")?;
97
98 if !output.status.success() {
99 let stderr = String::from_utf8_lossy(&output.stderr);
100 anyhow::bail!("tmux new-session failed: {stderr}");
101 }
102
103 tracing::info!(session = %name, task_id, "created tmux session");
104 Ok(name)
105 }
106
107 pub async fn session_exists(&self, session: &str) -> bool {
109 Command::new("tmux")
110 .args(["has-session", "-t", session])
111 .output_with_context()
112 .await
113 .map(|o| o.status.success())
114 .unwrap_or(false)
115 }
116
117 pub async fn kill_session(&self, session: &str) -> anyhow::Result<()> {
119 let output = Command::new("tmux")
120 .args(["kill-session", "-t", session])
121 .output_with_context()
122 .await?;
123
124 if !output.status.success() {
125 let stderr = String::from_utf8_lossy(&output.stderr);
126 if stderr.contains("can't find session") || stderr.contains("no server running") {
127 tracing::debug!(session, %stderr, "kill-session: session already gone");
128 } else {
129 tracing::warn!(session, %stderr, "kill-session failed unexpectedly");
130 }
131 }
132 Ok(())
133 }
134
135 pub async fn capture_pane(&self, session: &str, lines: i32) -> anyhow::Result<String> {
141 let start = format!("-{lines}");
142 let output = Command::new("tmux")
143 .args(["capture-pane", "-t", session, "-p", "-S", &start])
144 .output_with_context()
145 .await?;
146
147 if !output.status.success() {
148 let stderr = String::from_utf8_lossy(&output.stderr);
149 if stderr.contains("can't find pane")
153 || stderr.contains("can't find session")
154 || stderr.contains("no server running")
155 {
156 tracing::debug!(
157 session,
158 "capture_pane: session/pane already gone (teardown race)"
159 );
160 return Ok(String::new());
161 }
162 anyhow::bail!("capture-pane failed for {session}: {stderr}");
163 }
164 Ok(String::from_utf8_lossy(&output.stdout).to_string())
165 }
166
167 pub async fn is_session_active(&self, session: &str) -> bool {
170 let output = Command::new("tmux")
172 .args(["list-panes", "-t", session, "-F", "#{pane_dead}"])
173 .output_with_context()
174 .await;
175
176 match output {
177 Ok(o) if o.status.success() => {
178 let flag = String::from_utf8_lossy(&o.stdout);
179 flag.trim() == "0" }
181 _ => false,
182 }
183 }
184
185 #[cfg(test)]
187 pub async fn session_is_running(&self, session: &str) -> bool {
188 self.session_exists(session).await && self.is_session_active(session).await
189 }
190
191 pub async fn session_blocks_dispatch(&self, session: &str) -> bool {
196 if !self.session_exists(session).await {
197 return false;
198 }
199
200 if self.is_session_active(session).await {
201 return true;
202 }
203
204 tracing::warn!(
205 session,
206 "found stale tmux session with dead pane; cleaning up"
207 );
208 if let Err(err) = self.kill_session(session).await {
209 tracing::debug!(session, error = %err, "failed to clean up stale tmux session");
210 }
211
212 false
213 }
214
215 pub async fn session_blocks_dispatch_from_map(
221 &self,
222 session: &str,
223 session_map: &std::collections::HashMap<String, bool>,
224 ) -> bool {
225 match session_map.get(session).copied() {
226 None => false, Some(true) => true, Some(false) => {
229 tracing::warn!(
231 session,
232 "found stale tmux session with dead pane; cleaning up"
233 );
234 if let Err(err) = self.kill_session(session).await {
235 tracing::debug!(
236 session,
237 error = %err,
238 "failed to clean up stale tmux session"
239 );
240 }
241 false
242 }
243 }
244 }
245
246 pub(crate) fn parse_session_name(&self, name: &str) -> Option<(String, String)> {
260 let after_prefix = name.strip_prefix(&self.prefix)?;
261
262 let (base, is_review) = match after_prefix.strip_suffix("-review") {
264 Some(stripped) => (stripped, true),
265 None => (after_prefix, false),
266 };
267
268 if let Some(idx) = base.rfind("-internal-") {
270 let project = &base[..idx];
271 let task_base = &base[idx + 1..]; let after_internal = &task_base["internal-".len()..];
273 if !project.is_empty()
274 && !after_internal.is_empty()
275 && after_internal.chars().all(|c| c.is_ascii_digit())
276 {
277 let task_id = if is_review {
278 format!("{task_base}-review")
279 } else {
280 task_base.to_string()
281 };
282 return Some((project.to_string(), task_id));
283 }
284 }
285
286 if let Some(idx) = base.rfind('-') {
288 let project = &base[..idx];
289 let digits = &base[idx + 1..];
290 if !project.is_empty()
291 && !digits.is_empty()
292 && digits.chars().all(|c| c.is_ascii_digit())
293 {
294 let task_id = if is_review {
295 format!("{digits}-review")
296 } else {
297 digits.to_string()
298 };
299 return Some((project.to_string(), task_id));
300 }
301 }
302
303 None
304 }
305
306 pub async fn batch_session_active(&self) -> HashMap<String, bool> {
311 let mut map = HashMap::new();
312 let o = match Command::new("tmux")
313 .args(["list-panes", "-a", "-F", "#{session_name} #{pane_dead}"])
314 .output_with_context()
315 .await
316 {
317 Ok(o) => o,
318 Err(e) => {
319 tracing::warn!(error = %e, "batch_session_active: failed to spawn tmux");
320 return map;
321 }
322 };
323 if !o.status.success() {
324 let stderr = String::from_utf8_lossy(&o.stderr);
325 tracing::warn!(
326 status = ?o.status,
327 stderr = %stderr.trim(),
328 "batch_session_active: tmux list-panes exited non-zero"
329 );
330 return map;
331 }
332 for line in String::from_utf8_lossy(&o.stdout).lines() {
333 let mut parts = line.trim().splitn(2, ' ');
334 if let (Some(session), Some(dead)) = (parts.next(), parts.next()) {
335 map.insert(session.to_string(), dead.trim() == "0");
336 }
337 }
338 map
339 }
340
341 pub async fn list_sessions(&self) -> anyhow::Result<Vec<Session>> {
343 let output = Command::new("tmux")
345 .args(["list-sessions", "-F", "#{session_name} #{session_created}"])
346 .output_with_context()
347 .await?;
348
349 if !output.status.success() {
350 return Ok(vec![]);
351 }
352
353 let mut sessions = Vec::new();
354 for line in String::from_utf8_lossy(&output.stdout).lines() {
355 let parts: Vec<&str> = line.trim().splitn(2, ' ').collect();
356 if parts.is_empty() {
357 continue;
358 }
359 let name = parts[0].to_string();
360 let created_at = parts.get(1).and_then(|s| s.parse::<u64>().ok());
361
362 if let Some((project, task_id)) = self.parse_session_name(&name) {
363 sessions.push(Session {
364 name,
365 task_id,
366 project,
367 created_at,
368 });
369 }
370 }
371
372 Ok(sessions)
373 }
374
375 pub async fn kill_stale_sessions(&self, before_timestamp: u64) -> anyhow::Result<usize> {
378 let sessions = self.list_sessions().await?;
379 let mut killed = 0;
380
381 for session in sessions {
382 let should_kill = match session.created_at {
383 Some(created) => created < before_timestamp,
384 None => false,
386 };
387
388 if should_kill {
389 tracing::info!(
390 session = %session.name,
391 created_at = session.created_at,
392 "killing stale tmux session from previous run"
393 );
394 if let Err(e) = self.kill_session(&session.name).await {
395 tracing::warn!(session = %session.name, error = %e, "failed to kill stale session");
396 } else {
397 killed += 1;
398 }
399 }
400 }
401
402 if killed > 0 {
403 tracing::info!(killed, "killed stale tmux sessions on startup");
404 }
405
406 Ok(killed)
407 }
408
409 pub async fn wait_for_sessions_dead(
413 &self,
414 sessions: &[String],
415 poll_interval: std::time::Duration,
416 timeout: std::time::Duration,
417 ) -> usize {
418 let start = std::time::Instant::now();
419 let mut remaining: std::collections::HashSet<String> = sessions.iter().cloned().collect();
420
421 while !remaining.is_empty() && start.elapsed() < timeout {
422 let mut to_remove = Vec::new();
423 for session in &remaining {
424 if !self.session_exists(session).await {
425 to_remove.push(session.clone());
426 }
427 }
428 for session in to_remove {
429 remaining.remove(&session);
430 }
431 if !remaining.is_empty() {
432 tokio::time::sleep(poll_interval).await;
433 }
434 }
435
436 let still_alive = remaining.len();
437 if still_alive > 0 {
438 tracing::warn!(
439 still_alive,
440 sessions = ?remaining.iter().collect::<Vec<_>>(),
441 "tmux sessions did not terminate within timeout"
442 );
443 }
444 still_alive
445 }
446
447 pub async fn wait_for_completion(
450 &self,
451 session: &str,
452 poll_interval: std::time::Duration,
453 ) -> anyhow::Result<String> {
454 loop {
455 if !self.session_exists(session).await {
456 return Ok(String::new());
457 }
458
459 if !self.is_session_active(session).await {
460 match self.capture_pane(session, 500).await {
464 Ok(output) => return Ok(output),
465 Err(err) => {
466 tracing::error!(session = %session, error = %err, "failed to capture final pane output");
467 return Ok(String::new());
468 }
469 }
470 }
471
472 tokio::time::sleep(poll_interval).await;
473 }
474 }
475
476 pub async fn snapshot(&self) -> Vec<(Session, bool)> {
484 let sessions = match self.list_sessions().await {
485 Ok(s) => s,
486 Err(err) => {
487 tracing::error!(error = %err, "failed to list tmux sessions — returning empty snapshot");
488 Vec::new()
489 }
490 };
491 let active_map = self.batch_session_active().await;
493 sessions
494 .into_iter()
495 .map(|s| {
496 let active = active_map.get(&s.name).copied().unwrap_or(false);
498 (s, active)
499 })
500 .collect()
501 }
502
503 #[cfg(test)]
511 pub async fn set_env(&self, session: &str, key: &str, value: &str) -> anyhow::Result<()> {
512 let output = Command::new("tmux")
513 .args(["set-environment", "-t", session, key, value])
514 .output_with_context()
515 .await?;
516
517 if !output.status.success() {
518 let stderr = String::from_utf8_lossy(&output.stderr);
519 anyhow::bail!("tmux set-environment failed for {session}: {stderr}");
520 }
521
522 tracing::debug!(session, key, "set tmux environment variable");
523 Ok(())
524 }
525
526 #[allow(dead_code)]
528 pub async fn unset_env(&self, session: &str, key: &str) -> anyhow::Result<()> {
529 let output = Command::new("tmux")
530 .args(["set-environment", "-t", session, "-u", key])
531 .output_with_context()
532 .await?;
533
534 if !output.status.success() {
535 let stderr = String::from_utf8_lossy(&output.stderr);
536 anyhow::bail!("tmux unset-environment failed for {session}: {stderr}");
537 }
538
539 tracing::debug!(session, key, "unset tmux environment variable");
540 Ok(())
541 }
542
543 pub async fn unset_global_env(&self, key: &str) -> anyhow::Result<()> {
547 let output = Command::new("tmux")
548 .args(["set-environment", "-gu", key])
549 .output_with_context()
550 .await?;
551
552 if !output.status.success() {
553 let stderr = String::from_utf8_lossy(&output.stderr);
554 anyhow::bail!("tmux unset-global-environment failed for {key}: {stderr}");
555 }
556
557 tracing::debug!(key, "unset tmux global environment variable");
558 Ok(())
559 }
560}
561
562#[cfg(test)]
563pub(crate) fn test_session_name() -> String {
564 format!(
565 "orch-test-{}",
566 std::time::SystemTime::now()
567 .duration_since(std::time::UNIX_EPOCH)
568 .map(|d| d.as_millis())
569 .unwrap_or_else(|_| 0)
570 )
571}
572
573#[cfg(test)]
575pub(crate) struct SessionGuard(pub(crate) String);
576
577#[cfg(test)]
578impl Drop for SessionGuard {
579 fn drop(&mut self) {
580 let _ = std::process::Command::new("tmux")
581 .args(["kill-session", "-t", &self.0])
582 .output();
583 }
584}
585
586#[cfg(test)]
587mod tests {
588 use super::*;
589
590 #[test]
591 fn session_name_internal_task_id_is_sanitized() {
592 let tmux = TmuxManager::new();
593 assert_eq!(
594 tmux.session_name("owner/repo", "internal:8"),
595 "orch-repo-internal-8"
596 );
597 }
598
599 #[test]
600 fn task_id_from_session_name_parses_external_task() {
601 let tmux = TmuxManager::new();
602 assert_eq!(
604 tmux.parse_session_name("orch-repo-1234"),
605 Some(("repo".to_string(), "1234".to_string()))
606 );
607 }
608
609 #[test]
610 fn task_id_from_session_name_parses_internal_task() {
611 let tmux = TmuxManager::new();
612 assert_eq!(
614 tmux.parse_session_name("orch-orch-internal-21116"),
615 Some(("orch".to_string(), "internal-21116".to_string()))
616 );
617 }
618
619 #[test]
620 fn task_id_from_session_name_parses_internal_task_small_id() {
621 let tmux = TmuxManager::new();
622 assert_eq!(
624 tmux.parse_session_name("orch-repo-internal-8"),
625 Some(("repo".to_string(), "internal-8".to_string()))
626 );
627 }
628
629 #[test]
630 fn task_id_from_session_name_returns_none_for_non_orch_session() {
631 let tmux = TmuxManager::new();
632 assert_eq!(tmux.parse_session_name("some-other-session"), None);
633 }
634
635 #[test]
636 fn parse_session_name_hyphenated_project_external() {
637 let tmux = TmuxManager::new();
638 assert_eq!(
640 tmux.parse_session_name("orch-my-repo-1234"),
641 Some(("my-repo".to_string(), "1234".to_string()))
642 );
643 }
644
645 #[test]
646 fn parse_session_name_hyphenated_project_internal() {
647 let tmux = TmuxManager::new();
648 assert_eq!(
650 tmux.parse_session_name("orch-my-repo-internal-8"),
651 Some(("my-repo".to_string(), "internal-8".to_string()))
652 );
653 }
654
655 #[test]
656 fn parse_session_name_hyphenated_project_review() {
657 let tmux = TmuxManager::new();
658 assert_eq!(
660 tmux.parse_session_name("orch-my-repo-1234-review"),
661 Some(("my-repo".to_string(), "1234-review".to_string()))
662 );
663 }
664
665 #[test]
666 fn parse_session_name_multi_hyphen_project() {
667 let tmux = TmuxManager::new();
668 assert_eq!(
670 tmux.parse_session_name("orch-my-cool-repo-42"),
671 Some(("my-cool-repo".to_string(), "42".to_string()))
672 );
673 }
674
675 #[test]
676 fn parse_session_name_roundtrip_hyphenated_project() {
677 let tmux = TmuxManager::new();
678 let name = tmux.session_name("owner/my-repo", "1793");
679 assert_eq!(name, "orch-my-repo-1793");
680 assert_eq!(
681 tmux.parse_session_name(&name),
682 Some(("my-repo".to_string(), "1793".to_string()))
683 );
684 }
685
686 #[tokio::test]
689 async fn test_set_env() {
690 let tmux = TmuxManager::new();
691 let session = test_session_name();
692
693 let create_result = tokio::process::Command::new("tmux")
695 .args(["new-session", "-d", "-s", &session, "-c", "/tmp"])
696 .output()
697 .await;
698
699 match create_result {
701 Ok(o) if o.status.success() => {}
702 _ => {
703 eprintln!("Skipping test: tmux not available or failed to create test session");
704 return;
705 }
706 }
707 let _guard = SessionGuard(session.clone());
708
709 let result = tmux.set_env(&session, "TEST_VAR", "test_value").await;
711 assert!(result.is_ok(), "set_env should succeed");
712
713 let check_result = tokio::process::Command::new("tmux")
715 .args(["show-environment", "-t", &session, "TEST_VAR"])
716 .output()
717 .await;
718
719 let output = match check_result {
720 Ok(o) => o,
721 Err(_) => {
722 eprintln!("Skipping test: unable to check tmux environment");
723 return;
724 }
725 };
726 let stdout = String::from_utf8_lossy(&output.stdout);
727 assert!(
728 stdout.contains("TEST_VAR=test_value"),
729 "Expected TEST_VAR=test_value, got: {}",
730 stdout
731 );
732 }
733
734 #[tokio::test]
738 async fn test_unset_env() {
739 let tmux = TmuxManager::new();
740 let session = test_session_name();
741
742 let create_result = tokio::process::Command::new("tmux")
744 .args(["new-session", "-d", "-s", &session, "-c", "/tmp"])
745 .output()
746 .await;
747
748 match create_result {
749 Ok(o) if o.status.success() => {}
750 _ => {
751 eprintln!("Skipping test: tmux not available or failed to create test session");
752 return;
753 }
754 }
755 let _guard = SessionGuard(session.clone());
756
757 let set_result = tmux.set_env(&session, "TO_DELETE", "temporary").await;
759 assert!(set_result.is_ok(), "set_env should succeed");
760
761 let check_before = tokio::process::Command::new("tmux")
763 .args(["show-environment", "-t", &session, "TO_DELETE"])
764 .output()
765 .await;
766
767 let check_before = match check_before {
768 Ok(o) => o,
769 Err(_) => {
770 eprintln!("Skipping test: unable to check tmux environment");
771 return;
772 }
773 };
774
775 assert!(
776 String::from_utf8_lossy(&check_before.stdout).contains("TO_DELETE"),
777 "Variable should exist before unset"
778 );
779
780 let unset_result = tmux.unset_env(&session, "TO_DELETE").await;
782 assert!(unset_result.is_ok(), "unset_env should succeed");
783 }
784
785 #[tokio::test]
786 async fn session_blocks_dispatch_keeps_live_session() {
787 let tmux = TmuxManager::new();
788 let session = test_session_name();
789
790 let create_result = tokio::process::Command::new("tmux")
791 .args(["new-session", "-d", "-s", &session, "-c", "/tmp"])
792 .output()
793 .await;
794
795 match create_result {
796 Ok(o) if o.status.success() => {}
797 _ => {
798 eprintln!("Skipping test: tmux not available or failed to create test session");
799 return;
800 }
801 }
802 let _guard = SessionGuard(session.clone());
803
804 assert!(tmux.session_blocks_dispatch(&session).await);
805 assert!(tmux.session_exists(&session).await);
806 }
807
808 #[tokio::test]
809 async fn session_blocks_dispatch_cleans_dead_session() {
810 let tmux = TmuxManager::new();
811 let session = test_session_name();
812
813 let create_result = tokio::process::Command::new("tmux")
814 .args(["new-session", "-d", "-s", &session, "-c", "/tmp"])
815 .output()
816 .await;
817
818 match create_result {
819 Ok(o) if o.status.success() => {}
820 _ => {
821 eprintln!("Skipping test: tmux not available or failed to create test session");
822 return;
823 }
824 }
825 let _guard = SessionGuard(session.clone());
826
827 let set_option_result = tokio::process::Command::new("tmux")
828 .args(["set-option", "-t", &session, "remain-on-exit", "on"])
829 .output()
830 .await;
831 if !matches!(set_option_result, Ok(ref o) if o.status.success()) {
832 eprintln!("Skipping test: unable to set tmux remain-on-exit option");
833 return;
834 }
835
836 let send_exit_result = tokio::process::Command::new("tmux")
837 .args(["send-keys", "-t", &session, "exit", "Enter"])
838 .output()
839 .await;
840 if !matches!(send_exit_result, Ok(ref o) if o.status.success()) {
841 eprintln!("Skipping test: unable to exit tmux pane");
842 return;
843 }
844
845 for _ in 0..20 {
847 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
848 if !tmux.session_is_running(&session).await {
849 break;
850 }
851 }
852
853 assert!(tmux.session_exists(&session).await);
854 assert!(!tmux.session_is_running(&session).await);
855 assert!(!tmux.session_blocks_dispatch(&session).await);
856 assert!(!tmux.session_exists(&session).await);
857 }
858
859 #[tokio::test]
863 async fn capture_pane_returns_empty_for_gone_session() {
864 let tmux = TmuxManager::new();
865 let result = tmux
867 .capture_pane("orch-nonexistent-session-99999", 100)
868 .await;
869 assert!(
870 result.is_ok(),
871 "capture_pane should return Ok for non-existent session, got: {:?}",
872 result
873 );
874 assert_eq!(
875 result.unwrap(),
876 "",
877 "capture_pane should return empty string for non-existent session"
878 );
879 }
880
881 #[tokio::test]
885 async fn wait_for_completion_returns_empty_for_gone_session() {
886 let tmux = TmuxManager::new();
887 let result = tmux
889 .wait_for_completion(
890 "orch-nonexistent-session-99999",
891 std::time::Duration::from_millis(50),
892 )
893 .await;
894 assert!(
895 result.is_ok(),
896 "wait_for_completion should return Ok for non-existent session, got: {:?}",
897 result
898 );
899 assert_eq!(
900 result.unwrap(),
901 "",
902 "wait_for_completion should return empty string for non-existent session"
903 );
904 }
905}