+2 -2

README.md

reviewed

··· 16 16 17 17 **a system of intelligence, not just a system of record.** 18 18 19 19 - - **automatic transcription** — continuous audio capture with speaker identification. every conversation, transcribed and searchable. 19 19 + - **automatic transcription** — continuous audio capture (via standalone observers) with speaker identification. every conversation, transcribed and searchable. 20 20 - **entity tracking** — people, companies, and projects extracted from your conversations and tracked across time. 21 21 - **knowledge graphs** — relationships between entities mapped automatically. who works with whom, which projects connect to which people. 22 22 - **meeting detection** — meetings identified, summarized, and linked. meeting prep that surfaces what you discussed last time and personal context you'd forget. ··· 59 59 +-------------+ 60 60 ``` 61 61 62 62 - - **observe** — captures audio (PipeWire on Linux, solstone-macos native app on macOS) and screen activity. produces FLAC audio, WebM screen recordings, and timestamped metadata. 62 62 + - **observe** — receives captured audio and screen activity from standalone observers (solstone-linux, solstone-tmux, solstone-macos) via remote ingest. processes FLAC audio, WebM screen recordings, and timestamped metadata. 63 63 - **think** — transcribes audio (faster-whisper), analyzes screen captures, extracts entities, detects meetings, and indexes everything into SQLite. runs 30 configurable agent/generator templates from `muse/`. 64 64 - **cortex** — orchestrates agent execution. receives events, dispatches agents, writes results back to the journal. 65 65 - **callosum** — async message bus connecting all services. enables event-driven coordination between observe, think, cortex, and convey.

+3 -3

docs/CALLOSUM.md

reviewed

··· 70 70 71 71 ### `observe` - Multimodal capture and processing 72 72 **Sources:** 73 73 - - Capture: `observe/observer.py` → `observe/linux/observer.py` (macOS capture comes from remote/native observers) 73 73 + - Capture: standalone observer services (solstone-linux, solstone-tmux, solstone-macos) upload via remote ingest 74 74 - Processing: `observe/sense.py`, `observe/describe.py`, `observe/transcribe/` 75 75 76 76 **Events:** 77 77 | Event | Emitter | Purpose | 78 78 |-------|---------|---------| 79 79 - | `status` | observer, sense | Periodic state (every 5s) - see `emit_status()` in each source | 80 80 - | `observing` | observer | Recording window boundary crossed, files saved | 79 79 + | `status` | sense | Periodic state (every 5s) - see `emit_status()` in source | 80 80 + | `observing` | ingest | Recording window boundary crossed, files saved | 81 81 | `detected` | sense | File detected, handler spawned | 82 82 | `described` | describe | Vision analysis complete | 83 83 | `transcribed` | transcribe | Audio transcription complete (includes VAD metadata) |

+3 -3

docs/INSTALL.md

reviewed

··· 167 167 ``` 168 168 169 169 This starts: 170 170 - - **Observer** - Screen and audio capture 171 170 - **Sense** - File detection and processing dispatch 172 171 - **Callosum** - Message bus for inter-service communication 172 172 + - **Cortex** - Agent execution 173 173 174 174 ### Verify Services 175 175 ··· 227 227 228 228 ```bash 229 229 # Check services are running 230 230 - pgrep -af "sol:observer|sol:sense|sol:supervisor" 230 230 + pgrep -af "sol:sense|sol:supervisor" 231 231 232 232 # Check Callosum socket exists 233 233 ls -la journal/health/callosum.sock ··· 243 243 ## Next Steps 244 244 245 245 - Create your first facet (project/context) in the web interface 246 246 - - Start capturing - the observer runs automatically 246 246 + - Set up a standalone observer for your platform (solstone-linux, solstone-macos) 247 247 - Review captured content in the Calendar and Transcripts apps 248 248 - Chat with the AI about your journal content 249 249

+1 -1

docs/OBSERVE.md

reviewed

··· 8 8 9 9 | Observer | What it captures | Repo | Runs as | 10 10 |----------|-----------------|------|---------| 11 11 + | **solstone-linux** | Screen + audio on Linux | `solstone-linux` | systemd user service / standalone | 11 12 | **solstone-macos** | Screen + audio on macOS | `solstone-macos` | Native menu bar app | 12 13 | **solstone-tmux** | Tmux terminal sessions | `solstone-tmux` | systemd user service / standalone | 13 13 - | **Linux observer** | Screen + audio on Linux | Built-in (`observe/linux/`) | Supervisor-managed process | 14 14 15 15 ### Managing observers 16 16

-78

observe/detect.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - import logging 5 5 - import threading 6 6 - 7 7 - import numpy as np 8 8 - import soundcard as sc 9 9 - 10 10 - logger = logging.getLogger(__name__) 11 11 - 12 12 - 13 13 - def input_detect(duration=0.4, sample_rate=44100): 14 14 - t = np.linspace(0, duration, int(sample_rate * duration), endpoint=False) 15 15 - tone = 0.5 * np.sin(2 * np.pi * 18000 * t) # ultrasonic 16 16 - 17 17 - try: 18 18 - devices = sc.all_microphones(include_loopback=True) 19 19 - except Exception: 20 20 - logger.warning("Failed to enumerate audio devices") 21 21 - return None, None 22 22 - if not devices: 23 23 - logger.warning("No audio devices found") 24 24 - return None, None 25 25 - 26 26 - results = {} 27 27 - barrier = threading.Barrier(len(devices) + 1) 28 28 - 29 29 - def record_mic(mic, results): 30 30 - barrier.wait() 31 31 - try: 32 32 - audio = mic.record( 33 33 - samplerate=sample_rate, numframes=int(sample_rate * duration) 34 34 - ) 35 35 - results[mic.name] = audio 36 36 - except Exception: 37 37 - results[mic.name] = None 38 38 - 39 39 - def play_tone(): 40 40 - barrier.wait() 41 41 - try: 42 42 - sp = sc.default_speaker() 43 43 - sp.play(tone, samplerate=sample_rate) 44 44 - except Exception: 45 45 - logger.warning("No default speaker available for tone detection") 46 46 - 47 47 - threads = [] 48 48 - for mic in devices: 49 49 - thread = threading.Thread(target=record_mic, args=(mic, results)) 50 50 - thread.start() 51 51 - threads.append(thread) 52 52 - 53 53 - play_thread = threading.Thread(target=play_tone) 54 54 - play_thread.start() 55 55 - threads.append(play_thread) 56 56 - 57 57 - for thread in threads: 58 58 - thread.join() 59 59 - 60 60 - # Analyze the recordings with a simple amplitude threshold 61 61 - threshold = 0.001 62 62 - mic_detected = None 63 63 - loopback_detected = None 64 64 - for mic in devices: 65 65 - audio = results.get(mic.name) 66 66 - if audio is not None and np.max(np.abs(audio)) > threshold: 67 67 - # First match for each category 68 68 - if "microphone" in str(mic).lower() and mic_detected is None: 69 69 - mic_detected = mic 70 70 - if "loopback" in str(mic).lower() and loopback_detected is None: 71 71 - loopback_detected = mic 72 72 - return mic_detected, loopback_detected 73 73 - 74 74 - 75 75 - if __name__ == "__main__": 76 76 - mic, loopback = input_detect() 77 77 - print("Microphone detection:", mic.name if mic else "None") 78 78 - print("Loopback detection:", loopback.name if loopback else "None")

-18

observe/gnome/__init__.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - """GNOME-specific activity detection library.""" 5 5 - 6 6 - from observe.gnome.activity import ( 7 7 - get_idle_time_ms, 8 8 - get_monitor_geometries, 9 9 - is_power_save_active, 10 10 - is_screen_locked, 11 11 - ) 12 12 - 13 13 - __all__ = [ 14 14 - "get_idle_time_ms", 15 15 - "get_monitor_geometries", 16 16 - "is_power_save_active", 17 17 - "is_screen_locked", 18 18 - ]

-128

observe/gnome/activity.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - """GNOME-specific activity detection using Mutter and GTK DBus APIs.""" 5 5 - 6 6 - import os 7 7 - 8 8 - import gi 9 9 - from dbus_next.aio import MessageBus 10 10 - 11 11 - gi.require_version("Gdk", "4.0") # noqa: E402 12 12 - gi.require_version("Gtk", "4.0") # noqa: E402 13 13 - from gi.repository import Gdk, Gtk # noqa: E402 14 14 - 15 15 - # DBus service constants 16 16 - IDLE_MONITOR_BUS = "org.gnome.Mutter.IdleMonitor" 17 17 - IDLE_MONITOR_PATH = "/org/gnome/Mutter/IdleMonitor/Core" 18 18 - IDLE_MONITOR_IFACE = "org.gnome.Mutter.IdleMonitor" 19 19 - 20 20 - SCREENSAVER_BUS = "org.gnome.ScreenSaver" 21 21 - SCREENSAVER_PATH = "/org/gnome/ScreenSaver" 22 22 - SCREENSAVER_IFACE = "org.gnome.ScreenSaver" 23 23 - 24 24 - DISPLAY_CONFIG_BUS = "org.gnome.Mutter.DisplayConfig" 25 25 - DISPLAY_CONFIG_PATH = "/org/gnome/Mutter/DisplayConfig" 26 26 - DISPLAY_CONFIG_IFACE = "org.gnome.Mutter.DisplayConfig" 27 27 - 28 28 - 29 29 - async def get_idle_time_ms(bus: MessageBus) -> int: 30 30 - """ 31 31 - Get the current idle time in milliseconds. 32 32 - 33 33 - Args: 34 34 - bus: Connected DBus session bus 35 35 - 36 36 - Returns: 37 37 - Idle time in milliseconds 38 38 - """ 39 39 - introspection = await bus.introspect(IDLE_MONITOR_BUS, IDLE_MONITOR_PATH) 40 40 - proxy_obj = bus.get_proxy_object(IDLE_MONITOR_BUS, IDLE_MONITOR_PATH, introspection) 41 41 - idle_monitor = proxy_obj.get_interface(IDLE_MONITOR_IFACE) 42 42 - idle_time = await idle_monitor.call_get_idletime() 43 43 - return idle_time 44 44 - 45 45 - 46 46 - async def is_screen_locked(bus: MessageBus) -> bool: 47 47 - """ 48 48 - Check if the screen is currently locked using GNOME ScreenSaver. 49 49 - 50 50 - Args: 51 51 - bus: Connected DBus session bus 52 52 - 53 53 - Returns: 54 54 - True if screen is locked, False otherwise 55 55 - """ 56 56 - try: 57 57 - intro = await bus.introspect(SCREENSAVER_BUS, SCREENSAVER_PATH) 58 58 - obj = bus.get_proxy_object(SCREENSAVER_BUS, SCREENSAVER_PATH, intro) 59 59 - iface = obj.get_interface(SCREENSAVER_IFACE) 60 60 - return bool(await iface.call_get_active()) 61 61 - except Exception: 62 62 - # If the interface isn't present on this session, treat as unlocked 63 63 - return False 64 64 - 65 65 - 66 66 - async def is_power_save_active(bus: MessageBus) -> bool: 67 67 - """ 68 68 - Check if display power save mode is active (screen blanked). 69 69 - 70 70 - Args: 71 71 - bus: Connected DBus session bus 72 72 - 73 73 - Returns: 74 74 - True if power save is active, False otherwise 75 75 - """ 76 76 - try: 77 77 - intro = await bus.introspect(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH) 78 78 - obj = bus.get_proxy_object(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH, intro) 79 79 - iface = obj.get_interface("org.freedesktop.DBus.Properties") 80 80 - # Get("org.gnome.Mutter.DisplayConfig", "PowerSaveMode") -> int32 81 81 - # 0 = on, nonzero = blanked 82 82 - mode_variant = await iface.call_get(DISPLAY_CONFIG_IFACE, "PowerSaveMode") 83 83 - mode = int(mode_variant.value) 84 84 - return mode != 0 85 85 - except Exception: 86 86 - # Property or service not available -> assume not blanked 87 87 - return False 88 88 - 89 89 - 90 90 - def get_monitor_geometries() -> list[dict]: 91 91 - """ 92 92 - Get structured monitor information. 93 93 - 94 94 - Returns: 95 95 - List of dicts with format: 96 96 - [{"id": "connector-id", "box": [x1, y1, x2, y2], "position": "center|left|right|..."}, ...] 97 97 - where box contains [left, top, right, bottom] coordinates 98 98 - """ 99 99 - from observe.utils import assign_monitor_positions 100 100 - 101 101 - # Initialize GTK before using GDK functions 102 102 - Gtk.init() 103 103 - 104 104 - # Get the default display. If it is None, try opening one from the environment. 105 105 - display = Gdk.Display.get_default() 106 106 - if display is None: 107 107 - env_display = os.environ.get("WAYLAND_DISPLAY") or os.environ.get("DISPLAY") 108 108 - if env_display is not None: 109 109 - display = Gdk.Display.open(env_display) 110 110 - if display is None: 111 111 - raise RuntimeError("No display available") 112 112 - # In GTK 4, get_monitors() returns a list of Gdk.Monitor objects. 113 113 - monitors = display.get_monitors() 114 114 - 115 115 - # Collect monitor geometries 116 116 - geometries = [] 117 117 - for monitor in monitors: 118 118 - geom = monitor.get_geometry() 119 119 - connector = monitor.get_connector() or f"monitor-{len(geometries)}" 120 120 - geometries.append( 121 121 - { 122 122 - "id": connector, 123 123 - "box": [geom.x, geom.y, geom.x + geom.width, geom.y + geom.height], 124 124 - } 125 125 - ) 126 126 - 127 127 - # Assign position labels using shared algorithm 128 128 - return assign_monitor_positions(geometries)

+1 -189

observe/hear.py

reviewed

··· 1 1 # SPDX-License-Identifier: AGPL-3.0-only 2 2 # Copyright (c) 2026 sol pbc 3 3 4 4 - """Audio recording utilities for observe package.""" 4 4 + """Audio transcript utilities for observe package.""" 5 5 6 6 from __future__ import annotations 7 7 8 8 - import gc 9 9 - import io 10 8 import json 11 9 import logging 12 10 import os 13 11 import re 14 14 - import signal 15 15 - import threading 16 16 - import time 17 12 from datetime import datetime 18 13 from pathlib import Path 19 19 - from queue import Queue 20 14 from typing import Any 21 21 - 22 22 - import numpy as np 23 23 - import soundfile as sf 24 24 - 25 25 - from observe.utils import SAMPLE_RATE 26 26 - 27 27 - # Constants 28 28 - BLOCK_SIZE = 1024 29 29 - 30 30 - 31 31 - class AudioRecorder: 32 32 - """Records stereo audio from microphone and system audio.""" 33 33 - 34 34 - def __init__(self): 35 35 - # Queue now holds stereo chunks (mic=left, sys=right) 36 36 - self.audio_queue = Queue() 37 37 - self._running = True 38 38 - self.recording_thread = None 39 39 - 40 40 - def detect(self): 41 41 - """Detect microphone and system audio devices.""" 42 42 - from observe.detect import input_detect 43 43 - 44 44 - mic, loopback = input_detect() 45 45 - if mic is None or loopback is None: 46 46 - logging.error(f"Detection failed: mic {mic} sys {loopback}") 47 47 - return False 48 48 - logging.info(f"Detected microphone: {mic.name}") 49 49 - logging.info(f"Detected system audio: {loopback.name}") 50 50 - self.mic_device = mic 51 51 - self.sys_device = loopback 52 52 - return True 53 53 - 54 54 - def record_both(self): 55 55 - """Record from both mic and system audio in a loop.""" 56 56 - while self._running: 57 57 - try: 58 58 - with ( 59 59 - self.mic_device.recorder( 60 60 - samplerate=SAMPLE_RATE, channels=[-1], blocksize=BLOCK_SIZE 61 61 - ) as mic_rec, 62 62 - self.sys_device.recorder( 63 63 - samplerate=SAMPLE_RATE, channels=[-1], blocksize=BLOCK_SIZE 64 64 - ) as sys_rec, 65 65 - ): 66 66 - block_count = 0 67 67 - while self._running and block_count < 1000: 68 68 - try: 69 69 - mic_chunk = mic_rec.record(numframes=BLOCK_SIZE) 70 70 - sys_chunk = sys_rec.record(numframes=BLOCK_SIZE) 71 71 - 72 72 - # Basic validation 73 73 - if mic_chunk is None or mic_chunk.size == 0: 74 74 - logging.warning("Empty microphone buffer") 75 75 - continue 76 76 - if sys_chunk is None or sys_chunk.size == 0: 77 77 - logging.warning("Empty system buffer") 78 78 - continue 79 79 - 80 80 - try: 81 81 - # Try to create stereo chunk - this is where shape errors occur 82 82 - stereo_chunk = np.column_stack((mic_chunk, sys_chunk)) 83 83 - self.audio_queue.put(stereo_chunk) 84 84 - block_count += 1 85 85 - except (TypeError, ValueError, AttributeError) as e: 86 86 - # Audio device returned unexpected format - trigger clean shutdown 87 87 - error_msg = f"Fatal audio format error: {e}" 88 88 - logging.error( 89 89 - f"{error_msg} - triggering clean shutdown\n" 90 90 - f" mic_chunk type={type(mic_chunk)}, " 91 91 - f"shape={getattr(mic_chunk, 'shape', 'N/A')}, " 92 92 - f"dtype={getattr(mic_chunk, 'dtype', 'N/A')}\n" 93 93 - f" sys_chunk type={type(sys_chunk)}, " 94 94 - f"shape={getattr(sys_chunk, 'shape', 'N/A')}, " 95 95 - f"dtype={getattr(sys_chunk, 'dtype', 'N/A')}" 96 96 - ) 97 97 - 98 98 - # Notify user via Callosum 99 99 - from think.callosum import callosum_send 100 100 - 101 101 - callosum_send( 102 102 - "notification", 103 103 - "show", 104 104 - message=error_msg, 105 105 - title="Audio Capture Error", 106 106 - icon="🔇", 107 107 - app="hear", 108 108 - ) 109 109 - 110 110 - # Stop recording thread 111 111 - self._running = False 112 112 - # Send SIGTERM to trigger graceful shutdown (same as Ctrl-C) 113 113 - os.kill(os.getpid(), signal.SIGTERM) 114 114 - return 115 115 - except Exception as e: 116 116 - logging.error(f"Error recording audio: {e}") 117 117 - if not self._running: 118 118 - break 119 119 - time.sleep(0.5) 120 120 - del ( 121 121 - mic_rec, 122 122 - sys_rec, 123 123 - ) # Explicitly delete to reset system audio device resources 124 124 - gc.collect() # Force garbage collection after deleting recorders 125 125 - except Exception as e: 126 126 - logging.error(f"Error setting up recorders: {e}") 127 127 - if self._running: 128 128 - time.sleep(1) # Wait before retrying 129 129 - 130 130 - def get_buffers(self) -> np.ndarray: 131 131 - """Return concatenated stereo audio data from the queue.""" 132 132 - stereo_buffer = np.array([], dtype=np.float32).reshape(0, 2) 133 133 - 134 134 - while not self.audio_queue.empty(): 135 135 - stereo_chunk = self.audio_queue.get() 136 136 - 137 137 - if stereo_chunk is None or stereo_chunk.size == 0: 138 138 - logging.warning("Queue contained empty chunk") 139 139 - continue 140 140 - 141 141 - # Clean the data 142 142 - stereo_chunk = np.nan_to_num( 143 143 - stereo_chunk, nan=0.0, posinf=1e10, neginf=-1e10 144 144 - ) 145 145 - stereo_buffer = np.vstack((stereo_buffer, stereo_chunk)) 146 146 - 147 147 - if stereo_buffer.size == 0: 148 148 - logging.warning("No valid audio data retrieved from queue") 149 149 - 150 150 - return stereo_buffer 151 151 - 152 152 - def create_flac_bytes(self, stereo_data: np.ndarray) -> bytes: 153 153 - """Create FLAC bytes from stereo audio data.""" 154 154 - if stereo_data is None or stereo_data.size == 0: 155 155 - logging.warning("Audio data is empty. Returning empty bytes.") 156 156 - return b"" 157 157 - 158 158 - # Convert to int16 159 159 - audio_data = (np.clip(stereo_data, -1.0, 1.0) * 32767).astype(np.int16) 160 160 - 161 161 - buf = io.BytesIO() 162 162 - try: 163 163 - sf.write(buf, audio_data, SAMPLE_RATE, format="FLAC") 164 164 - except Exception as e: 165 165 - logging.error( 166 166 - f"Error creating FLAC: {e}. Audio data shape: {audio_data.shape}, dtype: {audio_data.dtype}" 167 167 - ) 168 168 - return b"" 169 169 - 170 170 - return buf.getvalue() 171 171 - 172 172 - def create_mono_flac_bytes(self, mono_data: np.ndarray) -> bytes: 173 173 - """Create FLAC bytes from mono audio data.""" 174 174 - if mono_data is None or mono_data.size == 0: 175 175 - logging.warning("Mono audio data is empty. Returning empty bytes.") 176 176 - return b"" 177 177 - 178 178 - # Convert to int16 179 179 - audio_data = (np.clip(mono_data, -1.0, 1.0) * 32767).astype(np.int16) 180 180 - 181 181 - buf = io.BytesIO() 182 182 - try: 183 183 - sf.write(buf, audio_data, SAMPLE_RATE, format="FLAC") 184 184 - except Exception as e: 185 185 - logging.error( 186 186 - f"Error creating mono FLAC: {e}. Audio shape: {audio_data.shape}" 187 187 - ) 188 188 - return b"" 189 189 - 190 190 - return buf.getvalue() 191 191 - 192 192 - def start_recording(self): 193 193 - """Start the recording thread.""" 194 194 - self._running = True 195 195 - self.recording_thread = threading.Thread(target=self.record_both, daemon=True) 196 196 - self.recording_thread.start() 197 197 - 198 198 - def stop_recording(self): 199 199 - """Stop the recording thread.""" 200 200 - self._running = False 201 201 - if self.recording_thread: 202 202 - self.recording_thread.join(timeout=2.0) 203 15 204 16 205 17 def load_transcript(

-4

observe/linux/__init__.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - """Linux desktop observation using XDG Portal and PipeWire."""

-44

observe/linux/audio.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - """Linux audio utilities using PulseAudio/PipeWire.""" 5 5 - 6 6 - import asyncio 7 7 - import logging 8 8 - 9 9 - logger = logging.getLogger(__name__) 10 10 - 11 11 - 12 12 - async def is_sink_muted() -> bool: 13 13 - """ 14 14 - Check if the default audio sink is muted using PulseAudio. 15 15 - 16 16 - Uses `pactl get-sink-mute @DEFAULT_SINK@` to query mute status. 17 17 - 18 18 - Returns: 19 19 - True if muted, False otherwise (including on error). 20 20 - """ 21 21 - try: 22 22 - proc = await asyncio.create_subprocess_exec( 23 23 - "pactl", 24 24 - "get-sink-mute", 25 25 - "@DEFAULT_SINK@", 26 26 - stdout=asyncio.subprocess.PIPE, 27 27 - stderr=asyncio.subprocess.PIPE, 28 28 - ) 29 29 - stdout, stderr = await proc.communicate() 30 30 - 31 31 - if proc.returncode != 0: 32 32 - stderr_text = stderr.decode().strip() if stderr else "" 33 33 - logger.warning(f"pactl failed (rc={proc.returncode}): {stderr_text}") 34 34 - return False 35 35 - 36 36 - output = stdout.decode().strip() 37 37 - return "Mute: yes" in output 38 38 - 39 39 - except FileNotFoundError: 40 40 - logger.warning("pactl not found, assuming unmuted") 41 41 - return False 42 42 - except Exception as e: 43 43 - logger.warning(f"Error checking sink mute status: {e}") 44 44 - return False

-735

observe/linux/observer.py

reviewed

··· 1 1 - #!/usr/bin/env python3 2 2 - # SPDX-License-Identifier: AGPL-3.0-only 3 3 - # Copyright (c) 2026 sol pbc 4 4 - 5 5 - """ 6 6 - Unified observer for audio and screencast capture. 7 7 - 8 8 - Continuously captures audio and manages screencast recording based on activity. 9 9 - Creates 5-minute windows, saving audio if voice activity detected and recording 10 10 - screencasts during active segments. 11 11 - 12 12 - State machine: 13 13 - SCREENCAST: Screen is active, recording video 14 14 - IDLE: Screen is inactive 15 15 - """ 16 16 - 17 17 - import argparse 18 18 - import asyncio 19 19 - import logging 20 20 - import os 21 21 - import platform 22 22 - import shutil 23 23 - import signal 24 24 - import socket 25 25 - import subprocess 26 26 - import sys 27 27 - import time 28 28 - from pathlib import Path 29 29 - 30 30 - import numpy as np 31 31 - from dbus_next.aio import MessageBus 32 32 - from dbus_next.constants import BusType 33 33 - 34 34 - from observe.gnome.activity import ( 35 35 - get_idle_time_ms, 36 36 - is_power_save_active, 37 37 - is_screen_locked, 38 38 - ) 39 39 - from observe.hear import AudioRecorder 40 40 - from observe.linux.audio import is_sink_muted 41 41 - from observe.linux.screencast import Screencaster, StreamInfo 42 42 - from observe.remote_client import ObserverClient, cleanup_draft, finalize_draft 43 43 - from observe.utils import create_draft_folder, get_timestamp_parts 44 44 - from think.streams import stream_name 45 45 - from think.utils import setup_cli 46 46 - 47 47 - logger = logging.getLogger(__name__) 48 48 - 49 49 - # Host identification 50 50 - HOST = socket.gethostname() 51 51 - PLATFORM = platform.system().lower() 52 52 - 53 53 - # Constants 54 54 - IDLE_THRESHOLD_MS = 5 * 60 * 1000 # 5 minutes 55 55 - RMS_THRESHOLD = 0.01 56 56 - MIN_HITS_FOR_SAVE = 3 57 57 - CHUNK_DURATION = 5 # seconds 58 58 - # Exit codes 59 59 - EXIT_TEMPFAIL = 75 # EX_TEMPFAIL: session not ready, retry later 60 60 - 61 61 - # Capture modes 62 62 - MODE_IDLE = "idle" 63 63 - MODE_SCREENCAST = "screencast" 64 64 - 65 65 - # Audio detection retry 66 66 - DETECT_RETRIES = 3 67 67 - DETECT_RETRY_DELAY = 5 # seconds 68 68 - 69 69 - 70 70 - class Observer: 71 71 - """Unified audio and screencast observer.""" 72 72 - 73 73 - def __init__(self, interval: int = 300): 74 74 - self.interval = interval 75 75 - self.audio_recorder = AudioRecorder() 76 76 - self.screencaster = Screencaster() 77 77 - self.bus: MessageBus | None = None 78 78 - self.running = True 79 79 - self.stream = stream_name(host=HOST) 80 80 - 81 81 - self._client: ObserverClient | None = None 82 82 - 83 83 - # State tracking 84 84 - self.start_at = time.time() # Wall-clock for filenames 85 85 - self.start_at_mono = time.monotonic() # Monotonic for elapsed calculations 86 86 - self.threshold_hits = 0 87 87 - self.accumulated_audio_buffer = np.array([], dtype=np.float32).reshape(0, 2) 88 88 - 89 89 - # Mode tracking (replaces screencast_running boolean) 90 90 - self.current_mode = MODE_IDLE 91 91 - 92 92 - # Draft folder for current segment (HHMMSS_draft/) 93 93 - self.draft_dir: str | None = None 94 94 - 95 95 - # Multi-file screencast tracking 96 96 - self.current_streams: list[StreamInfo] = [] 97 97 - 98 98 - # Activity status cache (updated each loop) 99 99 - self.cached_is_active = False 100 100 - self.cached_idle_time_ms = 0 101 101 - self.cached_screen_locked = False 102 102 - self.cached_is_muted = False 103 103 - self.cached_power_save = False 104 104 - 105 105 - # Mute state at segment start (determines save format) 106 106 - self.segment_is_muted = False 107 107 - 108 108 - async def setup(self): 109 109 - """Initialize audio devices and DBus connection.""" 110 110 - # Detect audio devices with retry (devices may still be initializing) 111 111 - detected = False 112 112 - for attempt in range(DETECT_RETRIES): 113 113 - if self.audio_recorder.detect(): 114 114 - detected = True 115 115 - break 116 116 - if attempt < DETECT_RETRIES - 1: 117 117 - logger.info( 118 118 - "Audio detection attempt %d/%d failed, retrying in %ds", 119 119 - attempt + 1, 120 120 - DETECT_RETRIES, 121 121 - DETECT_RETRY_DELAY, 122 122 - ) 123 123 - await asyncio.sleep(DETECT_RETRY_DELAY) 124 124 - if not detected: 125 125 - logger.error("Failed to detect audio devices") 126 126 - return False 127 127 - 128 128 - self.audio_recorder.start_recording() 129 129 - logger.info("Audio recording started") 130 130 - 131 131 - # Connect to DBus for idle/lock detection 132 132 - self.bus = await MessageBus(bus_type=BusType.SESSION).connect() 133 133 - logger.info("DBus connection established") 134 134 - 135 135 - # Verify portal is available (exit if not) 136 136 - if not await self.screencaster.connect(): 137 137 - logger.error("Screencast portal not available") 138 138 - return False 139 139 - logger.info("Screencast portal connected") 140 140 - 141 141 - self._client = ObserverClient(self.stream) 142 142 - logger.info("Remote client initialized") 143 143 - 144 144 - return True 145 145 - 146 146 - async def check_activity_status(self) -> str: 147 147 - """ 148 148 - Check system activity status and determine capture mode. 149 149 - 150 150 - Returns: 151 151 - Capture mode: MODE_SCREENCAST or MODE_IDLE 152 152 - """ 153 153 - idle_time = await get_idle_time_ms(self.bus) 154 154 - screen_locked = await is_screen_locked(self.bus) 155 155 - power_save = await is_power_save_active(self.bus) 156 156 - sink_muted = await is_sink_muted() 157 157 - 158 158 - # Cache values for status events 159 159 - self.cached_idle_time_ms = idle_time 160 160 - self.cached_screen_locked = screen_locked 161 161 - self.cached_is_muted = sink_muted 162 162 - self.cached_power_save = power_save 163 163 - 164 164 - # Determine screen activity 165 165 - screen_idle = (idle_time > IDLE_THRESHOLD_MS) or screen_locked or power_save 166 166 - screen_active = not screen_idle 167 167 - 168 168 - # Determine mode from screen activity 169 169 - if screen_active: 170 170 - mode = MODE_SCREENCAST 171 171 - else: 172 172 - mode = MODE_IDLE 173 173 - 174 174 - # Cache legacy is_active for audio threshold logic 175 175 - has_audio_activity = self.threshold_hits >= MIN_HITS_FOR_SAVE 176 176 - self.cached_is_active = screen_active or has_audio_activity 177 177 - 178 178 - return mode 179 179 - 180 180 - def compute_rms(self, audio_buffer: np.ndarray) -> float: 181 181 - """Compute per-channel RMS and return maximum (stereo: mic=left, sys=right).""" 182 182 - if audio_buffer.size == 0: 183 183 - return 0.0 184 184 - # Compute RMS for each channel separately 185 185 - rms_left = float(np.sqrt(np.mean(audio_buffer[:, 0] ** 2))) 186 186 - rms_right = float(np.sqrt(np.mean(audio_buffer[:, 1] ** 2))) 187 187 - return max(rms_left, rms_right) 188 188 - 189 189 - def _save_audio_segment(self, segment_dir: str, is_muted: bool) -> list[str]: 190 190 - """ 191 191 - Save accumulated audio buffer to segment directory. 192 192 - 193 193 - Args: 194 194 - segment_dir: Path to the segment directory (YYYYMMDD/HHMMSS_LEN/) 195 195 - is_muted: Whether to save as split mono files (muted) or stereo (unmuted) 196 196 - 197 197 - Returns: 198 198 - List of saved filenames (empty if nothing saved) 199 199 - """ 200 200 - if self.accumulated_audio_buffer.size == 0: 201 201 - logger.warning("No audio buffer to save") 202 202 - return [] 203 203 - 204 204 - segment_path = Path(segment_dir) 205 205 - 206 206 - if is_muted: 207 207 - # Split mode: save mic and sys as separate mono files 208 208 - mic_data = self.accumulated_audio_buffer[:, 0] 209 209 - sys_data = self.accumulated_audio_buffer[:, 1] 210 210 - 211 211 - mic_bytes = self.audio_recorder.create_mono_flac_bytes(mic_data) 212 212 - sys_bytes = self.audio_recorder.create_mono_flac_bytes(sys_data) 213 213 - 214 214 - mic_name = "mic_audio.flac" 215 215 - sys_name = "sys_audio.flac" 216 216 - 217 217 - mic_path = segment_path / mic_name 218 218 - sys_path = segment_path / sys_name 219 219 - 220 220 - with open(mic_path, "wb") as f: 221 221 - f.write(mic_bytes) 222 222 - with open(sys_path, "wb") as f: 223 223 - f.write(sys_bytes) 224 224 - 225 225 - logger.info(f"Saved split audio (muted): {mic_path}, {sys_path}") 226 226 - return [mic_name, sys_name] 227 227 - else: 228 228 - # Normal mode: save combined stereo file 229 229 - flac_bytes = self.audio_recorder.create_flac_bytes( 230 230 - self.accumulated_audio_buffer 231 231 - ) 232 232 - audio_name = "audio.flac" 233 233 - flac_path = segment_path / audio_name 234 234 - 235 235 - with open(flac_path, "wb") as f: 236 236 - f.write(flac_bytes) 237 237 - 238 238 - logger.info(f"Saved audio to {flac_path}") 239 239 - return [audio_name] 240 240 - 241 241 - async def handle_boundary(self, new_mode: str): 242 242 - """ 243 243 - Handle window boundary rollover. 244 244 - 245 245 - Closes the current draft folder, uploads segment files, and starts 246 246 - the next segment. 247 247 - 248 248 - Args: 249 249 - new_mode: The mode for the new segment 250 250 - """ 251 251 - # Get timestamp parts for this window and calculate duration 252 252 - date_part, time_part = get_timestamp_parts(self.start_at) 253 253 - duration = int(time.time() - self.start_at) 254 254 - 255 255 - # Stop screencast first (closes file handles) 256 256 - stopped_streams: list[StreamInfo] = [] 257 257 - screen_files: list[str] = [] 258 258 - 259 259 - if self.current_mode == MODE_SCREENCAST: 260 260 - logger.info("Stopping previous screencast") 261 261 - stopped_streams = await self.screencaster.stop() 262 262 - self.current_streams = [] 263 263 - 264 264 - # Collect screen filenames (files are already in draft dir with final names) 265 265 - screen_files = [stream.filename for stream in stopped_streams] 266 266 - 267 267 - # Save audio if we have enough threshold hits (to draft dir) 268 268 - did_save_audio = self.threshold_hits >= MIN_HITS_FOR_SAVE 269 269 - audio_files: list[str] = [] 270 270 - if did_save_audio and self.draft_dir: 271 271 - audio_files = self._save_audio_segment( 272 272 - self.draft_dir, self.segment_is_muted 273 273 - ) 274 274 - if audio_files: 275 275 - logger.info( 276 276 - f"Saved {len(audio_files)} audio file(s) ({self.threshold_hits} hits)" 277 277 - ) 278 278 - else: 279 279 - logger.debug( 280 280 - f"Skipping audio save (only {self.threshold_hits}/{MIN_HITS_FOR_SAVE} hits)" 281 281 - ) 282 282 - 283 283 - # Reset audio state 284 284 - self.accumulated_audio_buffer = np.array([], dtype=np.float32).reshape(0, 2) 285 285 - self.threshold_hits = 0 286 286 - 287 287 - # Collect all files saved in this segment 288 288 - files = audio_files + screen_files 289 289 - segment_key = f"{time_part}_{duration}" 290 290 - 291 291 - # Upload segment files from draft directory 292 292 - if self.draft_dir and files: 293 293 - draft_path = Path(self.draft_dir) 294 294 - draft_files = [ 295 295 - draft_path / f 296 296 - for f in os.listdir(self.draft_dir) 297 297 - if (draft_path / f).is_file() 298 298 - ] 299 299 - uploaded = False 300 300 - if draft_files and self._client and self.running: 301 301 - meta = {"host": HOST, "platform": PLATFORM, "stream": self.stream} 302 302 - result = await asyncio.to_thread( 303 303 - self._client.upload_segment, 304 304 - date_part, 305 305 - segment_key, 306 306 - draft_files, 307 307 - meta, 308 308 - ) 309 309 - if result.success: 310 310 - logger.info( 311 311 - f"Segment uploaded: {segment_key} ({len(draft_files)} files)" 312 312 - ) 313 313 - uploaded = True 314 314 - else: 315 315 - logger.error(f"Segment upload failed: {segment_key}") 316 316 - if uploaded: 317 317 - cleanup_draft(self.draft_dir) 318 318 - else: 319 319 - finalize_draft(self.draft_dir, segment_key) 320 320 - elif self.draft_dir and not files: 321 321 - cleanup_draft(self.draft_dir) 322 322 - 323 323 - self.draft_dir = None 324 324 - 325 325 - # Reset timing for new window 326 326 - self.start_at = time.time() # Wall-clock for filenames 327 327 - self.start_at_mono = time.monotonic() # Monotonic for elapsed 328 328 - 329 329 - # Update segment mute state for new segment 330 330 - self.segment_is_muted = self.cached_is_muted 331 331 - 332 332 - # Update mode 333 333 - old_mode = self.current_mode 334 334 - self.current_mode = new_mode 335 335 - 336 336 - # Start new capture based on mode (creates new draft folder) 337 337 - if new_mode == MODE_SCREENCAST and not self.cached_screen_locked: 338 338 - await self.initialize_screencast() 339 339 - elif new_mode == MODE_IDLE: 340 340 - self._create_draft_folder() 341 341 - 342 342 - logger.info(f"Mode transition: {old_mode} -> {new_mode}") 343 343 - 344 344 - def _create_draft_folder(self) -> str: 345 345 - """Create a draft folder for the current segment.""" 346 346 - self.draft_dir = create_draft_folder(self.start_at, self.stream) 347 347 - logger.debug(f"Created draft folder: {self.draft_dir}") 348 348 - return self.draft_dir 349 349 - 350 350 - async def initialize_screencast(self) -> bool: 351 351 - """ 352 352 - Start a new screencast recording. 353 353 - 354 354 - Creates a draft folder and starts GStreamer recording to it. 355 355 - 356 356 - Returns: 357 357 - True if screencast started successfully, False otherwise. 358 358 - 359 359 - Raises: 360 360 - RuntimeError: If recording fails to start (caller should exit). 361 361 - """ 362 362 - # Create draft folder for this segment 363 363 - draft_path = self._create_draft_folder() 364 364 - 365 365 - try: 366 366 - streams = await self.screencaster.start( 367 367 - draft_path, framerate=1, draw_cursor=True 368 368 - ) 369 369 - except RuntimeError as e: 370 370 - logger.error(f"Failed to start screencast: {e}") 371 371 - raise 372 372 - 373 373 - if not streams: 374 374 - logger.error("No streams returned from screencast start") 375 375 - raise RuntimeError("No streams available") 376 376 - 377 377 - self.current_streams = streams 378 378 - 379 379 - logger.info(f"Started screencast with {len(streams)} stream(s)") 380 380 - for stream in streams: 381 381 - logger.info(f" {stream.position} ({stream.connector}): {stream.file_path}") 382 382 - 383 383 - return True 384 384 - 385 385 - def emit_status(self): 386 386 - """Emit observe.status event with current state.""" 387 387 - if not self._client: 388 388 - return 389 389 - 390 390 - elapsed = int(time.monotonic() - self.start_at_mono) 391 391 - 392 392 - # Calculate screencast info 393 393 - if self.current_mode == MODE_SCREENCAST and self.current_streams: 394 394 - streams_info = [] 395 395 - for stream in self.current_streams: 396 396 - streams_info.append( 397 397 - { 398 398 - "position": stream.position, 399 399 - "connector": stream.connector, 400 400 - "file": stream.file_path, 401 401 - } 402 402 - ) 403 403 - 404 404 - screencast_info = { 405 405 - "recording": True, 406 406 - "streams": streams_info, 407 407 - "window_elapsed_seconds": elapsed, 408 408 - } 409 409 - else: 410 410 - screencast_info = {"recording": False} 411 411 - 412 412 - # Audio info 413 413 - audio_info = { 414 414 - "threshold_hits": self.threshold_hits, 415 415 - "will_save": self.threshold_hits >= MIN_HITS_FOR_SAVE, 416 416 - } 417 417 - 418 418 - # Activity info 419 419 - activity_info = { 420 420 - "active": self.cached_is_active, 421 421 - "idle_time_ms": self.cached_idle_time_ms, 422 422 - "screen_locked": self.cached_screen_locked, 423 423 - "sink_muted": self.cached_is_muted, 424 424 - "power_save": self.cached_power_save, 425 425 - } 426 426 - 427 427 - # Determine reported mode (segment type, not instantaneous state) 428 428 - if self.current_mode == MODE_SCREENCAST: 429 429 - reported_mode = MODE_SCREENCAST 430 430 - else: 431 431 - reported_mode = MODE_IDLE 432 432 - 433 433 - self._client.relay_event( 434 434 - "observe", 435 435 - "status", 436 436 - mode=reported_mode, 437 437 - screencast=screencast_info, 438 438 - audio=audio_info, 439 439 - activity=activity_info, 440 440 - host=HOST, 441 441 - platform=PLATFORM, 442 442 - stream=self.stream, 443 443 - ) 444 444 - 445 445 - async def main_loop(self): 446 446 - """Run the main observer loop.""" 447 447 - logger.info(f"Starting observer loop (interval={self.interval}s)") 448 448 - 449 449 - # Determine initial mode 450 450 - new_mode = await self.check_activity_status() 451 451 - self.segment_is_muted = self.cached_is_muted # Sync initial mute state 452 452 - self.current_mode = new_mode 453 453 - 454 454 - # Start initial capture based on mode (creates draft folder) 455 455 - if new_mode == MODE_SCREENCAST and not self.cached_screen_locked: 456 456 - try: 457 457 - await self.initialize_screencast() 458 458 - except RuntimeError: 459 459 - # Failed to start screencast, exit 460 460 - self.running = False 461 461 - return 462 462 - else: 463 463 - # Create draft folder for audio even without screencast 464 464 - self._create_draft_folder() 465 465 - 466 466 - logger.info(f"Initial mode: {self.current_mode}") 467 467 - 468 468 - while self.running: 469 469 - # Sleep for chunk duration 470 470 - await asyncio.sleep(CHUNK_DURATION) 471 471 - 472 472 - # Check activity status and determine new mode 473 473 - new_mode = await self.check_activity_status() 474 474 - 475 475 - # Check for GStreamer failure mid-recording 476 476 - if ( 477 477 - self.current_mode == MODE_SCREENCAST 478 478 - and not self.screencaster.is_healthy() 479 479 - ): 480 480 - logger.warning("Screencast recording failed, stopping gracefully") 481 481 - await self.screencaster.stop() 482 482 - 483 483 - # Files are already in draft folder, will be finalized at next boundary 484 484 - self.current_streams = [] 485 485 - # Force recalculate mode without screencast 486 486 - self.current_mode = MODE_IDLE 487 487 - 488 488 - # Detect mode change 489 489 - mode_changed = new_mode != self.current_mode 490 490 - if mode_changed: 491 491 - logger.info(f"Mode changing: {self.current_mode} -> {new_mode}") 492 492 - 493 493 - # Only trigger segment boundary on screencast transitions 494 494 - screencast_transition = mode_changed and ( 495 495 - self.current_mode == MODE_SCREENCAST or new_mode == MODE_SCREENCAST 496 496 - ) 497 497 - 498 498 - # Detect mute state transition 499 499 - mute_transition = self.cached_is_muted != self.segment_is_muted 500 500 - if mute_transition: 501 501 - logger.info( 502 502 - f"Mute state changed: " 503 503 - f"{'muted' if self.segment_is_muted else 'unmuted'} -> " 504 504 - f"{'muted' if self.cached_is_muted else 'unmuted'}" 505 505 - ) 506 506 - 507 507 - # Capture audio buffer for this chunk 508 508 - audio_chunk = self.audio_recorder.get_buffers() 509 509 - 510 510 - if audio_chunk.size > 0: 511 511 - # Append to accumulated buffer 512 512 - self.accumulated_audio_buffer = np.vstack( 513 513 - (self.accumulated_audio_buffer, audio_chunk) 514 514 - ) 515 515 - 516 516 - # Compute RMS and check threshold 517 517 - rms = self.compute_rms(audio_chunk) 518 518 - if rms > RMS_THRESHOLD: 519 519 - self.threshold_hits += 1 520 520 - logger.debug( 521 521 - f"RMS {rms:.4f} > threshold (hit {self.threshold_hits})" 522 522 - ) 523 523 - else: 524 524 - logger.debug(f"RMS {rms:.4f} below threshold") 525 525 - else: 526 526 - logger.debug("No audio data in chunk") 527 527 - 528 528 - # Check for window boundary (use monotonic to avoid DST/clock jumps) 529 529 - now_mono = time.monotonic() 530 530 - elapsed = now_mono - self.start_at_mono 531 531 - is_boundary = ( 532 532 - (elapsed >= self.interval) or screencast_transition or mute_transition 533 533 - ) 534 534 - 535 535 - if is_boundary: 536 536 - logger.info( 537 537 - f"Boundary: elapsed={elapsed:.1f}s screencast_change={screencast_transition} " 538 538 - f"mute_change={mute_transition} " 539 539 - f"hits={self.threshold_hits}/{MIN_HITS_FOR_SAVE}" 540 540 - ) 541 541 - await self.handle_boundary(new_mode) 542 542 - 543 543 - # Emit status event 544 544 - self.emit_status() 545 545 - 546 546 - # Cleanup on exit 547 547 - logger.info("Observer loop stopped, cleaning up...") 548 548 - await self.shutdown() 549 549 - 550 550 - async def shutdown(self): 551 551 - """Clean shutdown of observer.""" 552 552 - # Get timestamp parts for final save 553 553 - date_part, time_part = get_timestamp_parts(self.start_at) 554 554 - duration = int(time.time() - self.start_at) 555 555 - 556 556 - # Stop screencast first (closes file handles) 557 557 - stopped_streams: list[StreamInfo] = [] 558 558 - if self.current_mode == MODE_SCREENCAST: 559 559 - logger.info("Stopping screencast for shutdown") 560 560 - stopped_streams = await self.screencaster.stop() 561 561 - # Brief delay for files to be flushed 562 562 - await asyncio.sleep(0.5) 563 563 - 564 564 - # Save final audio if threshold met (to draft dir) 565 565 - audio_files: list[str] = [] 566 566 - if self.threshold_hits >= MIN_HITS_FOR_SAVE and self.draft_dir: 567 567 - audio_files = self._save_audio_segment( 568 568 - self.draft_dir, self.segment_is_muted 569 569 - ) 570 570 - if audio_files: 571 571 - logger.info(f"Saved final audio: {len(audio_files)} file(s)") 572 572 - 573 573 - # Collect all files and finalize segment 574 574 - screen_files = [stream.filename for stream in stopped_streams] 575 575 - files = audio_files + screen_files 576 576 - segment_key = f"{time_part}_{duration}" 577 577 - 578 578 - if self.draft_dir and files: 579 579 - finalize_draft(self.draft_dir, segment_key) 580 580 - logger.info( 581 581 - f"Finalized segment locally: {segment_key} (shutdown, skipping upload)" 582 582 - ) 583 583 - elif self.draft_dir: 584 584 - cleanup_draft(self.draft_dir) 585 585 - 586 586 - self.draft_dir = None 587 587 - 588 588 - # Stop audio recorder 589 589 - self.audio_recorder.stop_recording() 590 590 - logger.info("Audio recording stopped") 591 591 - 592 592 - if self._client: 593 593 - self._client.stop() 594 594 - self._client = None 595 595 - logger.info("Remote client stopped") 596 596 - 597 597 - 598 598 - def _recover_session_env() -> None: 599 599 - """Try to recover desktop session env vars from the systemd user manager. 600 600 - 601 601 - On GNOME Wayland, gnome-shell pushes DISPLAY, WAYLAND_DISPLAY, and 602 602 - DBUS_SESSION_BUS_ADDRESS into the systemd user environment on startup. 603 603 - When the observer is launched from a non-desktop shell, these vars may be missing 604 604 - from the inherited environment — but systemctl --user show-environment 605 605 - has them. 606 606 - """ 607 607 - needed = {"DISPLAY", "WAYLAND_DISPLAY", "DBUS_SESSION_BUS_ADDRESS"} 608 608 - missing = {v for v in needed if not os.environ.get(v)} 609 609 - if not missing: 610 610 - return 611 611 - 612 612 - # Ensure XDG_RUNTIME_DIR is set (required for systemctl --user to connect) 613 613 - if not os.environ.get("XDG_RUNTIME_DIR"): 614 614 - os.environ["XDG_RUNTIME_DIR"] = f"/run/user/{os.getuid()}" 615 615 - 616 616 - try: 617 617 - result = subprocess.run( 618 618 - ["systemctl", "--user", "show-environment"], 619 619 - capture_output=True, 620 620 - text=True, 621 621 - timeout=5, 622 622 - ) 623 623 - if result.returncode != 0: 624 624 - return 625 625 - except (FileNotFoundError, subprocess.TimeoutExpired): 626 626 - return 627 627 - 628 628 - recovered = [] 629 629 - for line in result.stdout.splitlines(): 630 630 - key, _, value = line.partition("=") 631 631 - if key in missing and value: 632 632 - os.environ[key] = value 633 633 - recovered.append(f"{key}={value}") 634 634 - 635 635 - if recovered: 636 636 - logger.info("Recovered session env from systemd: %s", ", ".join(recovered)) 637 637 - 638 638 - 639 639 - def check_session_ready() -> str | None: 640 640 - """Check if the desktop session is ready for observation. 641 641 - 642 642 - Returns None if ready, or a description of what's missing. 643 643 - """ 644 644 - # Try to recover missing session vars from systemd user manager 645 645 - _recover_session_env() 646 646 - 647 647 - # Display server 648 648 - if not os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"): 649 649 - return "no display server (DISPLAY/WAYLAND_DISPLAY not set)" 650 650 - 651 651 - # DBus session bus 652 652 - if not os.environ.get("DBUS_SESSION_BUS_ADDRESS"): 653 653 - return "no DBus session bus (DBUS_SESSION_BUS_ADDRESS not set)" 654 654 - 655 655 - # PulseAudio / PipeWire audio 656 656 - pactl = shutil.which("pactl") 657 657 - if pactl: 658 658 - try: 659 659 - subprocess.run( 660 660 - [pactl, "info"], 661 661 - capture_output=True, 662 662 - timeout=5, 663 663 - ).check_returncode() 664 664 - except (subprocess.CalledProcessError, subprocess.TimeoutExpired): 665 665 - return "audio server not responding (pactl info failed)" 666 666 - return None 667 667 - 668 668 - 669 669 - async def async_main(args): 670 670 - """Async entry point.""" 671 671 - # Pre-flight: check session prerequisites before attempting setup 672 672 - not_ready = check_session_ready() 673 673 - if not_ready: 674 674 - logger.warning("Session not ready: %s", not_ready) 675 675 - return EXIT_TEMPFAIL 676 676 - 677 677 - observer = Observer( 678 678 - interval=args.interval, 679 679 - ) 680 680 - 681 681 - # Setup signal handlers 682 682 - loop = asyncio.get_running_loop() 683 683 - 684 684 - def signal_handler(): 685 685 - logger.info("Received shutdown signal") 686 686 - observer.running = False 687 687 - 688 688 - for sig in (signal.SIGINT, signal.SIGTERM): 689 689 - loop.add_signal_handler(sig, signal_handler) 690 690 - 691 691 - # Initialize 692 692 - if not await observer.setup(): 693 693 - logger.error("Observer setup failed") 694 694 - return 1 695 695 - 696 696 - # Run main loop 697 697 - try: 698 698 - await observer.main_loop() 699 699 - except RuntimeError as e: 700 700 - logger.error(f"Observer runtime error: {e}") 701 701 - return 1 702 702 - except Exception as e: 703 703 - logger.error(f"Observer error: {e}", exc_info=True) 704 704 - return 1 705 705 - 706 706 - return 0 707 707 - 708 708 - 709 709 - def main(): 710 710 - """CLI entry point.""" 711 711 - parser = argparse.ArgumentParser( 712 712 - description="Unified audio and screencast observer for journaling." 713 713 - ) 714 714 - parser.add_argument( 715 715 - "--interval", 716 716 - type=int, 717 717 - default=300, 718 718 - help="Duration per screencast window in seconds (default: 300 = 5 minutes).", 719 719 - ) 720 720 - args = setup_cli(parser) 721 721 - 722 722 - # Run async main 723 723 - try: 724 724 - rc = asyncio.run(async_main(args)) 725 725 - sys.exit(rc) 726 726 - except KeyboardInterrupt: 727 727 - logger.info("Interrupted by user") 728 728 - sys.exit(0) 729 729 - except Exception as e: 730 730 - logger.error(f"Fatal error: {e}", exc_info=True) 731 731 - sys.exit(1) 732 732 - 733 733 - 734 734 - if __name__ == "__main__": 735 735 - main()

-498

observe/linux/screencast.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - """ 5 5 - Portal-based multi-monitor screencast recording. 6 6 - 7 7 - Uses xdg-desktop-portal ScreenCast API with PipeWire + GStreamer to record 8 8 - each monitor as a separate file. This replaces the old GNOME Shell D-Bus approach. 9 9 - 10 10 - Runtime deps: 11 11 - - xdg-desktop-portal with org.freedesktop.portal.ScreenCast 12 12 - - Portal backend: xdg-desktop-portal-gnome (or -kde, -wlr, etc.) 13 13 - - PipeWire running 14 14 - - GStreamer with PipeWire plugin: gst-launch-1.0 pipewiresrc 15 15 - """ 16 16 - 17 17 - import asyncio 18 18 - import logging 19 19 - import os 20 20 - import signal 21 21 - import subprocess 22 22 - import uuid 23 23 - from dataclasses import dataclass 24 24 - from pathlib import Path 25 25 - 26 26 - from dbus_next import Variant, introspection 27 27 - from dbus_next.aio import MessageBus 28 28 - from dbus_next.constants import BusType 29 29 - 30 30 - from observe.gnome.activity import get_monitor_geometries 31 31 - from think.utils import get_journal 32 32 - 33 33 - # Workaround for dbus-next issue #122: portal has properties with hyphens 34 34 - # (e.g., "power-saver-enabled") which violate strict D-Bus naming validation. 35 35 - introspection.assert_member_name_valid = lambda name: None 36 36 - 37 37 - logger = logging.getLogger(__name__) 38 38 - 39 39 - # Portal D-Bus constants 40 40 - PORTAL_BUS = "org.freedesktop.portal.Desktop" 41 41 - PORTAL_PATH = "/org/freedesktop/portal/desktop" 42 42 - SC_IFACE = "org.freedesktop.portal.ScreenCast" 43 43 - REQ_IFACE = "org.freedesktop.portal.Request" 44 44 - SESSION_IFACE = "org.freedesktop.portal.Session" 45 45 - 46 46 - 47 47 - @dataclass 48 48 - class StreamInfo: 49 49 - """Information about a single monitor's recording stream.""" 50 50 - 51 51 - node_id: int 52 52 - position: str 53 53 - connector: str 54 54 - x: int 55 55 - y: int 56 56 - width: int 57 57 - height: int 58 58 - file_path: str # Final path in segment directory 59 59 - 60 60 - @property 61 61 - def filename(self) -> str: 62 62 - """Return just the filename for event payloads.""" 63 63 - return os.path.basename(self.file_path) 64 64 - 65 65 - 66 66 - def _get_restore_token_path() -> Path: 67 67 - """Get path for restore token storage within the journal health directory.""" 68 68 - return Path(get_journal()) / "health" / "screencast_restore_token" 69 69 - 70 70 - 71 71 - def _load_restore_token() -> str | None: 72 72 - """Load restore token from disk.""" 73 73 - path = _get_restore_token_path() 74 74 - try: 75 75 - data = path.read_text(encoding="utf-8").strip() 76 76 - return data or None 77 77 - except (FileNotFoundError, OSError): 78 78 - return None 79 79 - 80 80 - 81 81 - def _save_restore_token(token: str) -> None: 82 82 - """Save restore token to disk.""" 83 83 - path = _get_restore_token_path() 84 84 - try: 85 85 - path.parent.mkdir(parents=True, exist_ok=True) 86 86 - path.write_text(token.strip() + "\n", encoding="utf-8") 87 87 - logger.debug(f"Saved restore token to {path}") 88 88 - except OSError as e: 89 89 - logger.warning(f"Failed to save restore token: {e}") 90 90 - 91 91 - 92 92 - def _make_request_handle(bus: MessageBus, token: str) -> str: 93 93 - """Compute expected Request object path for a handle_token.""" 94 94 - sender = bus.unique_name.lstrip(":").replace(".", "_") 95 95 - return f"/org/freedesktop/portal/desktop/request/{sender}/{token}" 96 96 - 97 97 - 98 98 - def _prepare_request_handler(bus: MessageBus, handle: str) -> asyncio.Future: 99 99 - """Set up signal handler for Request::Response before calling portal method.""" 100 100 - loop = asyncio.get_running_loop() 101 101 - fut: asyncio.Future = loop.create_future() 102 102 - 103 103 - def _message_handler(msg): 104 104 - if ( 105 105 - msg.message_type.name == "SIGNAL" 106 106 - and msg.path == handle 107 107 - and msg.interface == REQ_IFACE 108 108 - and msg.member == "Response" 109 109 - ): 110 110 - response = msg.body[0] 111 111 - results = msg.body[1] if len(msg.body) > 1 else {} 112 112 - if not fut.done(): 113 113 - fut.set_result((int(response), results)) 114 114 - bus.remove_message_handler(_message_handler) 115 115 - 116 116 - bus.add_message_handler(_message_handler) 117 117 - return fut 118 118 - 119 119 - 120 120 - def _variant_or_value(val): 121 121 - """Extract value from Variant if needed.""" 122 122 - if isinstance(val, Variant): 123 123 - return val.value 124 124 - return val 125 125 - 126 126 - 127 127 - def _match_streams_to_monitors(streams: list[dict], monitors: list[dict]) -> list[dict]: 128 128 - """ 129 129 - Match portal stream geometries to GDK monitor info. 130 130 - 131 131 - Portal streams have position (x, y) and size (width, height). 132 132 - GDK monitors have connector IDs and box coordinates. 133 133 - 134 134 - Returns streams augmented with connector and position labels. 135 135 - """ 136 136 - matched = [] 137 137 - 138 138 - for stream in streams: 139 139 - props = stream.get("props", {}) 140 140 - 141 141 - # Extract stream geometry from portal properties 142 142 - stream_pos = _variant_or_value(props.get("position", (0, 0))) 143 143 - stream_size = _variant_or_value(props.get("size", (0, 0))) 144 144 - 145 145 - if isinstance(stream_pos, (tuple, list)) and len(stream_pos) >= 2: 146 146 - sx, sy = int(stream_pos[0]), int(stream_pos[1]) 147 147 - else: 148 148 - sx, sy = 0, 0 149 149 - 150 150 - if isinstance(stream_size, (tuple, list)) and len(stream_size) >= 2: 151 151 - sw, sh = int(stream_size[0]), int(stream_size[1]) 152 152 - else: 153 153 - sw, sh = 0, 0 154 154 - 155 155 - # Find matching monitor by geometry 156 156 - best_match = None 157 157 - best_overlap = 0 158 158 - 159 159 - for monitor in monitors: 160 160 - mx1, my1, mx2, my2 = monitor["box"] 161 161 - mw, mh = mx2 - mx1, my2 - my1 162 162 - 163 163 - # Check if geometries match (within tolerance for scaling) 164 164 - if abs(sx - mx1) < 10 and abs(sy - my1) < 10: 165 165 - overlap = min(sw, mw) * min(sh, mh) 166 166 - if overlap > best_overlap: 167 167 - best_overlap = overlap 168 168 - best_match = monitor 169 169 - 170 170 - if best_match: 171 171 - stream["connector"] = best_match["id"] 172 172 - stream["position_label"] = best_match.get("position", "unknown") 173 173 - stream["x"] = best_match["box"][0] 174 174 - stream["y"] = best_match["box"][1] 175 175 - stream["width"] = best_match["box"][2] - best_match["box"][0] 176 176 - stream["height"] = best_match["box"][3] - best_match["box"][1] 177 177 - else: 178 178 - # Fallback: use stream index as identifier 179 179 - stream["connector"] = f"monitor-{stream['idx']}" 180 180 - stream["position_label"] = "unknown" 181 181 - stream["x"] = sx 182 182 - stream["y"] = sy 183 183 - stream["width"] = sw 184 184 - stream["height"] = sh 185 185 - 186 186 - matched.append(stream) 187 187 - 188 188 - return matched 189 189 - 190 190 - 191 191 - class Screencaster: 192 192 - """Portal-based multi-monitor screencast manager.""" 193 193 - 194 194 - def __init__(self): 195 195 - self.bus: MessageBus | None = None 196 196 - self.session_handle: str | None = None 197 197 - self.pw_fd: int | None = None 198 198 - self.gst_process: subprocess.Popen | None = None 199 199 - self.streams: list[StreamInfo] = [] 200 200 - self._started = False 201 201 - 202 202 - async def connect(self) -> bool: 203 203 - """ 204 204 - Establish D-Bus connection and verify portal availability. 205 205 - 206 206 - Returns: 207 207 - True if portal is available, False otherwise. 208 208 - """ 209 209 - if self.bus is not None: 210 210 - return True 211 211 - 212 212 - try: 213 213 - self.bus = await MessageBus( 214 214 - bus_type=BusType.SESSION, 215 215 - negotiate_unix_fd=True, 216 216 - ).connect() 217 217 - 218 218 - # Verify portal interface exists 219 219 - root_intro = await self.bus.introspect(PORTAL_BUS, PORTAL_PATH) 220 220 - root_obj = self.bus.get_proxy_object(PORTAL_BUS, PORTAL_PATH, root_intro) 221 221 - root_obj.get_interface(SC_IFACE) 222 222 - return True 223 223 - 224 224 - except Exception as e: 225 225 - logger.error(f"Portal not available: {e}") 226 226 - self.bus = None 227 227 - return False 228 228 - 229 229 - async def start( 230 230 - self, 231 231 - output_dir: str, 232 232 - framerate: int = 1, 233 233 - draw_cursor: bool = True, 234 234 - ) -> list[StreamInfo]: 235 235 - """ 236 236 - Start screencast recording for all monitors. 237 237 - 238 238 - Files are written directly to output_dir with final names (position_connector_screen.webm). 239 239 - The output_dir is typically a draft segment directory that will be renamed on completion. 240 240 - 241 241 - Args: 242 242 - output_dir: Directory for output files (e.g., YYYYMMDD/HHMMSS_draft/) 243 243 - framerate: Frames per second (default: 1) 244 244 - draw_cursor: Whether to draw mouse cursor (default: True) 245 245 - 246 246 - Returns: 247 247 - List of StreamInfo for each monitor being recorded. 248 248 - 249 249 - Raises: 250 250 - RuntimeError: If recording fails to start. 251 251 - """ 252 252 - if not await self.connect(): 253 253 - raise RuntimeError("Portal not available") 254 254 - 255 255 - # Get monitor info from GDK for connector IDs 256 256 - try: 257 257 - monitors = get_monitor_geometries() 258 258 - except Exception as e: 259 259 - logger.warning(f"Failed to get monitor geometries: {e}") 260 260 - monitors = [] 261 261 - 262 262 - # Get portal interface 263 263 - root_intro = await self.bus.introspect(PORTAL_BUS, PORTAL_PATH) 264 264 - root_obj = self.bus.get_proxy_object(PORTAL_BUS, PORTAL_PATH, root_intro) 265 265 - screencast = root_obj.get_interface(SC_IFACE) 266 266 - 267 267 - # 1) CreateSession 268 268 - create_token = "h_" + uuid.uuid4().hex 269 269 - create_handle = _make_request_handle(self.bus, create_token) 270 270 - create_fut = _prepare_request_handler(self.bus, create_handle) 271 271 - 272 272 - create_opts = { 273 273 - "handle_token": Variant("s", create_token), 274 274 - "session_handle_token": Variant("s", "s_" + uuid.uuid4().hex), 275 275 - } 276 276 - 277 277 - await screencast.call_create_session(create_opts) 278 278 - resp, results = await create_fut 279 279 - if resp != 0: 280 280 - raise RuntimeError(f"CreateSession failed with code {resp}") 281 281 - 282 282 - self.session_handle = str(_variant_or_value(results.get("session_handle"))) 283 283 - if not self.session_handle: 284 284 - raise RuntimeError("CreateSession returned no session_handle") 285 285 - 286 286 - logger.debug(f"Portal session: {self.session_handle}") 287 287 - 288 288 - # 2) SelectSources 289 289 - restore_token = _load_restore_token() 290 290 - if restore_token: 291 291 - logger.debug("Using saved restore token") 292 292 - 293 293 - cursor_mode = 1 if draw_cursor else 0 294 294 - 295 295 - select_token = "h_" + uuid.uuid4().hex 296 296 - select_handle = _make_request_handle(self.bus, select_token) 297 297 - select_fut = _prepare_request_handler(self.bus, select_handle) 298 298 - 299 299 - select_opts = { 300 300 - "handle_token": Variant("s", select_token), 301 301 - "types": Variant("u", 1), # 1 = MONITOR 302 302 - "multiple": Variant("b", True), 303 303 - "cursor_mode": Variant("u", cursor_mode), 304 304 - "persist_mode": Variant("u", 2), # Persist until revoked 305 305 - } 306 306 - if restore_token: 307 307 - select_opts["restore_token"] = Variant("s", restore_token) 308 308 - 309 309 - await screencast.call_select_sources(self.session_handle, select_opts) 310 310 - resp, _ = await select_fut 311 311 - if resp != 0: 312 312 - await self._close_session() 313 313 - raise RuntimeError(f"SelectSources failed with code {resp}") 314 314 - 315 315 - # 3) Start 316 316 - start_token = "h_" + uuid.uuid4().hex 317 317 - start_handle = _make_request_handle(self.bus, start_token) 318 318 - start_fut = _prepare_request_handler(self.bus, start_handle) 319 319 - 320 320 - start_opts = {"handle_token": Variant("s", start_token)} 321 321 - await screencast.call_start(self.session_handle, "", start_opts) 322 322 - resp, results = await start_fut 323 323 - if resp != 0: 324 324 - await self._close_session() 325 325 - raise RuntimeError(f"Start failed with code {resp}") 326 326 - 327 327 - portal_streams = _variant_or_value(results.get("streams")) or [] 328 328 - if not portal_streams: 329 329 - await self._close_session() 330 330 - raise RuntimeError("Start returned no streams") 331 331 - 332 332 - # Save new restore token if provided 333 333 - new_token = _variant_or_value(results.get("restore_token")) 334 334 - if isinstance(new_token, str) and new_token.strip(): 335 335 - _save_restore_token(new_token) 336 336 - 337 337 - # Parse streams 338 338 - stream_info = [] 339 339 - for idx, stream in enumerate(portal_streams): 340 340 - try: 341 341 - node_id = int(stream[0]) 342 342 - props = stream[1] if len(stream) > 1 else {} 343 343 - stream_info.append({"idx": idx, "node_id": node_id, "props": props}) 344 344 - except Exception as e: 345 345 - logger.warning(f"Could not parse stream {idx}: {e}") 346 346 - 347 347 - if not stream_info: 348 348 - await self._close_session() 349 349 - raise RuntimeError("No valid streams found") 350 350 - 351 351 - # Match streams to monitors 352 352 - stream_info = _match_streams_to_monitors(stream_info, monitors) 353 353 - 354 354 - logger.info(f"Portal returned {len(stream_info)} stream(s)") 355 355 - 356 356 - # 4) OpenPipeWireRemote 357 357 - fd_obj = await screencast.call_open_pipe_wire_remote(self.session_handle, {}) 358 358 - if hasattr(fd_obj, "take"): 359 359 - self.pw_fd = fd_obj.take() 360 360 - else: 361 361 - self.pw_fd = int(fd_obj) 362 362 - 363 363 - # 5) Build GStreamer pipeline 364 364 - self.streams = [] 365 365 - pipeline_parts = [] 366 366 - 367 367 - for info in stream_info: 368 368 - node_id = info["node_id"] 369 369 - position = info["position_label"] 370 370 - connector = info["connector"] 371 371 - 372 372 - # Final file path: position_connector_screen.webm 373 373 - # Written directly to output_dir (draft segment directory) 374 374 - file_path = os.path.join(output_dir, f"{position}_{connector}_screen.webm") 375 375 - 376 376 - stream_obj = StreamInfo( 377 377 - node_id=node_id, 378 378 - position=position, 379 379 - connector=connector, 380 380 - x=info["x"], 381 381 - y=info["y"], 382 382 - width=info["width"], 383 383 - height=info["height"], 384 384 - file_path=file_path, 385 385 - ) 386 386 - self.streams.append(stream_obj) 387 387 - 388 388 - # GStreamer branch for this stream 389 389 - # VP8 encoding optimized for screen content 390 390 - branch = ( 391 391 - f"pipewiresrc fd={self.pw_fd} path={node_id} ! " 392 392 - f"videorate ! video/x-raw,framerate={framerate}/1 ! " 393 393 - f"videoconvert ! vp8enc end-usage=cq cq-level=4 max-quantizer=15 " 394 394 - f"keyframe-max-dist=30 static-threshold=100 ! webmmux ! " 395 395 - f"filesink location={file_path}" 396 396 - ) 397 397 - pipeline_parts.append(branch) 398 398 - 399 399 - logger.info(f" Stream {node_id}: {position} ({connector}) -> {file_path}") 400 400 - 401 401 - pipeline_str = " ".join(pipeline_parts) 402 402 - cmd = ["gst-launch-1.0", "-e"] + pipeline_str.split() 403 403 - 404 404 - try: 405 405 - self.gst_process = subprocess.Popen( 406 406 - cmd, 407 407 - pass_fds=(self.pw_fd,), 408 408 - stdout=subprocess.DEVNULL, 409 409 - stderr=subprocess.PIPE, 410 410 - ) 411 411 - except FileNotFoundError: 412 412 - await self._close_session() 413 413 - raise RuntimeError("gst-launch-1.0 not found") 414 414 - except Exception as e: 415 415 - await self._close_session() 416 416 - raise RuntimeError(f"Failed to start GStreamer: {e}") 417 417 - 418 418 - # Brief delay to check for immediate failure 419 419 - await asyncio.sleep(0.2) 420 420 - if self.gst_process.poll() is not None: 421 421 - stderr = ( 422 422 - self.gst_process.stderr.read().decode() 423 423 - if self.gst_process.stderr 424 424 - else "" 425 425 - ) 426 426 - await self._close_session() 427 427 - raise RuntimeError(f"GStreamer exited immediately: {stderr[:200]}") 428 428 - 429 429 - self._started = True 430 430 - return self.streams 431 431 - 432 432 - async def stop(self) -> list[StreamInfo]: 433 433 - """ 434 434 - Stop screencast recording gracefully. 435 435 - 436 436 - Returns: 437 437 - List of StreamInfo with file_path for the recorded files. 438 438 - """ 439 439 - streams = self.streams.copy() 440 440 - 441 441 - # Stop GStreamer with SIGINT for clean EOS 442 442 - if self.gst_process and self.gst_process.poll() is None: 443 443 - try: 444 444 - self.gst_process.send_signal(signal.SIGINT) 445 445 - # Wait up to 5 seconds for clean shutdown 446 446 - try: 447 447 - await asyncio.wait_for( 448 448 - asyncio.to_thread(self.gst_process.wait), 449 449 - timeout=5.0, 450 450 - ) 451 451 - except asyncio.TimeoutError: 452 452 - logger.warning("GStreamer did not exit cleanly, killing") 453 453 - self.gst_process.kill() 454 454 - self.gst_process.wait() 455 455 - except Exception as e: 456 456 - logger.warning(f"Error stopping GStreamer: {e}") 457 457 - 458 458 - self.gst_process = None 459 459 - 460 460 - # Close PipeWire fd 461 461 - if self.pw_fd is not None: 462 462 - try: 463 463 - os.close(self.pw_fd) 464 464 - except OSError: 465 465 - pass 466 466 - self.pw_fd = None 467 467 - 468 468 - # Close portal session 469 469 - await self._close_session() 470 470 - 471 471 - self.streams = [] 472 472 - self._started = False 473 473 - 474 474 - return streams 475 475 - 476 476 - async def _close_session(self): 477 477 - """Close the portal session.""" 478 478 - if self.session_handle and self.bus: 479 479 - try: 480 480 - session_intro = await self.bus.introspect( 481 481 - PORTAL_BUS, self.session_handle 482 482 - ) 483 483 - session_obj = self.bus.get_proxy_object( 484 484 - PORTAL_BUS, self.session_handle, session_intro 485 485 - ) 486 486 - session_iface = session_obj.get_interface(SESSION_IFACE) 487 487 - await session_iface.call_close() 488 488 - except Exception: 489 489 - pass 490 490 - self.session_handle = None 491 491 - 492 492 - def is_healthy(self) -> bool: 493 493 - """Check if recording is still running.""" 494 494 - if not self._started: 495 495 - return False 496 496 - if self.gst_process is None: 497 497 - return False 498 498 - return self.gst_process.poll() is None

-40

observe/observer.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - """Unified observer entry point with platform detection. 5 5 - 6 6 - Detects the current platform and delegates to the appropriate 7 7 - platform-specific observer implementation. Currently supports Linux only; 8 8 - macOS capture is handled by the solstone-macos native companion app. 9 9 - """ 10 10 - 11 11 - import sys 12 12 - 13 13 - 14 14 - def main() -> None: 15 15 - """Platform-aware observer entry point. 16 16 - 17 17 - Detects the current platform and calls the appropriate observer: 18 18 - - Linux: observe.linux.observer 19 19 - - macOS: handled by solstone-macos native companion app (not this command) 20 20 - """ 21 21 - platform = sys.platform 22 22 - 23 23 - if platform == "linux": 24 24 - from observe.linux.observer import main as platform_main 25 25 - else: 26 26 - print( 27 27 - f"Error: Observer not available for platform '{platform}'", file=sys.stderr 28 28 - ) 29 29 - print( 30 30 - "Supported platform: Linux. macOS capture is handled by the" 31 31 - " solstone-macos native companion app.", 32 32 - file=sys.stderr, 33 33 - ) 34 34 - sys.exit(1) 35 35 - 36 36 - platform_main() 37 37 - 38 38 - 39 39 - if __name__ == "__main__": 40 40 - main()

-7

pyproject.toml

reviewed

··· 43 43 "timefhuman", 44 44 "Pillow", 45 45 "numpy", 46 46 - # Linux-only: GNOME/GTK integration 47 47 - "PyGObject; sys_platform == 'linux'", 48 48 - "dbus-next; sys_platform == 'linux'", 49 46 "desktop-notifier", 50 47 "setproctitle", 51 48 "av", ··· 66 63 "typer", 67 64 68 65 # Audio processing 69 69 - "soundcard", 70 66 "soundfile", 71 67 "faster-whisper>=1.0.0", 72 68 "resemblyzer>=0.1.0", ··· 119 115 [tool.ruff.lint] 120 116 select = ["F", "E", "W", "I"] 121 117 ignore = ["E501", "E203", "E402"] 122 122 - 123 123 - [tool.ruff.lint.per-file-ignores] 124 124 - "observe/gnome/activity.py" = ["E402"] 125 118 126 119 [tool.mypy] 127 120 python_version = "3.12"

-4

sol.py

reviewed

··· 59 59 "sense": "observe.sense", 60 60 "sync": "observe.sync", 61 61 "transfer": "observe.transfer", 62 62 - "observer": "observe.observer", 63 62 "remote": "observe.remote_cli", 64 64 - "observe-linux": "observe.linux.observer", 65 63 # AI agents (formerly muse package) 66 64 "agents": "think.agents", 67 65 "cortex": "think.cortex", ··· 116 114 "sense", 117 115 "sync", 118 116 "transfer", 119 119 - "observer", 120 117 "remote", 121 118 ], 122 119 "Muse (AI agents)": [ ··· 137 134 "journal-stats", 138 135 "formatter", 139 136 "detect-created", 140 140 - "observe-linux", 141 137 ], 142 138 "Help": ["help", "chat"], 143 139 }

-184

tests/test_session_env.py

reviewed

··· 1 1 - # SPDX-License-Identifier: AGPL-3.0-only 2 2 - # Copyright (c) 2026 sol pbc 3 3 - 4 4 - """Tests for desktop session environment recovery.""" 5 5 - 6 6 - import os 7 7 - import subprocess 8 8 - from unittest.mock import patch 9 9 - 10 10 - from observe.linux.observer import _recover_session_env, check_session_ready 11 11 - 12 12 - 13 13 - class TestRecoverSessionEnv: 14 14 - """Tests for _recover_session_env().""" 15 15 - 16 16 - def test_noop_when_vars_already_set(self, monkeypatch): 17 17 - """Should not call systemctl when all vars are present.""" 18 18 - monkeypatch.setenv("DISPLAY", ":1") 19 19 - monkeypatch.setenv("WAYLAND_DISPLAY", "wayland-0") 20 20 - monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/run/user/1000/bus") 21 21 - 22 22 - with patch("observe.linux.observer.subprocess.run") as mock_run: 23 23 - _recover_session_env() 24 24 - mock_run.assert_not_called() 25 25 - 26 26 - def test_recovers_missing_vars(self, monkeypatch): 27 27 - """Should recover missing vars from systemctl output.""" 28 28 - monkeypatch.delenv("DISPLAY", raising=False) 29 29 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 30 30 - monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False) 31 31 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 32 32 - 33 33 - systemctl_output = ( 34 34 - "HOME=/home/user\n" 35 35 - "DISPLAY=:0\n" 36 36 - "WAYLAND_DISPLAY=wayland-0\n" 37 37 - "DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus\n" 38 38 - "XDG_SESSION_TYPE=wayland\n" 39 39 - ) 40 40 - mock_result = subprocess.CompletedProcess( 41 41 - args=[], returncode=0, stdout=systemctl_output, stderr="" 42 42 - ) 43 43 - with patch("observe.linux.observer.subprocess.run", return_value=mock_result): 44 44 - _recover_session_env() 45 45 - 46 46 - assert os.environ.get("DISPLAY") == ":0" 47 47 - assert os.environ.get("WAYLAND_DISPLAY") == "wayland-0" 48 48 - assert ( 49 49 - os.environ.get("DBUS_SESSION_BUS_ADDRESS") == "unix:path=/run/user/1000/bus" 50 50 - ) 51 51 - 52 52 - def test_recovers_only_missing_vars(self, monkeypatch): 53 53 - """Should not overwrite vars that are already set.""" 54 54 - monkeypatch.setenv("DISPLAY", ":5") 55 55 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 56 56 - monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/run/user/1000/bus") 57 57 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 58 58 - 59 59 - systemctl_output = "DISPLAY=:0\nWAYLAND_DISPLAY=wayland-0\n" 60 60 - mock_result = subprocess.CompletedProcess( 61 61 - args=[], returncode=0, stdout=systemctl_output, stderr="" 62 62 - ) 63 63 - with patch("observe.linux.observer.subprocess.run", return_value=mock_result): 64 64 - _recover_session_env() 65 65 - 66 66 - assert os.environ.get("DISPLAY") == ":5" # unchanged 67 67 - assert os.environ.get("WAYLAND_DISPLAY") == "wayland-0" # recovered 68 68 - 69 69 - def test_sets_xdg_runtime_dir_if_missing(self, monkeypatch): 70 70 - """Should set XDG_RUNTIME_DIR from uid when missing.""" 71 71 - monkeypatch.delenv("DISPLAY", raising=False) 72 72 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 73 73 - monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False) 74 74 - monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) 75 75 - 76 76 - mock_result = subprocess.CompletedProcess( 77 77 - args=[], returncode=0, stdout="DISPLAY=:0\n", stderr="" 78 78 - ) 79 79 - with patch("observe.linux.observer.subprocess.run", return_value=mock_result): 80 80 - _recover_session_env() 81 81 - 82 82 - assert os.environ.get("XDG_RUNTIME_DIR") == f"/run/user/{os.getuid()}" 83 83 - 84 84 - def test_handles_systemctl_failure(self, monkeypatch): 85 85 - """Should silently handle systemctl failure.""" 86 86 - monkeypatch.delenv("DISPLAY", raising=False) 87 87 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 88 88 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 89 89 - 90 90 - mock_result = subprocess.CompletedProcess( 91 91 - args=[], returncode=1, stdout="", stderr="error" 92 92 - ) 93 93 - with patch("observe.linux.observer.subprocess.run", return_value=mock_result): 94 94 - _recover_session_env() 95 95 - 96 96 - assert not os.environ.get("DISPLAY") 97 97 - 98 98 - def test_handles_systemctl_not_found(self, monkeypatch): 99 99 - """Should silently handle missing systemctl binary.""" 100 100 - monkeypatch.delenv("DISPLAY", raising=False) 101 101 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 102 102 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 103 103 - 104 104 - with patch( 105 105 - "observe.linux.observer.subprocess.run", side_effect=FileNotFoundError 106 106 - ): 107 107 - _recover_session_env() 108 108 - 109 109 - assert not os.environ.get("DISPLAY") 110 110 - 111 111 - def test_handles_systemctl_timeout(self, monkeypatch): 112 112 - """Should silently handle systemctl timeout.""" 113 113 - monkeypatch.delenv("DISPLAY", raising=False) 114 114 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 115 115 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 116 116 - 117 117 - with patch( 118 118 - "observe.linux.observer.subprocess.run", 119 119 - side_effect=subprocess.TimeoutExpired(cmd="systemctl", timeout=5), 120 120 - ): 121 121 - _recover_session_env() 122 122 - 123 123 - assert not os.environ.get("DISPLAY") 124 124 - 125 125 - def test_ignores_empty_values(self, monkeypatch): 126 126 - """Should not set vars with empty values.""" 127 127 - monkeypatch.delenv("DISPLAY", raising=False) 128 128 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 129 129 - monkeypatch.setenv("DBUS_SESSION_BUS_ADDRESS", "unix:path=/run/user/1000/bus") 130 130 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 131 131 - 132 132 - systemctl_output = "DISPLAY=\nWAYLAND_DISPLAY=wayland-0\n" 133 133 - mock_result = subprocess.CompletedProcess( 134 134 - args=[], returncode=0, stdout=systemctl_output, stderr="" 135 135 - ) 136 136 - with patch("observe.linux.observer.subprocess.run", return_value=mock_result): 137 137 - _recover_session_env() 138 138 - 139 139 - assert not os.environ.get("DISPLAY") 140 140 - assert os.environ.get("WAYLAND_DISPLAY") == "wayland-0" 141 141 - 142 142 - 143 143 - class TestCheckSessionReady: 144 144 - """Tests for check_session_ready() with env recovery integration.""" 145 145 - 146 146 - def test_ready_after_recovery(self, monkeypatch): 147 147 - """Should pass after recovering vars from systemd.""" 148 148 - monkeypatch.delenv("DISPLAY", raising=False) 149 149 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 150 150 - monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False) 151 151 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 152 152 - 153 153 - systemctl_output = ( 154 154 - "DISPLAY=:0\n" 155 155 - "WAYLAND_DISPLAY=wayland-0\n" 156 156 - "DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus\n" 157 157 - ) 158 158 - mock_result = subprocess.CompletedProcess( 159 159 - args=[], returncode=0, stdout=systemctl_output, stderr="" 160 160 - ) 161 161 - with ( 162 162 - patch("observe.linux.observer.subprocess.run", return_value=mock_result), 163 163 - patch("observe.linux.observer.shutil.which", return_value=None), 164 164 - ): 165 165 - result = check_session_ready() 166 166 - 167 167 - assert result is None 168 168 - 169 169 - def test_fails_when_recovery_incomplete(self, monkeypatch): 170 170 - """Should fail when recovery doesn't provide display vars.""" 171 171 - monkeypatch.delenv("DISPLAY", raising=False) 172 172 - monkeypatch.delenv("WAYLAND_DISPLAY", raising=False) 173 173 - monkeypatch.delenv("DBUS_SESSION_BUS_ADDRESS", raising=False) 174 174 - monkeypatch.setenv("XDG_RUNTIME_DIR", "/run/user/1000") 175 175 - 176 176 - # systemctl returns nothing useful 177 177 - mock_result = subprocess.CompletedProcess( 178 178 - args=[], returncode=0, stdout="HOME=/home/user\n", stderr="" 179 179 - ) 180 180 - with patch("observe.linux.observer.subprocess.run", return_value=mock_result): 181 181 - result = check_session_ready() 182 182 - 183 183 - assert result is not None 184 184 - assert "DISPLAY" in result

+5 -211

tests/test_supervisor.py

reviewed

··· 3 3 4 4 import importlib 5 5 import io 6 6 - import logging 7 6 import os 8 7 import subprocess 9 8 import sys 10 10 - import time 11 9 from unittest.mock import MagicMock 12 10 13 11 import pytest 14 12 15 13 16 16 - def test_check_health(): 17 17 - """Test per-observer health checking based on observe.status event freshness.""" 18 18 - mod = importlib.import_module("think.supervisor") 19 19 - 20 20 - mod._enabled_observers = {"linux-observer"} 21 21 - mod._observer_health = {} 22 22 - 23 23 - # No status events yet - grace period, returns empty 24 24 - stale = mod.check_health(threshold=60) 25 25 - assert stale == [] 26 26 - 27 27 - # Simulate first status from linux observer 28 28 - mod._observer_health["archon"] = {"last_ts": time.time(), "ever_received": True} 29 29 - stale = mod.check_health(threshold=60) 30 30 - assert stale == [] 31 31 - 32 32 - # Linux observer goes stale 33 33 - mod._observer_health["archon"]["last_ts"] = time.time() - 100 34 34 - stale = mod.check_health(threshold=60) 35 35 - assert stale == ["archon"] 36 36 - 37 37 - # Fresh linux observer clears stale health 38 38 - mod._observer_health["archon"]["last_ts"] = time.time() 39 39 - stale = mod.check_health(threshold=60) 40 40 - assert stale == [] 41 41 - 42 42 - 43 43 - def test_check_health_observer_disabled(monkeypatch): 44 44 - """Test that health checks are skipped when no observers are enabled.""" 45 45 - mod = importlib.import_module("think.supervisor") 46 46 - 47 47 - monkeypatch.setattr(mod, "_enabled_observers", set()) 48 48 - 49 49 - # Even with stale health entries, should return empty 50 50 - mod._observer_health["archon"] = {"last_ts": 0.0, "ever_received": True} 51 51 - stale = mod.check_health(threshold=60) 52 52 - assert stale == [] 53 53 - 54 54 - 55 55 - def test_handle_observe_status(): 56 56 - """Test that observe.status events update per-observer health state.""" 57 57 - mod = importlib.import_module("think.supervisor") 58 58 - 59 59 - mod._observer_health = {} 60 60 - 61 61 - # Status with stream field creates health entry 62 62 - mod._handle_observe_status( 63 63 - {"tract": "observe", "event": "status", "stream": "archon"} 64 64 - ) 65 65 - assert "archon" in mod._observer_health 66 66 - assert mod._observer_health["archon"]["last_ts"] > 0 67 67 - assert mod._observer_health["archon"]["ever_received"] is True 68 68 - 69 69 - # Second stream creates separate entry 70 70 - mod._handle_observe_status( 71 71 - {"tract": "observe", "event": "status", "stream": "archon.tmux"} 72 72 - ) 73 73 - assert "archon.tmux" in mod._observer_health 74 74 - assert mod._observer_health["archon.tmux"]["ever_received"] is True 75 75 - 76 76 - # Non-observe messages should be ignored 77 77 - old_ts = mod._observer_health["archon"]["last_ts"] 78 78 - mod._handle_observe_status( 79 79 - {"tract": "supervisor", "event": "status", "stream": "archon"} 80 80 - ) 81 81 - assert mod._observer_health["archon"]["last_ts"] == old_ts 82 82 - 83 83 - # Events without stream field ignored (sense status) 84 84 - mod._observer_health = {} 85 85 - mod._handle_observe_status({"tract": "observe", "event": "status"}) 86 86 - assert mod._observer_health == {} 87 87 - 88 88 - 89 14 @pytest.mark.asyncio 90 15 async def test_send_notification(monkeypatch): 91 16 mod = importlib.import_module("think.supervisor") ··· 140 65 assert len(cleared) == 1 # Still just one clear call 141 66 142 67 143 143 - def test_start_observers_and_sense(tmp_path, mock_callosum, monkeypatch): 144 144 - """Test that linux-observer and sense launch correctly.""" 68 68 + def test_start_sense(tmp_path, mock_callosum, monkeypatch): 69 69 + """Test that sense launches correctly.""" 145 70 mod = importlib.import_module("think.supervisor") 146 71 147 72 started = [] ··· 174 99 monkeypatch.setattr(mod.subprocess, "Popen", fake_popen) 175 100 monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path)) 176 101 177 177 - # Test linux-observer 178 178 - linux_proc = mod._launch_process( 179 179 - "linux-observer", ["sol", "observer", "-v"], restart=True 180 180 - ) 181 181 - assert linux_proc is not None 182 182 - assert any(cmd == ["sol", "observer", "-v"] for cmd, _, _ in started) 183 183 - 184 102 # Test start_sense() 185 103 sense_proc = mod.start_sense() 186 104 assert sense_proc is not None ··· 258 176 assert args.remote is None 259 177 260 178 261 261 - def test_parse_args_observers_flag(): 262 262 - """Test --observers flag parsing.""" 263 263 - mod = importlib.reload(importlib.import_module("think.supervisor")) 264 264 - 265 265 - parser = mod.parse_args() 266 266 - 267 267 - args = parser.parse_args([]) 268 268 - assert args.observers == "linux" 269 269 - 270 270 - args = parser.parse_args(["--observers", "linux"]) 271 271 - assert args.observers == "linux" 272 272 - 273 273 - args = parser.parse_args(["--observers", "none"]) 274 274 - assert args.observers == "none" 275 275 - 276 276 - args = parser.parse_args(["--no-observers"]) 277 277 - assert args.no_observers is True 278 278 - 279 279 - 280 280 - def test_shutdown_pauses_after_observers(monkeypatch): 281 281 - """Shutdown stops observers first, drains 2s, then remaining services.""" 179 179 + def test_shutdown_stops_in_reverse_order(monkeypatch): 180 180 + """Shutdown stops services in reverse order.""" 282 181 mod = importlib.import_module("think.supervisor") 283 182 284 183 operations = [] ··· 311 210 procs = [ 312 211 MockManaged("convey"), 313 212 MockManaged("sense"), 314 314 - MockManaged("linux-observer"), 315 213 MockManaged("cortex"), 316 214 ] 317 215 318 318 - sleep_calls = [] 319 319 - 320 320 - def fake_sleep(seconds): 321 321 - operations.append(("sleep", seconds)) 322 322 - sleep_calls.append(seconds) 323 323 - 324 324 - monkeypatch.setattr(mod.time, "sleep", fake_sleep) 325 325 - 326 326 - observer_procs = [p for p in procs if p.name == "linux-observer"] 327 327 - other_procs = [p for p in procs if p.name != "linux-observer"] 328 328 - 329 329 - for managed in observer_procs: 330 330 - proc = managed.process 331 331 - try: 332 332 - proc.terminate() 333 333 - except Exception: 334 334 - pass 335 335 - try: 336 336 - proc.wait(timeout=managed.shutdown_timeout) 337 337 - except Exception: 338 338 - pass 339 339 - managed.cleanup() 340 340 - 341 341 - if observer_procs: 342 342 - mod.time.sleep(2) 343 343 - 344 344 - for managed in reversed(other_procs): 216 216 + for managed in reversed(procs): 345 217 proc = managed.process 346 218 try: 347 219 proc.terminate() ··· 354 226 managed.cleanup() 355 227 356 228 assert operations == [ 357 357 - ("terminate", "linux-observer"), 358 358 - ("wait", "linux-observer"), 359 359 - ("cleanup", "linux-observer"), 360 360 - ("sleep", 2), 361 229 ("terminate", "cortex"), 362 230 ("wait", "cortex"), 363 231 ("cleanup", "cortex"), ··· 368 236 ("wait", "convey"), 369 237 ("cleanup", "convey"), 370 238 ] 371 371 - assert sleep_calls == [2] 372 372 - 373 373 - operations.clear() 374 374 - sleep_calls.clear() 375 375 - procs_no_obs = [MockManaged("sense"), MockManaged("cortex")] 376 376 - observer_procs = [p for p in procs_no_obs if p.name == "linux-observer"] 377 377 - other_procs = [p for p in procs_no_obs if p.name != "linux-observer"] 378 378 - 379 379 - for managed in observer_procs: 380 380 - managed.process.terminate() 381 381 - managed.process.wait(timeout=managed.shutdown_timeout) 382 382 - managed.cleanup() 383 383 - 384 384 - if observer_procs: 385 385 - mod.time.sleep(2) 386 386 - 387 387 - for managed in reversed(other_procs): 388 388 - managed.process.terminate() 389 389 - managed.process.wait(timeout=managed.shutdown_timeout) 390 390 - managed.cleanup() 391 391 - 392 392 - assert sleep_calls == [] 393 393 - 394 394 - 395 395 - @pytest.mark.asyncio 396 396 - async def test_supervise_logs_recovery(mock_callosum, monkeypatch, caplog): 397 397 - mod = importlib.reload(importlib.import_module("think.supervisor")) 398 398 - mod.shutdown_requested = False 399 399 - 400 400 - health_states = [["archon"], []] 401 401 - time_counter = {"value": 0.0} # Use dict to allow mutation in closure 402 402 - 403 403 - def fake_time(): 404 404 - """Auto-incrementing time mock that won't run out of values.""" 405 405 - current = time_counter["value"] 406 406 - time_counter["value"] += 1.0 407 407 - return current 408 408 - 409 409 - def fake_check_health(threshold): 410 410 - state = health_states.pop(0) 411 411 - if not health_states: 412 412 - mod.shutdown_requested = True 413 413 - return state 414 414 - 415 415 - async def fake_send_notification(*args, **kwargs): 416 416 - pass 417 417 - 418 418 - async def fake_clear_notification(*args, **kwargs): 419 419 - pass 420 420 - 421 421 - async def fake_sleep(_): 422 422 - pass 423 423 - 424 424 - def fake_handle_daily_tasks(): 425 425 - pass 426 426 - 427 427 - monkeypatch.setattr(mod, "check_runner_exits", lambda procs: []) 428 428 - monkeypatch.setattr(mod, "check_health", fake_check_health) 429 429 - monkeypatch.setattr(mod, "send_notification", fake_send_notification) 430 430 - monkeypatch.setattr(mod, "clear_notification", fake_clear_notification) 431 431 - monkeypatch.setattr(mod, "handle_daily_tasks", fake_handle_daily_tasks) 432 432 - monkeypatch.setattr(mod.time, "time", fake_time) 433 433 - monkeypatch.setattr(mod.asyncio, "sleep", fake_sleep) 434 434 - 435 435 - monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", "/test/journal") 436 436 - 437 437 - with caplog.at_level(logging.INFO): 438 438 - await mod.supervise(threshold=1, interval=1, schedule=False, procs=[]) 439 439 - 440 440 - messages = [record.getMessage() for record in caplog.records] 441 441 - assert "archon heartbeat recovered" in messages 442 442 - assert messages.count("Heartbeat OK") == 1 443 443 - 444 444 - mod.shutdown_requested = False 445 239 446 240 447 241 def test_get_command_name():

+9 -208

think/supervisor.py

reviewed

··· 421 421 # Restart request tracking for SIGKILL enforcement 422 422 _restart_requests: dict[str, tuple[float, subprocess.Popen]] = {} 423 423 424 424 - # Per-observer health state, keyed by stream name from observe.status events. 425 425 - # Each value: {"last_ts": float, "ever_received": bool} 426 426 - # Populated when observe.status events arrive with a stream field. 427 427 - _observer_health: dict[str, dict] = {} 428 428 - 429 429 - # Set of enabled observer process names (e.g. {"linux-observer"}). 430 430 - # Empty set means no observers. Used to gate health checks. 431 431 - _enabled_observers: set[str] = set() 432 432 - 433 424 # Track whether running in remote mode (upload-only, no local processing) 434 425 _is_remote_mode: bool = False 435 426 ··· 552 543 ) 553 544 554 545 555 555 - def check_health(threshold: int = DEFAULT_THRESHOLD) -> list[str]: 556 556 - """Return a list of stale observer stream names based on observe.status events. 557 557 - 558 558 - Returns stream names of observers that haven't sent status within threshold. 559 559 - During startup grace period (before first status event per stream), 560 560 - returns empty list for that stream to avoid false alerts. 561 561 - 562 562 - When no observers are enabled, always returns empty list. 563 563 - """ 564 564 - if not _enabled_observers: 565 565 - return [] 566 566 - 567 567 - stale = [] 568 568 - now = time.time() 569 569 - for stream, state in _observer_health.items(): 570 570 - if not state["ever_received"]: 571 571 - continue 572 572 - if now - state["last_ts"] > threshold: 573 573 - stale.append(stream) 574 574 - 575 575 - return stale 576 576 - 577 577 - 578 578 - def _latest_observe_ts() -> float: 579 579 - """Return the most recent observe.status timestamp across all observers.""" 580 580 - if not _observer_health: 581 581 - return 0.0 582 582 - return max(s["last_ts"] for s in _observer_health.values()) 583 583 - 584 584 - 585 546 def _get_notifier() -> DesktopNotifier: 586 547 """Get or create the global desktop notifier instance.""" 587 548 global _notifier ··· 794 755 # Running tasks 795 756 tasks = _task_queue.collect_task_status() if _task_queue else [] 796 757 queues = _task_queue.collect_queue_counts() if _task_queue else {} 797 797 - 798 798 - # Stale heartbeats 799 799 - stale = check_health() 800 758 801 759 # Scheduled tasks 802 760 schedules = scheduler.collect_status() ··· 808 766 "crashed": crashed, 809 767 "tasks": tasks, 810 768 "queues": queues, 811 811 - "stale_heartbeats": stale, 769 769 + "stale_heartbeats": [], 812 770 "schedules": schedules, 813 771 "callosum_clients": callosum_clients, 814 772 } ··· 1011 969 logging.info("Not restarting %s", managed.name) 1012 970 1013 971 1014 1014 - async def handle_health_checks( 1015 1015 - last_check: float, 1016 1016 - interval: int, 1017 1017 - threshold: int, 1018 1018 - alert_mgr: AlertManager, 1019 1019 - prev_stale: set[str], 1020 1020 - ) -> tuple[float, set[str]]: 1021 1021 - """Perform periodic health checks. Returns (new_last_check, new_prev_stale).""" 1022 1022 - now = time.time() 1023 1023 - if now - last_check < interval: 1024 1024 - return last_check, prev_stale 1025 1025 - 1026 1026 - stale = check_health(threshold) 1027 1027 - stale_set = set(stale) 1028 1028 - 1029 1029 - recovered = sorted(prev_stale - stale_set) 1030 1030 - for name in recovered: 1031 1031 - logging.info("%s heartbeat recovered", name) 1032 1032 - # Clear notifications for recovered heartbeats 1033 1033 - stale_key = ("stale", tuple(sorted(prev_stale))) 1034 1034 - await alert_mgr.clear(stale_key) 1035 1035 - 1036 1036 - # Write capture status to awareness on transitions 1037 1037 - if stale_set != prev_stale: 1038 1038 - try: 1039 1039 - from think.awareness import update_state 1040 1040 - 1041 1041 - if stale_set: 1042 1042 - update_state( 1043 1043 - "capture", 1044 1044 - { 1045 1045 - "status": "stale", 1046 1046 - "last_seen": _latest_observe_ts(), 1047 1047 - }, 1048 1048 - ) 1049 1049 - elif prev_stale: 1050 1050 - update_state( 1051 1051 - "capture", 1052 1052 - { 1053 1053 - "status": "ok", 1054 1054 - "last_seen": _latest_observe_ts(), 1055 1055 - }, 1056 1056 - ) 1057 1057 - except Exception: 1058 1058 - logging.debug("Failed to write capture status to awareness", exc_info=True) 1059 1059 - 1060 1060 - if stale_set: 1061 1061 - msg = f"Journaling offline: {', '.join(sorted(stale_set))}" 1062 1062 - logging.warning(msg) 1063 1063 - 1064 1064 - stale_key = ("stale", tuple(sorted(stale_set))) 1065 1065 - 1066 1066 - # Clear any previous stale notifications with different keys 1067 1067 - for key in list(_notification_ids.keys()): 1068 1068 - if key[0] == "stale" and key != stale_key: 1069 1069 - await clear_notification(key) 1070 1070 - 1071 1071 - await alert_mgr.alert_if_ready(stale_key, msg) 1072 1072 - 1073 1073 - # Retain only alert state entries still relevant 1074 1074 - alert_mgr.clear_matching( 1075 1075 - lambda k, v: k[0] == "stale" and not set(k[1]).issubset(stale_set) 1076 1076 - ) 1077 1077 - else: 1078 1078 - if prev_stale: 1079 1079 - logging.info("Heartbeat OK") 1080 1080 - # Clear alert state for stale services when they recover 1081 1081 - alert_mgr.clear_matching(lambda k, v: k[0] == "stale") 1082 1082 - 1083 1083 - return now, stale_set 1084 1084 - 1085 1085 - 1086 972 def handle_daily_tasks() -> None: 1087 973 """Check for day change and submit daily dream for updated days (non-blocking). 1088 974 ··· 1243 1129 ) 1244 1130 1245 1131 1246 1246 - def _handle_observe_status(message: dict) -> None: 1247 1247 - """Handle observe.status events for per-observer health monitoring. 1248 1248 - 1249 1249 - Tracks status freshness per stream. The stream field identifies the 1250 1250 - observer (e.g. "archon" for linux, "archon.tmux" for tmux). 1251 1251 - Events without a stream field (e.g. from sense) are ignored for health. 1252 1252 - """ 1253 1253 - if message.get("tract") != "observe" or message.get("event") != "status": 1254 1254 - return 1255 1255 - 1256 1256 - stream = message.get("stream") 1257 1257 - if not stream: 1258 1258 - return # sense status events don't have stream field 1259 1259 - 1260 1260 - if stream not in _observer_health: 1261 1261 - _observer_health[stream] = {"last_ts": 0.0, "ever_received": False} 1262 1262 - 1263 1263 - _observer_health[stream]["last_ts"] = time.time() 1264 1264 - _observer_health[stream]["ever_received"] = True 1265 1265 - 1266 1266 - 1267 1132 def _handle_segment_event_log(message: dict) -> None: 1268 1133 """Log observe, dream, and activity events with day+segment to segment/events.jsonl. 1269 1134 ··· 1370 1235 _handle_supervisor_request(message) 1371 1236 _handle_code_shipped(message) 1372 1237 _handle_segment_observed(message) 1373 1373 - _handle_observe_status(message) 1374 1238 _handle_activity_recorded(message) 1375 1239 _handle_dream_daily_complete(message) 1376 1240 _handle_segment_event_log(message) ··· 1378 1242 1379 1243 async def supervise( 1380 1244 *, 1381 1381 - threshold: int = DEFAULT_THRESHOLD, 1382 1382 - interval: int = CHECK_INTERVAL, 1383 1245 daily: bool = True, 1384 1246 schedule: bool = True, 1385 1247 procs: list[ManagedProcess] | None = None, 1386 1248 ) -> None: 1387 1387 - """Monitor health via Callosum events and alert when stale. 1249 1249 + """Main supervision loop. Runs at 1-second intervals for responsiveness. 1388 1250 1389 1389 - Health is derived from observe.status events (see check_health()). 1390 1390 - Main supervision loop runs at 1-second intervals for responsiveness. 1391 1391 - Subsystems manage their own timing (health checks every interval seconds, 1392 1392 - scheduled agents check continuously but only advance when ready). 1251 1251 + Monitors runner health, emits status, triggers daily processing, 1252 1252 + and checks scheduled agents. 1393 1253 """ 1394 1254 alert_mgr = AlertManager() 1395 1395 - last_health_check = 0.0 1396 1255 last_status_emit = 0.0 1397 1397 - prev_stale: set[str] = set() 1398 1256 1399 1257 try: 1400 1258 while ( ··· 1418 1276 if procs: 1419 1277 await handle_runner_exits(procs, alert_mgr) 1420 1278 1421 1421 - # Check health periodically (interval-based timing) 1422 1422 - last_health_check, prev_stale = await handle_health_checks( 1423 1423 - last_health_check, interval, threshold, alert_mgr, prev_stale 1424 1424 - ) 1425 1425 - 1426 1279 # Emit status every 5 seconds 1427 1280 now = time.time() 1428 1281 if now - last_status_emit >= 5: ··· 1471 1324 "--interval", type=int, default=CHECK_INTERVAL, help="Polling interval seconds" 1472 1325 ) 1473 1326 parser.add_argument( 1474 1474 - "--no-observers", 1475 1475 - action="store_true", 1476 1476 - help="Do not start observers (sense still runs for remote/imports)", 1477 1477 - ) 1478 1478 - parser.add_argument( 1479 1479 - "--observers", 1480 1480 - type=str, 1481 1481 - default="linux", 1482 1482 - help="Comma-separated observers to start: linux, none (default: linux)", 1483 1483 - ) 1484 1484 - parser.add_argument( 1485 1327 "--no-daily", 1486 1328 action="store_true", 1487 1329 help="Disable daily processing run at midnight", ··· 1588 1430 print(f"Journal: {path} (from {source})") 1589 1431 logging.info("Supervisor starting...") 1590 1432 1591 1591 - global _managed_procs, _supervisor_callosum, _enabled_observers, _is_remote_mode 1433 1433 + global _managed_procs, _supervisor_callosum, _is_remote_mode 1592 1434 global _task_queue 1593 1435 procs: list[ManagedProcess] = [] 1594 1436 convey_port = None 1595 1437 1596 1438 # Remote mode: run sync instead of local processing 1597 1439 _is_remote_mode = bool(args.remote) 1598 1598 - if args.no_observers: 1599 1599 - _enabled_observers = set() 1600 1600 - else: 1601 1601 - obs_names = [o.strip() for o in args.observers.split(",")] 1602 1602 - if "none" in obs_names: 1603 1603 - _enabled_observers = set() 1604 1604 - else: 1605 1605 - valid = {"linux"} 1606 1606 - for name in obs_names: 1607 1607 - if name not in valid: 1608 1608 - parser.error( 1609 1609 - f"Invalid observer: {name}. Choose from: linux, none" 1610 1610 - ) 1611 1611 - _enabled_observers = {f"{name}-observer" for name in obs_names} 1612 1440 1613 1441 # Start Callosum in-process first - it's the message bus that other services depend on 1614 1442 try: ··· 1655 1483 parser.error(f"Remote server not available: {message}") 1656 1484 logging.info(f"Remote server verified: {message}") 1657 1485 procs.append(start_sync(args.remote)) 1658 1658 - # Start enabled observers (they upload to remote convey ingest) 1659 1659 - if "linux-observer" in _enabled_observers: 1660 1660 - procs.append( 1661 1661 - _launch_process( 1662 1662 - "linux-observer", ["sol", "observer", "-v"], restart=True 1663 1663 - ) 1664 1664 - ) 1665 1486 else: 1666 1666 - # Local mode: convey first (observers upload via HTTP to convey ingest) 1487 1487 + # Local mode: convey first, then sense for file processing 1667 1488 if not args.no_convey: 1668 1489 proc, convey_port = start_convey_server( 1669 1490 verbose=args.verbose, debug=args.debug, port=args.port ··· 1671 1492 procs.append(proc) 1672 1493 # Sense handles file processing 1673 1494 procs.append(start_sense()) 1674 1674 - # Start enabled observers 1675 1675 - if "linux-observer" in _enabled_observers: 1676 1676 - procs.append( 1677 1677 - _launch_process( 1678 1678 - "linux-observer", ["sol", "observer", "-v"], restart=True 1679 1679 - ) 1680 1680 - ) 1681 1681 - # Cortex after observers 1495 1495 + # Cortex for agent execution 1682 1496 if not args.no_cortex: 1683 1497 procs.append(start_cortex_server()) 1684 1498 ··· 1707 1521 try: 1708 1522 asyncio.run( 1709 1523 supervise( 1710 1710 - threshold=args.threshold, 1711 1711 - interval=args.interval, 1712 1524 daily=daily_enabled, 1713 1525 schedule=schedule_enabled, 1714 1526 procs=procs if procs else None, ··· 1719 1531 finally: 1720 1532 logging.info("Stopping all processes...") 1721 1533 print("\nShutting down gracefully (this may take a moment)...", flush=True) 1722 1722 - observer_procs = [p for p in procs if p.name == "linux-observer"] 1723 1723 - other_procs = [p for p in procs if p.name != "linux-observer"] 1724 1534 1725 1535 def _stop_process(managed: ManagedProcess) -> None: 1726 1536 name = managed.name ··· 1745 1555 pass 1746 1556 managed.cleanup() 1747 1557 1748 1748 - # Stop observers first 1749 1749 - for managed in observer_procs: 1750 1750 - _stop_process(managed) 1751 1751 - 1752 1752 - # Drain pause: let in-flight HTTP uploads complete to convey 1753 1753 - if observer_procs: 1754 1754 - logging.info("Pausing for observer uploads to drain") 1755 1755 - time.sleep(2) 1756 1756 - 1757 1757 - # Stop remaining services in reverse order 1758 1758 - for managed in reversed(other_procs): 1558 1558 + # Stop services in reverse order 1559 1559 + for managed in reversed(procs): 1759 1560 _stop_process(managed) 1760 1561 1761 1562 # Save scheduler state before disconnecting

-44

uv.lock

reviewed

··· 726 726 ] 727 727 728 728 [[package]] 729 729 - name = "dbus-next" 730 730 - version = "0.2.3" 731 731 - source = { registry = "https://pypi.org/simple" } 732 732 - sdist = { url = "https://files.pythonhosted.org/packages/ce/45/6a40fbe886d60a8c26f480e7d12535502b5ba123814b3b9a0b002ebca198/dbus_next-0.2.3.tar.gz", hash = "sha256:f4eae26909332ada528c0a3549dda8d4f088f9b365153952a408e28023a626a5", size = 71112, upload-time = "2021-07-25T22:11:28.398Z" } 733 733 - wheels = [ 734 734 - { url = "https://files.pythonhosted.org/packages/d2/fc/c0a3f4c4eaa5a22fbef91713474666e13d0ea2a69c84532579490a9f2cc8/dbus_next-0.2.3-py3-none-any.whl", hash = "sha256:58948f9aff9db08316734c0be2a120f6dc502124d9642f55e90ac82ffb16a18b", size = 57885, upload-time = "2021-07-25T22:11:25.466Z" }, 735 735 - ] 736 736 - 737 737 - [[package]] 738 729 name = "decorator" 739 730 version = "5.2.1" 740 731 source = { registry = "https://pypi.org/simple" } ··· 2398 2389 ] 2399 2390 2400 2391 [[package]] 2401 2401 - name = "pycairo" 2402 2402 - version = "1.29.0" 2403 2403 - source = { registry = "https://pypi.org/simple" } 2404 2404 - sdist = { url = "https://files.pythonhosted.org/packages/22/d9/1728840a22a4ef8a8f479b9156aa2943cd98c3907accd3849fb0d5f82bfd/pycairo-1.29.0.tar.gz", hash = "sha256:f3f7fde97325cae80224c09f12564ef58d0d0f655da0e3b040f5807bd5bd3142", size = 665871, upload-time = "2025-11-11T19:13:01.584Z" } 2405 2405 - 2406 2406 - [[package]] 2407 2392 name = "pycparser" 2408 2393 version = "3.0" 2409 2394 source = { registry = "https://pypi.org/simple" } ··· 2579 2564 wheels = [ 2580 2565 { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" }, 2581 2566 ] 2582 2582 - 2583 2583 - [[package]] 2584 2584 - name = "pygobject" 2585 2585 - version = "3.54.5" 2586 2586 - source = { registry = "https://pypi.org/simple" } 2587 2587 - dependencies = [ 2588 2588 - { name = "pycairo" }, 2589 2589 - ] 2590 2590 - sdist = { url = "https://files.pythonhosted.org/packages/d3/a5/68f883df1d8442e3b267cb92105a4b2f0de819bd64ac9981c2d680d3f49f/pygobject-3.54.5.tar.gz", hash = "sha256:b6656f6348f5245606cf15ea48c384c7f05156c75ead206c1b246c80a22fb585", size = 1274658, upload-time = "2025-10-18T13:45:03.121Z" } 2591 2567 2592 2568 [[package]] 2593 2569 name = "pyjwt" ··· 3530 3506 { name = "anthropic" }, 3531 3507 { name = "av" }, 3532 3508 { name = "blessed" }, 3533 3533 - { name = "dbus-next", marker = "sys_platform == 'linux'" }, 3534 3509 { name = "desktop-notifier" }, 3535 3510 { name = "faster-whisper" }, 3536 3511 { name = "flask", extra = ["async"] }, ··· 3550 3525 { name = "pillow" }, 3551 3526 { name = "playwright" }, 3552 3527 { name = "psutil" }, 3553 3553 - { name = "pygobject", marker = "sys_platform == 'linux'" }, 3554 3528 { name = "pypdf" }, 3555 3529 { name = "pytesseract" }, 3556 3530 { name = "pytest" }, ··· 3565 3539 { name = "resemblyzer" }, 3566 3540 { name = "ruff" }, 3567 3541 { name = "setproctitle" }, 3568 3568 - { name = "soundcard" }, 3569 3542 { name = "soundfile" }, 3570 3543 { name = "timefhuman" }, 3571 3544 { name = "typer" }, ··· 3578 3551 { name = "anthropic" }, 3579 3552 { name = "av" }, 3580 3553 { name = "blessed", specifier = ">=1.20.0" }, 3581 3581 - { name = "dbus-next", marker = "sys_platform == 'linux'" }, 3582 3554 { name = "desktop-notifier" }, 3583 3555 { name = "faster-whisper", specifier = ">=1.0.0" }, 3584 3556 { name = "flask", extras = ["async"] }, ··· 3597 3569 { name = "pillow" }, 3598 3570 { name = "playwright", specifier = ">=1.40.0" }, 3599 3571 { name = "psutil" }, 3600 3600 - { name = "pygobject", marker = "sys_platform == 'linux'" }, 3601 3572 { name = "pypdf" }, 3602 3573 { name = "pytesseract" }, 3603 3574 { name = "pytest" }, ··· 3612 3583 { name = "resemblyzer", specifier = ">=0.1.0" }, 3613 3584 { name = "ruff" }, 3614 3585 { name = "setproctitle" }, 3615 3615 - { name = "soundcard" }, 3616 3586 { name = "soundfile" }, 3617 3587 { name = "timefhuman" }, 3618 3588 { name = "typer" }, 3619 3589 { name = "tzlocal" }, 3620 3590 { name = "webrtcvad-wheels", specifier = ">=2.0.12" }, 3621 3621 - ] 3622 3622 - 3623 3623 - [[package]] 3624 3624 - name = "soundcard" 3625 3625 - version = "0.4.5" 3626 3626 - source = { registry = "https://pypi.org/simple" } 3627 3627 - dependencies = [ 3628 3628 - { name = "cffi" }, 3629 3629 - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, 3630 3630 - { name = "numpy", version = "2.3.5", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, 3631 3631 - ] 3632 3632 - sdist = { url = "https://files.pythonhosted.org/packages/b8/b4/2064f857a9e6e6e3b152fae339f32beb65cfbcd11878d2e4a020e88fc36d/soundcard-0.4.5.tar.gz", hash = "sha256:07272ba927e32cafdf634e4a1ca53b9a3218321a60c7d2e08f54b832a56946aa", size = 40831, upload-time = "2025-09-15T19:08:42.95Z" } 3633 3633 - wheels = [ 3634 3634 - { url = "https://files.pythonhosted.org/packages/fa/d5/700e3018056716c627d8ea783594fc939f74c5878a671c0673281d9e069d/soundcard-0.4.5-py3-none-any.whl", hash = "sha256:8a19c3d5486250bb2990dbd91a75ddf9f15cab39f24ff607e5bce812f43b4917", size = 43923, upload-time = "2025-09-15T19:08:41.339Z" }, 3635 3591 ] 3636 3592 3637 3593 [[package]]