Integrate tmux capture into GNOME observer with three-state machine

+163 -50

observe/gnome/observer.py

··· 2 2 """ 3 3 Unified observer for audio and screencast capture. 4 4 5 - Continuously captures audio and manages screencast recording based on activity. 5 + Continuously captures audio and manages screencast/tmux recording based on activity. 6 6 Creates 5-minute windows, saving audio if voice activity detected and recording 7 - screencasts during active segments. Each monitor is recorded as a separate file. 7 + screencasts during active segments. When screen is idle but tmux sessions are active, 8 + captures tmux terminal content instead. 9 + 10 + State machine: 11 + SCREENCAST: Screen is active, recording video 12 + TMUX: Screen is idle but tmux has recent activity 13 + IDLE: Both screen and tmux are inactive 8 14 """ 9 15 10 16 import argparse ··· 28 34 ) 29 35 from observe.gnome.screencast import Screencaster, StreamInfo 30 36 from observe.hear import AudioRecorder 37 + from observe.tmux.capture import TmuxCapture, write_captures_jsonl 31 38 from think.callosum import CallosumConnection 32 39 from think.utils import day_path, setup_cli 33 40 ··· 39 46 MIN_HITS_FOR_SAVE = 3 40 47 CHUNK_DURATION = 5 # seconds 41 48 49 + # Capture modes 50 + MODE_IDLE = "idle" 51 + MODE_SCREENCAST = "screencast" 52 + MODE_TMUX = "tmux" 53 + 42 54 43 55 class Observer: 44 - """Unified audio and screencast observer.""" 56 + """Unified audio and screencast/tmux observer.""" 45 57 46 58 def __init__(self, interval: int = 300): 47 59 self.interval = interval 48 60 self.audio_recorder = AudioRecorder() 49 61 self.screencaster = Screencaster() 62 + self.tmux_capture = TmuxCapture() 50 63 self.bus: MessageBus | None = None 51 64 self.running = True 52 65 self.callosum: CallosumConnection | None = None ··· 56 69 self.start_at_mono = time.monotonic() # Monotonic for elapsed calculations 57 70 self.threshold_hits = 0 58 71 self.accumulated_audio_buffer = np.array([], dtype=np.float32).reshape(0, 2) 59 - self.screencast_running = False 72 + 73 + # Mode tracking (replaces screencast_running boolean) 74 + self.current_mode = MODE_IDLE 60 75 61 76 # Multi-file screencast tracking 62 77 self.current_streams: list[StreamInfo] = [] 63 78 self.pending_finalizations: list[tuple[str, str]] | None = None 64 79 self.last_screencast_sizes: dict[str, int] = {} 65 80 81 + # Tmux capture tracking 82 + self.tmux_captures: list[dict] = [] 83 + self.tmux_capture_id = 0 84 + self.tmux_sessions_seen: set[str] = set() 85 + 66 86 # Activity status cache (updated each loop) 67 87 self.cached_is_active = False 68 88 self.cached_idle_time_ms = 0 69 89 self.cached_screen_locked = False 70 90 self.cached_is_muted = False 71 91 self.cached_power_save = False 92 + self.cached_tmux_active = False 72 93 73 94 # Mute state at segment start (determines save format) 74 95 self.segment_is_muted = False ··· 96 117 return False 97 118 logger.info("Screencast portal connected") 98 119 120 + # Check tmux availability 121 + if self.tmux_capture.is_available(): 122 + logger.info("Tmux available for fallback capture") 123 + else: 124 + logger.info("Tmux not available (will only use screencast)") 125 + 99 126 # Start Callosum connection for status events 100 127 self.callosum = CallosumConnection() 101 128 self.callosum.start() ··· 103 130 104 131 return True 105 132 106 - async def check_activity_status(self) -> bool: 133 + async def check_activity_status(self) -> str: 107 134 """ 108 - Check system activity status and cache values. 135 + Check system activity status and determine capture mode. 109 136 110 137 Returns: 111 - True if user is active (not idle/locked/power-save, OR has audio activity) 138 + Capture mode: MODE_SCREENCAST, MODE_TMUX, or MODE_IDLE 112 139 """ 113 140 idle_time = await get_idle_time_ms(self.bus) 114 141 screen_locked = await is_screen_locked(self.bus) ··· 121 148 self.cached_is_muted = sink_muted 122 149 self.cached_power_save = power_save 123 150 124 - is_idle = (idle_time > IDLE_THRESHOLD_MS) or screen_locked or power_save 151 + # Determine screen activity 152 + screen_idle = (idle_time > IDLE_THRESHOLD_MS) or screen_locked or power_save 153 + screen_active = not screen_idle 154 + 155 + # Check tmux activity (only if screen is idle) 156 + if screen_active: 157 + tmux_active = False 158 + else: 159 + tmux_active = self.tmux_capture.is_active(poll_interval=CHUNK_DURATION) 160 + self.cached_tmux_active = tmux_active 161 + 162 + # Determine mode with priority: screen > tmux > idle 163 + if screen_active: 164 + mode = MODE_SCREENCAST 165 + elif tmux_active: 166 + mode = MODE_TMUX 167 + else: 168 + mode = MODE_IDLE 169 + 170 + # Cache legacy is_active for audio threshold logic 125 171 has_audio_activity = self.threshold_hits >= MIN_HITS_FOR_SAVE 126 - is_active = (not is_idle) or has_audio_activity 172 + self.cached_is_active = screen_active or tmux_active or has_audio_activity 127 173 128 - # Cache result 129 - self.cached_is_active = is_active 130 - 131 - return is_active 174 + return mode 132 175 133 176 def compute_rms(self, audio_buffer: np.ndarray) -> float: 134 177 """Compute per-channel RMS and return maximum (stereo: mic=left, sys=right).""" ··· 207 250 logger.info(f"Saved audio to {flac_path}") 208 251 return [audio_name] 209 252 210 - async def handle_boundary(self, is_active: bool): 253 + async def handle_boundary(self, new_mode: str): 211 254 """ 212 255 Handle window boundary rollover. 213 256 214 257 Args: 215 - is_active: Whether system is currently active 258 + new_mode: The mode for the new segment 216 259 """ 217 260 # Get timestamp parts for this window and calculate duration 218 261 date_part, time_part = self.get_timestamp_parts(self.start_at) ··· 239 282 self.accumulated_audio_buffer = np.array([], dtype=np.float32).reshape(0, 2) 240 283 self.threshold_hits = 0 241 284 242 - # Handle screencast rollover 285 + # Handle screencast rollover (if we were in screencast mode) 243 286 stopped_streams: list[StreamInfo] = [] 244 287 screen_files: list[str] = [] 245 288 246 - if self.screencast_running: 289 + if self.current_mode == MODE_SCREENCAST: 247 290 logger.info("Stopping previous screencast") 248 291 stopped_streams = await self.screencaster.stop() 249 - self.screencast_running = False 250 292 self.current_streams = [] 251 293 self.last_screencast_sizes = {} 252 294 ··· 261 303 if finalizations: 262 304 self.pending_finalizations = finalizations 263 305 306 + # Handle tmux capture save (if we were in tmux mode) 307 + tmux_files: list[str] = [] 308 + if self.current_mode == MODE_TMUX and self.tmux_captures: 309 + segment_key = f"{time_part}_{duration}" 310 + segment_dir = day_dir / segment_key 311 + tmux_files = write_captures_jsonl(self.tmux_captures, segment_dir) 312 + 313 + # Reset tmux state 314 + self.tmux_captures = [] 315 + self.tmux_capture_id = 0 316 + self.tmux_sessions_seen = set() 317 + self.tmux_capture.reset_hashes() 318 + 264 319 # Reset timing for new window 265 320 self.start_at = time.time() # Wall-clock for filenames 266 321 self.start_at_mono = time.monotonic() # Monotonic for elapsed ··· 268 323 # Update segment mute state for new segment 269 324 self.segment_is_muted = self.cached_is_muted 270 325 271 - # Start new screencast if active AND screen not locked 272 - # (is_active can be True due to audio activity even when locked) 273 - if is_active and not self.cached_screen_locked: 326 + # Update mode 327 + old_mode = self.current_mode 328 + self.current_mode = new_mode 329 + 330 + # Start new capture based on mode 331 + if new_mode == MODE_SCREENCAST and not self.cached_screen_locked: 274 332 await self.initialize_screencast() 333 + # MODE_TMUX doesn't need initialization, captures happen in main loop 334 + 335 + logger.info(f"Mode transition: {old_mode} -> {new_mode}") 275 336 276 337 # Emit observing event with what we saved this boundary 277 - files = audio_files + screen_files 338 + files = audio_files + screen_files + tmux_files 278 339 279 340 if files: 280 341 segment = f"{time_part}_{duration}" ··· 311 372 logger.error("No streams returned from screencast start") 312 373 raise RuntimeError("No streams available") 313 374 314 - self.screencast_running = True 315 375 self.current_streams = streams 316 376 self.last_screencast_sizes = {s.temp_path: 0 for s in streams} 317 377 ··· 321 381 322 382 return True 323 383 384 + def capture_tmux(self): 385 + """Poll tmux and accumulate captures for this chunk.""" 386 + active_sessions = self.tmux_capture.get_active_sessions(CHUNK_DURATION) 387 + if not active_sessions: 388 + return 389 + 390 + ts = time.time() 391 + 392 + for session_info in active_sessions: 393 + session = session_info["session"] 394 + self.tmux_sessions_seen.add(session) 395 + 396 + result = self.tmux_capture.capture_changed(session) 397 + if not result: 398 + continue 399 + 400 + self.tmux_capture_id += 1 401 + relative_ts = ts - self.start_at 402 + capture_dict = self.tmux_capture.result_to_dict( 403 + result, self.tmux_capture_id, relative_ts 404 + ) 405 + self.tmux_captures.append(capture_dict) 406 + logger.debug(f"Captured tmux session {session}: {len(result.panes)} panes") 407 + 324 408 def emit_status(self): 325 409 """Emit observe.status event with current state.""" 326 410 journal_path = os.getenv("JOURNAL_PATH", "") 411 + elapsed = int(time.monotonic() - self.start_at_mono) 327 412 328 413 # Calculate screencast info 329 - if self.screencast_running and self.current_streams: 330 - elapsed = int(time.monotonic() - self.start_at_mono) 414 + if self.current_mode == MODE_SCREENCAST and self.current_streams: 331 415 streams_info = [] 332 416 for stream in self.current_streams: 333 417 try: ··· 356 440 else: 357 441 screencast_info = {"recording": False, "files_growing": False} 358 442 443 + # Calculate tmux info 444 + if self.current_mode == MODE_TMUX: 445 + tmux_info = { 446 + "capturing": True, 447 + "captures": len(self.tmux_captures), 448 + "sessions": sorted(self.tmux_sessions_seen), 449 + "window_elapsed_seconds": elapsed, 450 + } 451 + else: 452 + tmux_info = {"capturing": False} 453 + 359 454 # Audio info 360 455 audio_info = { 361 456 "threshold_hits": self.threshold_hits, ··· 369 464 "screen_locked": self.cached_screen_locked, 370 465 "sink_muted": self.cached_is_muted, 371 466 "power_save": self.cached_power_save, 467 + "tmux_active": self.cached_tmux_active, 372 468 } 373 469 374 470 self.callosum.emit( 375 471 "observe", 376 472 "status", 473 + mode=self.current_mode, 377 474 screencast=screencast_info, 475 + tmux=tmux_info, 378 476 audio=audio_info, 379 477 activity=activity_info, 380 478 ) ··· 401 499 """Run the main observer loop.""" 402 500 logger.info(f"Starting observer loop (interval={self.interval}s)") 403 501 404 - # Start screencast immediately if active and screen not locked 405 - is_active = await self.check_activity_status() 502 + # Determine initial mode 503 + new_mode = await self.check_activity_status() 406 504 self.segment_is_muted = self.cached_is_muted # Sync initial mute state 407 - if is_active and not self.cached_screen_locked: 505 + self.current_mode = new_mode 506 + 507 + # Start initial capture based on mode 508 + if new_mode == MODE_SCREENCAST and not self.cached_screen_locked: 408 509 try: 409 510 await self.initialize_screencast() 410 511 except RuntimeError: 411 512 # Failed to start screencast, exit 412 513 self.running = False 413 514 return 515 + 516 + logger.info(f"Initial mode: {self.current_mode}") 414 517 415 518 while self.running: 416 519 # Sleep for chunk duration ··· 425 528 logger.warning(f"Pending screencast not found: {temp_path}") 426 529 self.pending_finalizations = None 427 530 428 - # Check activity status 429 - is_active = await self.check_activity_status() 531 + # Check activity status and determine new mode 532 + new_mode = await self.check_activity_status() 430 533 431 534 # Check for GStreamer failure mid-recording 432 - if self.screencast_running and not self.screencaster.is_healthy(): 535 + if self.current_mode == MODE_SCREENCAST and not self.screencaster.is_healthy(): 433 536 logger.warning("Screencast recording failed, stopping gracefully") 434 537 stopped_streams = await self.screencaster.stop() 435 - self.screencast_running = False 436 538 437 539 # Finalize whatever we have 438 540 if stopped_streams: ··· 449 551 450 552 self.current_streams = [] 451 553 self.last_screencast_sizes = {} 554 + # Force recalculate mode without screencast 555 + self.current_mode = MODE_IDLE 452 556 453 - # Transition from idle to active 454 - activation_edge = is_active and not self.screencast_running 557 + # Detect mode transition 558 + mode_transition = new_mode != self.current_mode 559 + if mode_transition: 560 + logger.info(f"Mode changing: {self.current_mode} -> {new_mode}") 455 561 456 562 # Detect mute state transition 457 563 mute_transition = self.cached_is_muted != self.segment_is_muted ··· 483 589 else: 484 590 logger.debug("No audio data in chunk") 485 591 592 + # Capture tmux if in tmux mode 593 + if self.current_mode == MODE_TMUX: 594 + self.capture_tmux() 595 + 486 596 # Check for window boundary (use monotonic to avoid DST/clock jumps) 487 597 now_mono = time.monotonic() 488 598 elapsed = now_mono - self.start_at_mono 489 599 is_boundary = ( 490 - (elapsed >= self.interval) or activation_edge or mute_transition 600 + (elapsed >= self.interval) or mode_transition or mute_transition 491 601 ) 492 602 493 603 if is_boundary: 494 604 logger.info( 495 - f"Boundary: elapsed={elapsed:.1f}s edge={activation_edge} " 605 + f"Boundary: elapsed={elapsed:.1f}s mode_change={mode_transition} " 496 606 f"mute_change={mute_transition} " 497 607 f"hits={self.threshold_hits}/{MIN_HITS_FOR_SAVE}" 498 608 ) 499 - await self.handle_boundary(is_active) 609 + await self.handle_boundary(new_mode) 500 610 501 611 # Check if screencast files are actively growing (for health reporting) 502 - if self.screencast_running and self.current_streams: 612 + if self.current_mode == MODE_SCREENCAST and self.current_streams: 503 613 any_growing = False 504 614 for stream in self.current_streams: 505 615 if os.path.exists(stream.temp_path): ··· 521 631 522 632 async def shutdown(self): 523 633 """Clean shutdown of observer.""" 524 - # Perform final boundary logic without restarting screencast 525 - if self.threshold_hits >= MIN_HITS_FOR_SAVE: 526 - date_part, time_part = self.get_timestamp_parts(self.start_at) 527 - duration = int(time.time() - self.start_at) 528 - day_dir = day_path(date_part) 634 + # Get timestamp parts for final save 635 + date_part, time_part = self.get_timestamp_parts(self.start_at) 636 + duration = int(time.time() - self.start_at) 637 + day_dir = day_path(date_part) 529 638 639 + # Save final audio if threshold met 640 + if self.threshold_hits >= MIN_HITS_FOR_SAVE: 530 641 audio_files = self._save_audio_segment( 531 642 day_dir, time_part, duration, self.segment_is_muted 532 643 ) ··· 534 645 logger.info(f"Saved final audio: {len(audio_files)} file(s)") 535 646 536 647 # Stop screencast if running 537 - if self.screencast_running: 648 + if self.current_mode == MODE_SCREENCAST: 538 649 logger.info("Stopping screencast for shutdown") 539 650 stopped_streams = await self.screencaster.stop() 540 651 ··· 542 653 # Brief delay for files to be written 543 654 await asyncio.sleep(0.5) 544 655 545 - duration = int(time.time() - self.start_at) 546 - date_part, time_part = self.get_timestamp_parts(self.start_at) 547 - day_dir = day_path(date_part) 548 - 549 656 for stream in stopped_streams: 550 657 if os.path.exists(stream.temp_path): 551 658 final_path = str( ··· 557 664 f"Screencast file not found after shutdown: {stream.temp_path}" 558 665 ) 559 666 560 - self.screencast_running = False 667 + # Save tmux captures if in tmux mode 668 + if self.current_mode == MODE_TMUX and self.tmux_captures: 669 + segment_key = f"{time_part}_{duration}" 670 + segment_dir = day_dir / segment_key 671 + tmux_files = write_captures_jsonl(self.tmux_captures, segment_dir) 672 + if tmux_files: 673 + logger.info(f"Saved final tmux captures: {len(tmux_files)} file(s)") 561 674 562 675 # Process any remaining pending finalizations 563 676 if self.pending_finalizations: ··· 616 729 def main(): 617 730 """CLI entry point.""" 618 731 parser = argparse.ArgumentParser( 619 - description="Unified audio and screencast observer for journaling." 732 + description="Unified audio, screencast, and tmux observer for journaling." 620 733 ) 621 734 parser.add_argument( 622 735 "--interval",

+18

observe/tmux/__init__.py

··· 1 1 """Tmux terminal capture for observe package.""" 2 + 3 + from observe.tmux.capture import ( 4 + CaptureResult, 5 + PaneInfo, 6 + TmuxCapture, 7 + WindowInfo, 8 + run_tmux_command, 9 + write_captures_jsonl, 10 + ) 11 + 12 + __all__ = [ 13 + "TmuxCapture", 14 + "CaptureResult", 15 + "PaneInfo", 16 + "WindowInfo", 17 + "run_tmux_command", 18 + "write_captures_jsonl", 19 + ]

+382

observe/tmux/capture.py

··· 1 + """Tmux terminal capture library. 2 + 3 + Provides functions for capturing tmux session content, designed for use 4 + by the GNOME observer for fallback capture when screen is idle. 5 + """ 6 + 7 + import hashlib 8 + import json 9 + import logging 10 + import subprocess 11 + import time 12 + from dataclasses import dataclass 13 + from pathlib import Path 14 + 15 + logger = logging.getLogger(__name__) 16 + 17 + 18 + @dataclass 19 + class PaneInfo: 20 + """Information about a tmux pane.""" 21 + 22 + id: str 23 + index: int 24 + left: int 25 + top: int 26 + width: int 27 + height: int 28 + active: bool 29 + content: str = "" 30 + 31 + 32 + @dataclass 33 + class WindowInfo: 34 + """Information about a tmux window.""" 35 + 36 + id: str 37 + index: int 38 + name: str 39 + active: bool 40 + 41 + 42 + @dataclass 43 + class CaptureResult: 44 + """Result of capturing a session's active window.""" 45 + 46 + session: str 47 + window: WindowInfo 48 + windows: list[WindowInfo] 49 + panes: list[PaneInfo] 50 + 51 + 52 + def run_tmux_command(args: list[str]) -> str | None: 53 + """Run a tmux command and return stdout, or None on error.""" 54 + try: 55 + result = subprocess.run( 56 + ["tmux"] + args, 57 + capture_output=True, 58 + text=True, 59 + timeout=5, 60 + ) 61 + if result.returncode != 0: 62 + return None 63 + return result.stdout 64 + except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e: 65 + logger.debug(f"tmux command failed: {e}") 66 + return None 67 + 68 + 69 + class TmuxCapture: 70 + """Tmux terminal capture with deduplication.""" 71 + 72 + def __init__(self): 73 + # Deduplication: session -> hash of last capture 74 + self.last_hash: dict[str, str] = {} 75 + 76 + def reset_hashes(self): 77 + """Reset deduplication hashes (call at segment boundary).""" 78 + self.last_hash.clear() 79 + 80 + def is_available(self) -> bool: 81 + """Check if tmux is available on this system.""" 82 + return run_tmux_command(["list-sessions"]) is not None 83 + 84 + def is_active(self, poll_interval: float = 5.0) -> bool: 85 + """Check if any tmux sessions have recent activity. 86 + 87 + Args: 88 + poll_interval: Consider sessions active if client activity 89 + within this many seconds. 90 + 91 + Returns: 92 + True if any active sessions found. 93 + """ 94 + return len(self.get_active_sessions(poll_interval)) > 0 95 + 96 + def get_active_sessions(self, poll_interval: float = 5.0) -> list[dict]: 97 + """Get sessions with recent client activity. 98 + 99 + Args: 100 + poll_interval: Consider sessions active if client activity 101 + within this many seconds. 102 + 103 + Returns: 104 + List of dicts with 'session' and 'activity' keys. 105 + """ 106 + output = run_tmux_command([ 107 + "list-clients", 108 + "-F", "#{client_session} #{client_activity}", 109 + ]) 110 + if not output: 111 + return [] 112 + 113 + now = time.time() 114 + active = [] 115 + seen_sessions = set() 116 + 117 + for line in output.strip().split("\n"): 118 + if not line: 119 + continue 120 + parts = line.split(" ", 1) 121 + if len(parts) != 2: 122 + continue 123 + 124 + session, activity_str = parts 125 + try: 126 + activity = int(activity_str) 127 + except ValueError: 128 + continue 129 + 130 + # Check if active within poll interval 131 + if now - activity <= poll_interval and session not in seen_sessions: 132 + active.append({"session": session, "activity": activity}) 133 + seen_sessions.add(session) 134 + 135 + return active 136 + 137 + def get_windows(self, session: str) -> list[WindowInfo]: 138 + """Get all windows for a session.""" 139 + output = run_tmux_command([ 140 + "list-windows", 141 + "-t", session, 142 + "-F", "#{window_active} #{window_id} #{window_index} #{window_name}", 143 + ]) 144 + if not output: 145 + return [] 146 + 147 + windows = [] 148 + for line in output.strip().split("\n"): 149 + if not line: 150 + continue 151 + parts = line.split(" ", 3) 152 + if len(parts) < 4: 153 + continue 154 + 155 + active_str, window_id, index_str, name = parts 156 + try: 157 + windows.append(WindowInfo( 158 + id=window_id, 159 + index=int(index_str), 160 + name=name, 161 + active=(active_str == "1"), 162 + )) 163 + except ValueError: 164 + continue 165 + 166 + return windows 167 + 168 + def get_panes(self, window_id: str) -> list[PaneInfo]: 169 + """Get all panes for a window with layout info.""" 170 + output = run_tmux_command([ 171 + "list-panes", 172 + "-t", window_id, 173 + "-F", "#{pane_id} #{pane_index} #{pane_left} #{pane_top} #{pane_width} #{pane_height} #{pane_active}", 174 + ]) 175 + if not output: 176 + return [] 177 + 178 + panes = [] 179 + for line in output.strip().split("\n"): 180 + if not line: 181 + continue 182 + parts = line.split(" ") 183 + if len(parts) != 7: 184 + continue 185 + 186 + try: 187 + panes.append(PaneInfo( 188 + id=parts[0], 189 + index=int(parts[1]), 190 + left=int(parts[2]), 191 + top=int(parts[3]), 192 + width=int(parts[4]), 193 + height=int(parts[5]), 194 + active=(parts[6] == "1"), 195 + )) 196 + except ValueError: 197 + continue 198 + 199 + return panes 200 + 201 + def capture_pane(self, pane_id: str) -> str: 202 + """Capture visible pane content with ANSI escape codes.""" 203 + output = run_tmux_command([ 204 + "capture-pane", 205 + "-p", # Print to stdout 206 + "-e", # Include escape sequences (ANSI codes) 207 + "-t", pane_id, 208 + ]) 209 + return output if output else "" 210 + 211 + def capture_session(self, session: str) -> CaptureResult | None: 212 + """Capture the active window of a session with all its panes. 213 + 214 + Returns None if session doesn't exist or has no active window. 215 + """ 216 + windows = self.get_windows(session) 217 + if not windows: 218 + return None 219 + 220 + # Find active window 221 + active_window = next((w for w in windows if w.active), None) 222 + if not active_window: 223 + return None 224 + 225 + # Get panes for active window 226 + panes = self.get_panes(active_window.id) 227 + if not panes: 228 + return None 229 + 230 + # Capture content for each pane 231 + for pane in panes: 232 + pane.content = self.capture_pane(pane.id) 233 + 234 + return CaptureResult( 235 + session=session, 236 + window=active_window, 237 + windows=windows, 238 + panes=panes, 239 + ) 240 + 241 + def compute_hash(self, result: CaptureResult) -> str: 242 + """Compute hash of capture for deduplication.""" 243 + # Hash window id + all pane contents 244 + parts = [result.window.id] 245 + for pane in sorted(result.panes, key=lambda p: p.id): 246 + parts.append(pane.content) 247 + content = "\n".join(parts) 248 + return hashlib.md5(content.encode()).hexdigest() 249 + 250 + def capture_changed(self, session: str) -> CaptureResult | None: 251 + """Capture session if content changed since last capture. 252 + 253 + Uses internal hash tracking for deduplication. 254 + 255 + Returns: 256 + CaptureResult if content changed, None if unchanged or error. 257 + """ 258 + result = self.capture_session(session) 259 + if not result: 260 + return None 261 + 262 + content_hash = self.compute_hash(result) 263 + if self.last_hash.get(session) == content_hash: 264 + logger.debug(f"Session {session} unchanged, skipping") 265 + return None 266 + 267 + self.last_hash[session] = content_hash 268 + return result 269 + 270 + def result_to_dict( 271 + self, result: CaptureResult, capture_id: int, relative_ts: float 272 + ) -> dict: 273 + """Convert CaptureResult to JSON-serializable dict. 274 + 275 + Output format matches screen.jsonl structure from observe-describe: 276 + - frame_id: Capture sequence number 277 + - timestamp: Seconds since segment start 278 + - requests: Empty list (no AI processing) 279 + - analysis: Category info with templated visual_description 280 + - tmux: All terminal-specific data 281 + 282 + Args: 283 + result: Capture result to convert. 284 + capture_id: Sequential frame ID for this capture. 285 + relative_ts: Timestamp relative to segment start. 286 + 287 + Returns: 288 + Dict ready for JSON serialization. 289 + """ 290 + # Build visual description from tmux info 291 + pane_count = len(result.panes) 292 + pane_word = "pane" if pane_count == 1 else "panes" 293 + visual_description = ( 294 + f"Terminal session '{result.session}' with {pane_count} {pane_word} " 295 + f"in window '{result.window.name}'" 296 + ) 297 + 298 + return { 299 + "frame_id": capture_id, 300 + "timestamp": relative_ts, 301 + "requests": [], 302 + "analysis": { 303 + "visual_description": visual_description, 304 + "primary": "tmux", 305 + "secondary": "none", 306 + "overlap": False, 307 + }, 308 + "tmux": { 309 + "session": result.session, 310 + "window": { 311 + "id": result.window.id, 312 + "index": result.window.index, 313 + "name": result.window.name, 314 + }, 315 + "windows": [ 316 + {"id": w.id, "index": w.index, "name": w.name, "active": w.active} 317 + for w in result.windows 318 + ], 319 + "panes": [ 320 + { 321 + "id": p.id, 322 + "index": p.index, 323 + "left": p.left, 324 + "top": p.top, 325 + "width": p.width, 326 + "height": p.height, 327 + "active": p.active, 328 + "content": p.content, 329 + } 330 + for p in result.panes 331 + ], 332 + }, 333 + } 334 + 335 + 336 + def write_captures_jsonl(captures: list[dict], segment_dir: Path) -> list[str]: 337 + """Write tmux captures to JSONL files, grouped by session. 338 + 339 + Creates one file per session: tmux_{session}_screen.jsonl 340 + Format matches screen.jsonl for unified formatting/indexing. 341 + 342 + Args: 343 + captures: List of capture dicts from result_to_dict() 344 + segment_dir: Directory to write files to (created if needed) 345 + 346 + Returns: 347 + List of filenames written (empty if no captures) 348 + """ 349 + if not captures: 350 + return [] 351 + 352 + segment_dir.mkdir(parents=True, exist_ok=True) 353 + 354 + # Group captures by session 355 + by_session: dict[str, list[dict]] = {} 356 + for capture in captures: 357 + session = capture.get("tmux", {}).get("session", "unknown") 358 + if session not in by_session: 359 + by_session[session] = [] 360 + by_session[session].append(capture) 361 + 362 + # Write one file per session 363 + files_written = [] 364 + for session, session_captures in by_session.items(): 365 + # Sanitize session name for filename 366 + safe_session = session.replace("/", "_").replace(" ", "_") 367 + filename = f"tmux_{safe_session}_screen.jsonl" 368 + output_path = segment_dir / filename 369 + 370 + with open(output_path, "w") as f: 371 + # Header matching screen.jsonl format 372 + header = {"raw": filename} 373 + f.write(json.dumps(header) + "\n") 374 + 375 + # Write each capture 376 + for capture in session_captures: 377 + f.write(json.dumps(capture) + "\n") 378 + 379 + files_written.append(filename) 380 + logger.info(f"Wrote {len(session_captures)} tmux captures to {output_path}") 381 + 382 + return files_written

-525

observe/tmux/observer.py

··· 1 - #!/usr/bin/env python3 2 - """ 3 - Tmux terminal observer for journaling. 4 - 5 - Polls tmux for active sessions and captures terminal content from the active 6 - window's panes. Outputs tmux_{session}_screen.jsonl files in screen.jsonl- 7 - compatible format for unified formatting and indexing. 8 - """ 9 - 10 - import argparse 11 - import hashlib 12 - import json 13 - import logging 14 - import os 15 - import signal 16 - import subprocess 17 - import time 18 - from dataclasses import dataclass 19 - 20 - from think.callosum import CallosumConnection 21 - from think.utils import day_path, setup_cli 22 - 23 - logger = logging.getLogger(__name__) 24 - 25 - # Constants 26 - DEFAULT_INTERVAL = 300 # 5 minutes 27 - DEFAULT_POLL_INTERVAL = 5 # seconds 28 - 29 - 30 - @dataclass 31 - class PaneInfo: 32 - """Information about a tmux pane.""" 33 - 34 - id: str 35 - index: int 36 - left: int 37 - top: int 38 - width: int 39 - height: int 40 - active: bool 41 - content: str = "" 42 - 43 - 44 - @dataclass 45 - class WindowInfo: 46 - """Information about a tmux window.""" 47 - 48 - id: str 49 - index: int 50 - name: str 51 - active: bool 52 - 53 - 54 - @dataclass 55 - class CaptureResult: 56 - """Result of capturing a session's active window.""" 57 - 58 - session: str 59 - window: WindowInfo 60 - windows: list[WindowInfo] 61 - panes: list[PaneInfo] 62 - 63 - 64 - def run_tmux_command(args: list[str]) -> str | None: 65 - """Run a tmux command and return stdout, or None on error.""" 66 - try: 67 - result = subprocess.run( 68 - ["tmux"] + args, 69 - capture_output=True, 70 - text=True, 71 - timeout=5, 72 - ) 73 - if result.returncode != 0: 74 - return None 75 - return result.stdout 76 - except (subprocess.TimeoutExpired, FileNotFoundError, OSError) as e: 77 - logger.debug(f"tmux command failed: {e}") 78 - return None 79 - 80 - 81 - class TmuxObserver: 82 - """Observer for tmux terminal activity.""" 83 - 84 - def __init__(self, interval: int = DEFAULT_INTERVAL, poll_interval: int = DEFAULT_POLL_INTERVAL): 85 - self.interval = interval 86 - self.poll_interval = poll_interval 87 - self.running = True 88 - 89 - # Segment state 90 - self.start_at = time.time() 91 - self.captures: list[dict] = [] 92 - self.capture_id = 0 93 - self.sessions_seen: set[str] = set() 94 - 95 - # Deduplication: session -> hash of last capture 96 - self.last_hash: dict[str, str] = {} 97 - 98 - # Callosum for status events 99 - self.callosum: CallosumConnection | None = None 100 - 101 - def get_active_sessions(self) -> list[dict]: 102 - """Get sessions with recent client activity. 103 - 104 - Returns list of dicts with 'session' and 'activity' keys. 105 - """ 106 - output = run_tmux_command([ 107 - "list-clients", 108 - "-F", "#{client_session} #{client_activity}", 109 - ]) 110 - if not output: 111 - return [] 112 - 113 - now = time.time() 114 - active = [] 115 - seen_sessions = set() 116 - 117 - for line in output.strip().split("\n"): 118 - if not line: 119 - continue 120 - parts = line.split(" ", 1) 121 - if len(parts) != 2: 122 - continue 123 - 124 - session, activity_str = parts 125 - try: 126 - activity = int(activity_str) 127 - except ValueError: 128 - continue 129 - 130 - # Check if active within poll interval 131 - if now - activity <= self.poll_interval and session not in seen_sessions: 132 - active.append({"session": session, "activity": activity}) 133 - seen_sessions.add(session) 134 - 135 - return active 136 - 137 - def get_windows(self, session: str) -> list[WindowInfo]: 138 - """Get all windows for a session.""" 139 - output = run_tmux_command([ 140 - "list-windows", 141 - "-t", session, 142 - "-F", "#{window_active} #{window_id} #{window_index} #{window_name}", 143 - ]) 144 - if not output: 145 - return [] 146 - 147 - windows = [] 148 - for line in output.strip().split("\n"): 149 - if not line: 150 - continue 151 - parts = line.split(" ", 3) 152 - if len(parts) < 4: 153 - continue 154 - 155 - active_str, window_id, index_str, name = parts 156 - try: 157 - windows.append(WindowInfo( 158 - id=window_id, 159 - index=int(index_str), 160 - name=name, 161 - active=(active_str == "1"), 162 - )) 163 - except ValueError: 164 - continue 165 - 166 - return windows 167 - 168 - def get_panes(self, window_id: str) -> list[PaneInfo]: 169 - """Get all panes for a window with layout info.""" 170 - output = run_tmux_command([ 171 - "list-panes", 172 - "-t", window_id, 173 - "-F", "#{pane_id} #{pane_index} #{pane_left} #{pane_top} #{pane_width} #{pane_height} #{pane_active}", 174 - ]) 175 - if not output: 176 - return [] 177 - 178 - panes = [] 179 - for line in output.strip().split("\n"): 180 - if not line: 181 - continue 182 - parts = line.split(" ") 183 - if len(parts) != 7: 184 - continue 185 - 186 - try: 187 - panes.append(PaneInfo( 188 - id=parts[0], 189 - index=int(parts[1]), 190 - left=int(parts[2]), 191 - top=int(parts[3]), 192 - width=int(parts[4]), 193 - height=int(parts[5]), 194 - active=(parts[6] == "1"), 195 - )) 196 - except ValueError: 197 - continue 198 - 199 - return panes 200 - 201 - def capture_pane(self, pane_id: str) -> str: 202 - """Capture visible pane content with ANSI escape codes.""" 203 - output = run_tmux_command([ 204 - "capture-pane", 205 - "-p", # Print to stdout 206 - "-e", # Include escape sequences (ANSI codes) 207 - "-t", pane_id, 208 - ]) 209 - return output if output else "" 210 - 211 - def capture_session(self, session: str) -> CaptureResult | None: 212 - """Capture the active window of a session with all its panes. 213 - 214 - Returns None if session doesn't exist or has no active window. 215 - """ 216 - windows = self.get_windows(session) 217 - if not windows: 218 - return None 219 - 220 - # Find active window 221 - active_window = next((w for w in windows if w.active), None) 222 - if not active_window: 223 - return None 224 - 225 - # Get panes for active window 226 - panes = self.get_panes(active_window.id) 227 - if not panes: 228 - return None 229 - 230 - # Capture content for each pane 231 - for pane in panes: 232 - pane.content = self.capture_pane(pane.id) 233 - 234 - return CaptureResult( 235 - session=session, 236 - window=active_window, 237 - windows=windows, 238 - panes=panes, 239 - ) 240 - 241 - def compute_hash(self, result: CaptureResult) -> str: 242 - """Compute hash of capture for deduplication.""" 243 - # Hash window id + all pane contents 244 - parts = [result.window.id] 245 - for pane in sorted(result.panes, key=lambda p: p.id): 246 - parts.append(pane.content) 247 - content = "\n".join(parts) 248 - return hashlib.md5(content.encode()).hexdigest() 249 - 250 - def result_to_dict(self, result: CaptureResult, ts: float) -> dict: 251 - """Convert CaptureResult to JSON-serializable dict. 252 - 253 - Output format matches screen.jsonl structure from observe-describe: 254 - - frame_id: Capture sequence number 255 - - timestamp: Seconds since segment start 256 - - requests: Empty list (no AI processing) 257 - - analysis: Category info with templated visual_description 258 - - tmux: All terminal-specific data 259 - """ 260 - self.capture_id += 1 261 - 262 - # Calculate relative timestamp from segment start 263 - relative_ts = ts - self.start_at 264 - 265 - # Build visual description from tmux info 266 - pane_count = len(result.panes) 267 - pane_word = "pane" if pane_count == 1 else "panes" 268 - visual_description = ( 269 - f"Terminal session '{result.session}' with {pane_count} {pane_word} " 270 - f"in window '{result.window.name}'" 271 - ) 272 - 273 - return { 274 - "frame_id": self.capture_id, 275 - "timestamp": relative_ts, 276 - "requests": [], 277 - "analysis": { 278 - "visual_description": visual_description, 279 - "primary": "tmux", 280 - "secondary": "none", 281 - "overlap": False, 282 - }, 283 - "tmux": { 284 - "session": result.session, 285 - "window": { 286 - "id": result.window.id, 287 - "index": result.window.index, 288 - "name": result.window.name, 289 - }, 290 - "windows": [ 291 - {"id": w.id, "index": w.index, "name": w.name, "active": w.active} 292 - for w in result.windows 293 - ], 294 - "panes": [ 295 - { 296 - "id": p.id, 297 - "index": p.index, 298 - "left": p.left, 299 - "top": p.top, 300 - "width": p.width, 301 - "height": p.height, 302 - "active": p.active, 303 - "content": p.content, 304 - } 305 - for p in result.panes 306 - ], 307 - }, 308 - } 309 - 310 - def poll_and_capture(self) -> list[dict]: 311 - """Poll for active sessions and capture changed content. 312 - 313 - Returns list of capture dicts for sessions that changed. 314 - """ 315 - active_sessions = self.get_active_sessions() 316 - if not active_sessions: 317 - return [] 318 - 319 - ts = time.time() 320 - new_captures = [] 321 - 322 - for session_info in active_sessions: 323 - session = session_info["session"] 324 - self.sessions_seen.add(session) 325 - 326 - result = self.capture_session(session) 327 - if not result: 328 - continue 329 - 330 - # Check if content changed 331 - content_hash = self.compute_hash(result) 332 - if self.last_hash.get(session) == content_hash: 333 - logger.debug(f"Session {session} unchanged, skipping") 334 - continue 335 - 336 - self.last_hash[session] = content_hash 337 - capture_dict = self.result_to_dict(result, ts) 338 - new_captures.append(capture_dict) 339 - logger.debug(f"Captured session {session}: {len(result.panes)} panes") 340 - 341 - return new_captures 342 - 343 - def handle_boundary(self): 344 - """Write accumulated captures to segment JSONL files and reset. 345 - 346 - Writes one file per session: tmux_{session}_screen.jsonl 347 - Format matches screen.jsonl for unified formatting/indexing. 348 - """ 349 - if not self.captures: 350 - logger.info("No captures in segment, skipping write") 351 - self.reset_segment() 352 - return 353 - 354 - # Build segment path 355 - start_dt = time.localtime(self.start_at) 356 - date_part = time.strftime("%Y%m%d", start_dt) 357 - time_part = time.strftime("%H%M%S", start_dt) 358 - duration = int(time.time() - self.start_at) 359 - segment_key = f"{time_part}_{duration}" 360 - 361 - segment_dir = day_path(date_part) / segment_key 362 - segment_dir.mkdir(parents=True, exist_ok=True) 363 - 364 - # Group captures by session 365 - by_session: dict[str, list[dict]] = {} 366 - for capture in self.captures: 367 - session = capture.get("tmux", {}).get("session", "unknown") 368 - if session not in by_session: 369 - by_session[session] = [] 370 - by_session[session].append(capture) 371 - 372 - # Write one file per session 373 - files_written = [] 374 - for session, captures in by_session.items(): 375 - # Sanitize session name for filename 376 - safe_session = session.replace("/", "_").replace(" ", "_") 377 - filename = f"tmux_{safe_session}_screen.jsonl" 378 - output_path = segment_dir / filename 379 - 380 - with open(output_path, "w") as f: 381 - # Header matching screen.jsonl format 382 - header = {"raw": filename} 383 - f.write(json.dumps(header) + "\n") 384 - 385 - # Write each capture 386 - for capture in captures: 387 - f.write(json.dumps(capture) + "\n") 388 - 389 - files_written.append(filename) 390 - logger.info(f"Wrote {len(captures)} captures to {output_path}") 391 - 392 - # Emit event 393 - if self.callosum: 394 - self.callosum.emit( 395 - "observe", 396 - "tmux_captured", 397 - segment=segment_key, 398 - captures=len(self.captures), 399 - sessions=sorted(self.sessions_seen), 400 - files=files_written, 401 - ) 402 - 403 - self.reset_segment() 404 - 405 - def reset_segment(self): 406 - """Reset state for new segment.""" 407 - self.start_at = time.time() 408 - self.captures = [] 409 - self.capture_id = 0 410 - self.sessions_seen = set() 411 - # Keep last_hash for cross-segment deduplication 412 - 413 - def emit_status(self): 414 - """Emit observe.status event for health monitoring.""" 415 - if not self.callosum: 416 - return 417 - 418 - elapsed = int(time.time() - self.start_at) 419 - self.callosum.emit( 420 - "observe", 421 - "status", 422 - tmux={ 423 - "captures": len(self.captures), 424 - "sessions": sorted(self.sessions_seen), 425 - "window_elapsed_seconds": elapsed, 426 - }, 427 - ) 428 - 429 - def main_loop(self): 430 - """Run the main observer loop.""" 431 - logger.info( 432 - f"Starting tmux observer (interval={self.interval}s, poll={self.poll_interval}s)" 433 - ) 434 - 435 - # Start Callosum connection 436 - self.callosum = CallosumConnection() 437 - self.callosum.start() 438 - 439 - last_status_emit = 0.0 440 - 441 - while self.running: 442 - # Poll and capture 443 - new_captures = self.poll_and_capture() 444 - self.captures.extend(new_captures) 445 - 446 - # Check for segment boundary 447 - elapsed = time.time() - self.start_at 448 - if elapsed >= self.interval: 449 - logger.info(f"Segment boundary at {elapsed:.1f}s") 450 - self.handle_boundary() 451 - 452 - # Emit status every 5 seconds 453 - now = time.time() 454 - if now - last_status_emit >= 5: 455 - self.emit_status() 456 - last_status_emit = now 457 - 458 - # Sleep until next poll 459 - time.sleep(self.poll_interval) 460 - 461 - # Cleanup 462 - self.shutdown() 463 - 464 - def shutdown(self): 465 - """Clean shutdown - write any pending captures.""" 466 - logger.info("Shutting down tmux observer...") 467 - 468 - if self.captures: 469 - logger.info(f"Writing {len(self.captures)} pending captures") 470 - self.handle_boundary() 471 - 472 - if self.callosum: 473 - self.callosum.stop() 474 - 475 - 476 - def main(): 477 - """CLI entry point.""" 478 - parser = argparse.ArgumentParser( 479 - description="Tmux terminal observer for journaling." 480 - ) 481 - parser.add_argument( 482 - "--interval", 483 - type=int, 484 - default=DEFAULT_INTERVAL, 485 - help=f"Segment duration in seconds (default: {DEFAULT_INTERVAL})", 486 - ) 487 - parser.add_argument( 488 - "--poll", 489 - type=int, 490 - default=DEFAULT_POLL_INTERVAL, 491 - help=f"Poll interval in seconds (default: {DEFAULT_POLL_INTERVAL})", 492 - ) 493 - args = setup_cli(parser) 494 - 495 - # Verify journal path exists 496 - journal = os.getenv("JOURNAL_PATH") 497 - if not journal or not os.path.exists(journal): 498 - logger.error(f"JOURNAL_PATH not set or does not exist: {journal}") 499 - return 1 500 - 501 - # Check tmux is available 502 - if run_tmux_command(["list-sessions"]) is None: 503 - logger.warning("tmux not available or no sessions - will poll for server") 504 - 505 - observer = TmuxObserver(interval=args.interval, poll_interval=args.poll) 506 - 507 - # Signal handlers 508 - def signal_handler(signum, frame): 509 - logger.info("Received shutdown signal") 510 - observer.running = False 511 - 512 - signal.signal(signal.SIGINT, signal_handler) 513 - signal.signal(signal.SIGTERM, signal_handler) 514 - 515 - try: 516 - observer.main_loop() 517 - except Exception as e: 518 - logger.error(f"Observer error: {e}", exc_info=True) 519 - return 1 520 - 521 - return 0 522 - 523 - 524 - if __name__ == "__main__": 525 - exit(main())

Configure Feed

Configure Feed