Initial solstone-linux package — standalone Linux desktop observer

Extracts the Linux desktop observer (screen + audio capture) from the
solstone monorepo into a standalone package. Follows the same pattern
as solstone-tmux (phase 5a) with key improvements:

- Local cache + async sync instead of inline upload at segment boundary
- Circuit breaker tuned by error type (auth=immediate, transient=5-10)
- Respects configured sync_max_retries (no hard min(config,3) cap)
- Recovery .metadata file with start timestamp for accurate duration
- Synced-days pruning at 90 days
- Session env recovery for CLI launch, PassEnvironment for systemd
- Screencast restore token at ~/.local/share/solstone-linux/config/

48 tests passing covering config, streams, monitor positions, recovery,
sync collection, error classification, circuit breaker thresholds, and
session readiness checks.

Co-Authored-By: Claude Opus 4.6 (1M context) <[e-mail]>

LICENSE

···
+GNU AFFERO GENERAL PUBLIC LICENSE
+Version 3, 19 November 2007
+
+Copyright (c) 2026 sol pbc
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published
+by the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.

README.md

···
+# solstone-linux
+
+Standalone Linux desktop observer for [solstone](https://solpbc.org). Captures screen and audio from a GNOME Wayland session, stores segments locally, and syncs to a solstone server.
+
+## System Dependencies
+
+**Fedora:**
+```bash
+dnf install python3-gobject gtk4 gstreamer1-plugins-base pipewire-gstreamer alsa-lib-devel pulseaudio-utils pipewire-pulseaudio
+```
+
+**Debian/Ubuntu:**
+```bash
+apt install python3-gi gir1.2-gdk-4.0 gir1.2-gtk-4.0 gstreamer1.0-pipewire libasound2-dev pulseaudio-utils pipewire-pulse
+```
+
+## Install
+
+```bash
+pipx install --system-site-packages solstone-linux
+```
+
+`--system-site-packages` is required for PyGObject/GStreamer access.
+
+## Setup
+
+```bash
+solstone-linux setup
+```
+
+## Run
+
+```bash
+# Foreground
+solstone-linux run
+
+# As a systemd user service
+solstone-linux install-service
+```
+
+## Status
+
+```bash
+solstone-linux status
+```
+
+## License
+
+AGPL-3.0-only — Copyright (c) 2026 sol pbc

contrib/solstone-linux.service

···
+[Unit]
+Description=Solstone Linux Desktop Observer
+After=graphical-session.target
+BindsTo=graphical-session.target
+
+[Service]
+Type=simple
+ExecStart=/usr/bin/solstone-linux run
+PassEnvironment=DISPLAY WAYLAND_DISPLAY DBUS_SESSION_BUS_ADDRESS XDG_RUNTIME_DIR
+Restart=on-failure
+RestartSec=10
+StartLimitIntervalSec=300
+StartLimitBurst=5
+
+[Install]
+WantedBy=graphical-session.target

pyproject.toml

···
+[project]
+name = "solstone-linux"
+version = "0.1.0"
+description = "Standalone Linux desktop observer for solstone"
+readme = "README.md"
+license = "AGPL-3.0-only"
+requires-python = ">=3.10"
+dependencies = [
+    "requests",
+    "numpy",
+    "soundfile",
+    "soundcard",
+    "dbus-next",
+    "PyGObject",
+]
+
+[project.scripts]
+solstone-linux = "solstone_linux.cli:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

src/solstone_linux/__init__.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Standalone Linux desktop observer for solstone."""
+
+__version__ = "0.1.0"

src/solstone_linux/activity.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""GNOME-specific activity detection using Mutter and GTK DBus APIs.
+
+Extracted from solstone's observe/gnome/activity.py.
+
+Changes from monorepo version:
+- Replaces `from observe.utils import assign_monitor_positions` with local module
+"""
+
+import os
+
+import gi
+from dbus_next.aio import MessageBus
+
+gi.require_version("Gdk", "4.0")  # noqa: E402
+gi.require_version("Gtk", "4.0")  # noqa: E402
+from gi.repository import Gdk, Gtk  # noqa: E402
+
+# DBus service constants
+IDLE_MONITOR_BUS = "org.gnome.Mutter.IdleMonitor"
+IDLE_MONITOR_PATH = "/org/gnome/Mutter/IdleMonitor/Core"
+IDLE_MONITOR_IFACE = "org.gnome.Mutter.IdleMonitor"
+
+SCREENSAVER_BUS = "org.gnome.ScreenSaver"
+SCREENSAVER_PATH = "/org/gnome/ScreenSaver"
+SCREENSAVER_IFACE = "org.gnome.ScreenSaver"
+
+DISPLAY_CONFIG_BUS = "org.gnome.Mutter.DisplayConfig"
+DISPLAY_CONFIG_PATH = "/org/gnome/Mutter/DisplayConfig"
+DISPLAY_CONFIG_IFACE = "org.gnome.Mutter.DisplayConfig"
+
+
+async def get_idle_time_ms(bus: MessageBus) -> int:
+    """
+    Get the current idle time in milliseconds.
+
+    Args:
+        bus: Connected DBus session bus
+
+    Returns:
+        Idle time in milliseconds
+    """
+    introspection = await bus.introspect(IDLE_MONITOR_BUS, IDLE_MONITOR_PATH)
+    proxy_obj = bus.get_proxy_object(IDLE_MONITOR_BUS, IDLE_MONITOR_PATH, introspection)
+    idle_monitor = proxy_obj.get_interface(IDLE_MONITOR_IFACE)
+    idle_time = await idle_monitor.call_get_idletime()
+    return idle_time
+
+
+async def is_screen_locked(bus: MessageBus) -> bool:
+    """
+    Check if the screen is currently locked using GNOME ScreenSaver.
+
+    Args:
+        bus: Connected DBus session bus
+
+    Returns:
+        True if screen is locked, False otherwise
+    """
+    try:
+        intro = await bus.introspect(SCREENSAVER_BUS, SCREENSAVER_PATH)
+        obj = bus.get_proxy_object(SCREENSAVER_BUS, SCREENSAVER_PATH, intro)
+        iface = obj.get_interface(SCREENSAVER_IFACE)
+        return bool(await iface.call_get_active())
+    except Exception:
+        return False
+
+
+async def is_power_save_active(bus: MessageBus) -> bool:
+    """
+    Check if display power save mode is active (screen blanked).
+
+    Args:
+        bus: Connected DBus session bus
+
+    Returns:
+        True if power save is active, False otherwise
+    """
+    try:
+        intro = await bus.introspect(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH)
+        obj = bus.get_proxy_object(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH, intro)
+        iface = obj.get_interface("org.freedesktop.DBus.Properties")
+        mode_variant = await iface.call_get(DISPLAY_CONFIG_IFACE, "PowerSaveMode")
+        mode = int(mode_variant.value)
+        return mode != 0
+    except Exception:
+        return False
+
+
+def get_monitor_geometries() -> list[dict]:
+    """
+    Get structured monitor information.
+
+    Returns:
+        List of dicts with format:
+        [{"id": "connector-id", "box": [x1, y1, x2, y2], "position": "center|left|right|..."}, ...]
+        where box contains [left, top, right, bottom] coordinates
+    """
+    from .monitor_positions import assign_monitor_positions
+
+    # Initialize GTK before using GDK functions
+    Gtk.init()
+
+    # Get the default display. If it is None, try opening one from the environment.
+    display = Gdk.Display.get_default()
+    if display is None:
+        env_display = os.environ.get("WAYLAND_DISPLAY") or os.environ.get("DISPLAY")
+        if env_display is not None:
+            display = Gdk.Display.open(env_display)
+        if display is None:
+            raise RuntimeError("No display available")
+    monitors = display.get_monitors()
+
+    # Collect monitor geometries
+    geometries = []
+    for monitor in monitors:
+        geom = monitor.get_geometry()
+        connector = monitor.get_connector() or f"monitor-{len(geometries)}"
+        geometries.append(
+            {
+                "id": connector,
+                "box": [geom.x, geom.y, geom.x + geom.width, geom.y + geom.height],
+            }
+        )
+
+    # Assign position labels using shared algorithm
+    return assign_monitor_positions(geometries)

src/solstone_linux/audio_detect.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Audio device detection via ultrasonic tone.
+
+Direct copy from solstone's observe/detect.py — no solstone imports.
+Plays an ultrasonic tone and records from all mics to identify
+microphone vs loopback devices.
+"""
+
+import logging
+import threading
+
+import numpy as np
+import soundcard as sc
+
+logger = logging.getLogger(__name__)
+
+
+def input_detect(duration=0.4, sample_rate=44100):
+    t = np.linspace(0, duration, int(sample_rate * duration), endpoint=False)
+    tone = 0.5 * np.sin(2 * np.pi * 18000 * t)  # ultrasonic
+
+    try:
+        devices = sc.all_microphones(include_loopback=True)
+    except Exception:
+        logger.warning("Failed to enumerate audio devices")
+        return None, None
+    if not devices:
+        logger.warning("No audio devices found")
+        return None, None
+
+    results = {}
+    barrier = threading.Barrier(len(devices) + 1)
+
+    def record_mic(mic, results):
+        barrier.wait()
+        try:
+            audio = mic.record(
+                samplerate=sample_rate, numframes=int(sample_rate * duration)
+            )
+            results[mic.name] = audio
+        except Exception:
+            results[mic.name] = None
+
+    def play_tone():
+        barrier.wait()
+        try:
+            sp = sc.default_speaker()
+            sp.play(tone, samplerate=sample_rate)
+        except Exception:
+            logger.warning("No default speaker available for tone detection")
+
+    threads = []
+    for mic in devices:
+        thread = threading.Thread(target=record_mic, args=(mic, results))
+        thread.start()
+        threads.append(thread)
+
+    play_thread = threading.Thread(target=play_tone)
+    play_thread.start()
+    threads.append(play_thread)
+
+    for thread in threads:
+        thread.join()
+
+    # Analyze the recordings with a simple amplitude threshold
+    threshold = 0.001
+    mic_detected = None
+    loopback_detected = None
+    for mic in devices:
+        audio = results.get(mic.name)
+        if audio is not None and np.max(np.abs(audio)) > threshold:
+            # First match for each category
+            if "microphone" in str(mic).lower() and mic_detected is None:
+                mic_detected = mic
+            if "loopback" in str(mic).lower() and loopback_detected is None:
+                loopback_detected = mic
+    return mic_detected, loopback_detected

src/solstone_linux/audio_mute.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Linux audio mute detection using PulseAudio/PipeWire.
+
+Direct copy from solstone's observe/linux/audio.py — no solstone imports.
+"""
+
+import asyncio
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def is_sink_muted() -> bool:
+    """
+    Check if the default audio sink is muted using PulseAudio.
+
+    Uses `pactl get-sink-mute @DEFAULT_SINK@` to query mute status.
+
+    Returns:
+        True if muted, False otherwise (including on error).
+    """
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            "pactl",
+            "get-sink-mute",
+            "@DEFAULT_SINK@",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await proc.communicate()
+
+        if proc.returncode != 0:
+            stderr_text = stderr.decode().strip() if stderr else ""
+            logger.warning(f"pactl failed (rc={proc.returncode}): {stderr_text}")
+            return False
+
+        output = stdout.decode().strip()
+        return "Mute: yes" in output
+
+    except FileNotFoundError:
+        logger.warning("pactl not found, assuming unmuted")
+        return False
+    except Exception as e:
+        logger.warning(f"Error checking sink mute status: {e}")
+        return False

src/solstone_linux/audio_recorder.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Audio recording for Linux desktop observer.
+
+Extracted from solstone's observe/hear.py — AudioRecorder class only.
+load_transcript() and format_audio() remain in solstone core (used by 15+ files).
+
+Changes from monorepo version:
+- Replaces `from observe.detect import input_detect` with local audio_detect
+- Replaces conditional `think.callosum` import with local logging
+- Defines SAMPLE_RATE locally (was from observe.utils)
+"""
+
+from __future__ import annotations
+
+import gc
+import io
+import logging
+import os
+import signal
+import threading
+import time
+from queue import Queue
+
+import numpy as np
+import soundfile as sf
+
+logger = logging.getLogger(__name__)
+
+# Standard sample rate for audio processing
+SAMPLE_RATE = 16000
+BLOCK_SIZE = 1024
+
+
+class AudioRecorder:
+    """Records stereo audio from microphone and system audio."""
+
+    def __init__(self):
+        # Queue holds stereo chunks (mic=left, sys=right)
+        self.audio_queue = Queue()
+        self._running = True
+        self.recording_thread = None
+
+    def detect(self):
+        """Detect microphone and system audio devices."""
+        from .audio_detect import input_detect
+
+        mic, loopback = input_detect()
+        if mic is None or loopback is None:
+            logger.error(f"Detection failed: mic {mic} sys {loopback}")
+            return False
+        logger.info(f"Detected microphone: {mic.name}")
+        logger.info(f"Detected system audio: {loopback.name}")
+        self.mic_device = mic
+        self.sys_device = loopback
+        return True
+
+    def record_both(self):
+        """Record from both mic and system audio in a loop."""
+        while self._running:
+            try:
+                with (
+                    self.mic_device.recorder(
+                        samplerate=SAMPLE_RATE, channels=[-1], blocksize=BLOCK_SIZE
+                    ) as mic_rec,
+                    self.sys_device.recorder(
+                        samplerate=SAMPLE_RATE, channels=[-1], blocksize=BLOCK_SIZE
+                    ) as sys_rec,
+                ):
+                    block_count = 0
+                    while self._running and block_count < 1000:
+                        try:
+                            mic_chunk = mic_rec.record(numframes=BLOCK_SIZE)
+                            sys_chunk = sys_rec.record(numframes=BLOCK_SIZE)
+
+                            # Basic validation
+                            if mic_chunk is None or mic_chunk.size == 0:
+                                logger.warning("Empty microphone buffer")
+                                continue
+                            if sys_chunk is None or sys_chunk.size == 0:
+                                logger.warning("Empty system buffer")
+                                continue
+
+                            try:
+                                stereo_chunk = np.column_stack((mic_chunk, sys_chunk))
+                                self.audio_queue.put(stereo_chunk)
+                                block_count += 1
+                            except (TypeError, ValueError, AttributeError) as e:
+                                error_msg = f"Fatal audio format error: {e}"
+                                logger.error(
+                                    f"{error_msg} - triggering clean shutdown\n"
+                                    f"  mic_chunk type={type(mic_chunk)}, "
+                                    f"shape={getattr(mic_chunk, 'shape', 'N/A')}, "
+                                    f"dtype={getattr(mic_chunk, 'dtype', 'N/A')}\n"
+                                    f"  sys_chunk type={type(sys_chunk)}, "
+                                    f"shape={getattr(sys_chunk, 'shape', 'N/A')}, "
+                                    f"dtype={getattr(sys_chunk, 'dtype', 'N/A')}"
+                                )
+                                # Stop recording thread and trigger shutdown
+                                self._running = False
+                                os.kill(os.getpid(), signal.SIGTERM)
+                                return
+                        except Exception as e:
+                            logger.error(f"Error recording audio: {e}")
+                            if not self._running:
+                                break
+                            time.sleep(0.5)
+                del mic_rec, sys_rec
+                gc.collect()
+            except Exception as e:
+                logger.error(f"Error setting up recorders: {e}")
+                if self._running:
+                    time.sleep(1)
+
+    def get_buffers(self) -> np.ndarray:
+        """Return concatenated stereo audio data from the queue."""
+        stereo_buffer = np.array([], dtype=np.float32).reshape(0, 2)
+
+        while not self.audio_queue.empty():
+            stereo_chunk = self.audio_queue.get()
+
+            if stereo_chunk is None or stereo_chunk.size == 0:
+                logger.warning("Queue contained empty chunk")
+                continue
+
+            # Clean the data
+            stereo_chunk = np.nan_to_num(
+                stereo_chunk, nan=0.0, posinf=1e10, neginf=-1e10
+            )
+            stereo_buffer = np.vstack((stereo_buffer, stereo_chunk))
+
+        if stereo_buffer.size == 0:
+            logger.warning("No valid audio data retrieved from queue")
+
+        return stereo_buffer
+
+    def create_flac_bytes(self, stereo_data: np.ndarray) -> bytes:
+        """Create FLAC bytes from stereo audio data."""
+        if stereo_data is None or stereo_data.size == 0:
+            logger.warning("Audio data is empty. Returning empty bytes.")
+            return b""
+
+        audio_data = (np.clip(stereo_data, -1.0, 1.0) * 32767).astype(np.int16)
+
+        buf = io.BytesIO()
+        try:
+            sf.write(buf, audio_data, SAMPLE_RATE, format="FLAC")
+        except Exception as e:
+            logger.error(
+                f"Error creating FLAC: {e}. Audio data shape: {audio_data.shape}, dtype: {audio_data.dtype}"
+            )
+            return b""
+
+        return buf.getvalue()
+
+    def create_mono_flac_bytes(self, mono_data: np.ndarray) -> bytes:
+        """Create FLAC bytes from mono audio data."""
+        if mono_data is None or mono_data.size == 0:
+            logger.warning("Mono audio data is empty. Returning empty bytes.")
+            return b""
+
+        audio_data = (np.clip(mono_data, -1.0, 1.0) * 32767).astype(np.int16)
+
+        buf = io.BytesIO()
+        try:
+            sf.write(buf, audio_data, SAMPLE_RATE, format="FLAC")
+        except Exception as e:
+            logger.error(
+                f"Error creating mono FLAC: {e}. Audio shape: {audio_data.shape}"
+            )
+            return b""
+
+        return buf.getvalue()
+
+    def start_recording(self):
+        """Start the recording thread."""
+        self._running = True
+        self.recording_thread = threading.Thread(target=self.record_both, daemon=True)
+        self.recording_thread.start()
+
+    def stop_recording(self):
+        """Stop the recording thread."""
+        self._running = False
+        if self.recording_thread:
+            self.recording_thread.join(timeout=2.0)

src/solstone_linux/cli.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""CLI entry point for solstone-linux.
+
+Subcommands:
+    run             Start capture loop + sync service (default)
+    setup           Interactive configuration
+    install-service Write systemd user unit, enable, start
+    status          Show capture and sync state
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+import shutil
+import socket
+import subprocess
+import sys
+from pathlib import Path
+
+from .config import Config, load_config, save_config
+from .streams import stream_name
+
+
+def _setup_logging(verbose: bool = False) -> None:
+    level = logging.DEBUG if verbose else logging.INFO
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        datefmt="%H:%M:%S",
+    )
+
+
+def cmd_run(args: argparse.Namespace) -> int:
+    """Start the capture loop + sync service."""
+    from .observer import async_run
+    from .recovery import recover_incomplete_segments
+
+    config = load_config()
+    config.ensure_dirs()
+
+    if not config.stream:
+        try:
+            config.stream = stream_name(host=socket.gethostname())
+        except ValueError as e:
+            print(f"Error: {e}", file=sys.stderr)
+            return 1
+
+    if args.interval:
+        config.segment_interval = args.interval
+
+    # Crash recovery before starting
+    recovered = recover_incomplete_segments(config.captures_dir)
+    if recovered:
+        print(f"Recovered {recovered} incomplete segment(s)")
+
+    try:
+        return asyncio.run(async_run(config))
+    except KeyboardInterrupt:
+        return 0
+
+
+def cmd_setup(args: argparse.Namespace) -> int:
+    """Interactive setup — configure server URL and register."""
+    from .upload import UploadClient
+
+    config = load_config()
+
+    # Prompt for server URL
+    default_url = config.server_url or ""
+    url = input(f"Solstone server URL [{default_url}]: ").strip()
+    if url:
+        config.server_url = url
+    elif not config.server_url:
+        print("Error: server URL is required", file=sys.stderr)
+        return 1
+
+    # Derive stream name
+    if not config.stream:
+        try:
+            config.stream = stream_name(host=socket.gethostname())
+        except ValueError as e:
+            print(f"Error deriving stream name: {e}", file=sys.stderr)
+            return 1
+    print(f"Stream: {config.stream}")
+
+    # Save config before registration (so URL is persisted)
+    config.ensure_dirs()
+    save_config(config)
+
+    # Auto-register
+    if not config.key:
+        print("Registering with server...")
+        client = UploadClient(config)
+        if client.ensure_registered(config):
+            config = load_config()
+            print(f"Registered (key: {config.key[:8]}...)")
+        else:
+            print("Warning: registration failed. Run setup again when server is available.")
+    else:
+        print(f"Already registered (key: {config.key[:8]}...)")
+
+    print(f"\nConfig saved to {config.config_path}")
+    print(f"Captures will go to {config.captures_dir}")
+    print(f"\nRun 'solstone-linux run' to start, or 'solstone-linux install-service' for systemd.")
+    return 0
+
+
+def cmd_install_service(args: argparse.Namespace) -> int:
+    """Write systemd user unit file, enable, and start the service."""
+    binary = shutil.which("solstone-linux")
+    if not binary:
+        print("Error: solstone-linux not found on PATH", file=sys.stderr)
+        print("Install with: pipx install --system-site-packages solstone-linux", file=sys.stderr)
+        return 1
+
+    unit_dir = Path.home() / ".config" / "systemd" / "user"
+    unit_dir.mkdir(parents=True, exist_ok=True)
+    unit_path = unit_dir / "solstone-linux.service"
+
+    unit_content = f"""\
+[Unit]
+Description=Solstone Linux Desktop Observer
+After=graphical-session.target
+BindsTo=graphical-session.target
+
+[Service]
+Type=simple
+ExecStart={binary} run
+PassEnvironment=DISPLAY WAYLAND_DISPLAY DBUS_SESSION_BUS_ADDRESS XDG_RUNTIME_DIR
+Restart=on-failure
+RestartSec=10
+StartLimitIntervalSec=300
+StartLimitBurst=5
+
+[Install]
+WantedBy=graphical-session.target
+"""
+
+    unit_path.write_text(unit_content)
+    print(f"Wrote {unit_path}")
+
+    # Reload, enable, start
+    try:
+        subprocess.run(["systemctl", "--user", "daemon-reload"], check=True)
+        subprocess.run(
+            ["systemctl", "--user", "enable", "--now", "solstone-linux.service"],
+            check=True,
+        )
+        print("Service enabled and started.")
+        subprocess.run(
+            ["systemctl", "--user", "status", "solstone-linux.service"],
+            check=False,
+        )
+    except FileNotFoundError:
+        print("Warning: systemctl not found. Enable the service manually.")
+    except subprocess.CalledProcessError as e:
+        print(f"Warning: systemctl command failed: {e}")
+
+    return 0
+
+
+def cmd_status(args: argparse.Namespace) -> int:
+    """Show capture and sync state."""
+    config = load_config()
+
+    print(f"Config: {config.config_path}")
+    print(f"Server: {config.server_url or '(not configured)'}")
+    print(f"Key:    {config.key[:8] + '...' if config.key else '(not registered)'}")
+    print(f"Stream: {config.stream or '(not set)'}")
+    print()
+
+    # Cache size
+    captures_dir = config.captures_dir
+    if captures_dir.exists():
+        total_size = 0
+        segment_count = 0
+        day_count = 0
+        incomplete_count = 0
+
+        for day_dir in sorted(captures_dir.iterdir()):
+            if not day_dir.is_dir():
+                continue
+            day_count += 1
+            for stream_dir in day_dir.iterdir():
+                if not stream_dir.is_dir():
+                    continue
+                for seg_dir in stream_dir.iterdir():
+                    if not seg_dir.is_dir():
+                        continue
+                    if seg_dir.name.endswith(".incomplete"):
+                        incomplete_count += 1
+                        continue
+                    if seg_dir.name.endswith(".failed"):
+                        continue
+                    segment_count += 1
+                    for f in seg_dir.iterdir():
+                        if f.is_file():
+                            total_size += f.stat().st_size
+
+        size_mb = total_size / (1024 * 1024)
+        print(f"Cache:  {captures_dir}")
+        print(f"        {segment_count} segments across {day_count} day(s), {size_mb:.1f} MB")
+        if incomplete_count:
+            print(f"        {incomplete_count} incomplete segment(s)")
+    else:
+        print(f"Cache:  {captures_dir} (not created yet)")
+
+    # Synced days
+    synced_path = config.state_dir / "synced_days.json"
+    if synced_path.exists():
+        try:
+            with open(synced_path) as f:
+                synced = json.load(f)
+            print(f"Synced: {len(synced)} day(s) fully synced")
+        except (json.JSONDecodeError, OSError):
+            pass
+
+    # Systemd status
+    try:
+        result = subprocess.run(
+            ["systemctl", "--user", "is-active", "solstone-linux.service"],
+            capture_output=True,
+            text=True,
+        )
+        state = result.stdout.strip()
+        print(f"\nService: {state}")
+    except FileNotFoundError:
+        pass
+
+    return 0
+
+
+def main() -> None:
+    """CLI entry point."""
+    parser = argparse.ArgumentParser(
+        prog="solstone-linux",
+        description="Standalone Linux desktop observer for solstone",
+    )
+    parser.add_argument(
+        "-v", "--verbose", action="store_true", help="Enable debug logging"
+    )
+    subparsers = parser.add_subparsers(dest="command")
+
+    # run
+    run_parser = subparsers.add_parser("run", help="Start capture + sync")
+    run_parser.add_argument(
+        "--interval",
+        type=int,
+        default=None,
+        help="Segment duration in seconds (default: 300)",
+    )
+
+    # setup
+    subparsers.add_parser("setup", help="Interactive configuration")
+
+    # install-service
+    subparsers.add_parser("install-service", help="Install systemd user service")
+
+    # status
+    subparsers.add_parser("status", help="Show capture and sync state")
+
+    args = parser.parse_args()
+    _setup_logging(args.verbose)
+
+    # Default to run if no subcommand
+    command = args.command or "run"
+
+    commands = {
+        "run": cmd_run,
+        "setup": cmd_setup,
+        "install-service": cmd_install_service,
+        "status": cmd_status,
+    }
+
+    handler = commands.get(command)
+    if handler:
+        sys.exit(handler(args))
+    else:
+        parser.print_help()
+        sys.exit(1)

src/solstone_linux/config.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Configuration loading and persistence for solstone-linux.
+
+Config lives at ~/.local/share/solstone-linux/config/config.json.
+Captures go to ~/.local/share/solstone-linux/captures/.
+Screencast restore token at ~/.local/share/solstone-linux/config/restore_token.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import stat
+from dataclasses import dataclass, field
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_BASE_DIR = Path.home() / ".local" / "share" / "solstone-linux"
+DEFAULT_SEGMENT_INTERVAL = 300
+DEFAULT_SYNC_RETRY_DELAYS = [5, 30, 120, 300]
+DEFAULT_SYNC_MAX_RETRIES = 10
+
+
+@dataclass
+class Config:
+    """Configuration for the Linux desktop observer."""
+
+    server_url: str = ""
+    key: str = ""
+    stream: str = ""
+    segment_interval: int = DEFAULT_SEGMENT_INTERVAL
+    sync_retry_delays: list[int] = field(default_factory=lambda: list(DEFAULT_SYNC_RETRY_DELAYS))
+    sync_max_retries: int = DEFAULT_SYNC_MAX_RETRIES
+    base_dir: Path = DEFAULT_BASE_DIR
+
+    @property
+    def captures_dir(self) -> Path:
+        return self.base_dir / "captures"
+
+    @property
+    def config_dir(self) -> Path:
+        return self.base_dir / "config"
+
+    @property
+    def state_dir(self) -> Path:
+        return self.base_dir / "state"
+
+    @property
+    def config_path(self) -> Path:
+        return self.config_dir / "config.json"
+
+    @property
+    def restore_token_path(self) -> Path:
+        return self.config_dir / "restore_token"
+
+    def ensure_dirs(self) -> None:
+        """Create all required directories."""
+        self.captures_dir.mkdir(parents=True, exist_ok=True)
+        self.config_dir.mkdir(parents=True, exist_ok=True)
+        self.state_dir.mkdir(parents=True, exist_ok=True)
+
+
+def load_config(base_dir: Path | None = None) -> Config:
+    """Load config from disk, returning defaults if not found."""
+    config = Config()
+    if base_dir:
+        config.base_dir = base_dir
+
+    config_path = config.config_path
+    if not config_path.exists():
+        return config
+
+    try:
+        with open(config_path, encoding="utf-8") as f:
+            data = json.load(f)
+    except (json.JSONDecodeError, OSError) as e:
+        logger.warning(f"Failed to load config from {config_path}: {e}")
+        return config
+
+    config.server_url = data.get("server_url", "")
+    config.key = data.get("key", "")
+    config.stream = data.get("stream", "")
+    config.segment_interval = data.get("segment_interval", DEFAULT_SEGMENT_INTERVAL)
+    if "sync_retry_delays" in data:
+        config.sync_retry_delays = data["sync_retry_delays"]
+    if "sync_max_retries" in data:
+        config.sync_max_retries = data["sync_max_retries"]
+
+    return config
+
+
+def save_config(config: Config) -> None:
+    """Save config to disk with user-only permissions."""
+    config.ensure_dirs()
+
+    data = {
+        "server_url": config.server_url,
+        "key": config.key,
+        "stream": config.stream,
+        "segment_interval": config.segment_interval,
+        "sync_retry_delays": config.sync_retry_delays,
+        "sync_max_retries": config.sync_max_retries,
+    }
+
+    config_path = config.config_path
+    tmp_path = config_path.with_suffix(f".{os.getpid()}.tmp")
+
+    with open(tmp_path, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2)
+        f.write("\n")
+
+    # Set user-only read/write before moving into place
+    os.chmod(tmp_path, stat.S_IRUSR | stat.S_IWUSR)
+    os.rename(str(tmp_path), str(config_path))
+    logger.info(f"Config saved to {config_path}")

src/solstone_linux/monitor_positions.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Monitor position assignment based on geometry.
+
+Extracted from solstone's observe/utils.py — the assign_monitor_positions()
+function only. Also remains in solstone core (used by server-side naming).
+"""
+
+from __future__ import annotations
+
+
+def assign_monitor_positions(monitors: list[dict]) -> list[dict]:
+    """
+    Assign position labels to monitors based on relative positions.
+
+    Uses pairwise comparison to determine positions. Vertical labels (top/bottom)
+    are only assigned when monitors actually overlap horizontally, avoiding
+    phantom relationships from offset monitors.
+
+    Parameters
+    ----------
+    monitors : list[dict]
+        List of monitor dicts, each with keys:
+        - id: Monitor identifier (e.g., "DP-3", "HDMI-1")
+        - box: [x1, y1, x2, y2] coordinates
+
+    Returns
+    -------
+    list[dict]
+        Same monitors with "position" key added to each:
+        - "center": No monitors on both sides
+        - "left"/"right": Horizontal position
+        - "top"/"bottom": Vertical position (only with horizontal overlap)
+        - "left-top", "right-bottom", etc.: Corner positions
+    """
+    if not monitors:
+        return []
+
+    if len(monitors) == 1:
+        monitors[0]["position"] = "center"
+        return monitors
+
+    # Tolerance for center classification
+    epsilon = 1
+
+    for m in monitors:
+        x1, y1, x2, y2 = m["box"]
+        center_x = (x1 + x2) / 2
+        center_y = (y1 + y2) / 2
+
+        has_left = False
+        has_right = False
+        has_above = False
+        has_below = False
+
+        for other in monitors:
+            if other is m:
+                continue
+
+            ox1, oy1, ox2, oy2 = other["box"]
+            other_center_x = (ox1 + ox2) / 2
+            other_center_y = (oy1 + oy2) / 2
+
+            # Horizontal relationship (always check)
+            if other_center_x < center_x - epsilon:
+                has_left = True
+            elif other_center_x > center_x + epsilon:
+                has_right = True
+
+            # Vertical relationship only if horizontal overlap exists
+            # Overlap means ranges intersect (not just touch)
+            h_overlap = (x1 < ox2) and (x2 > ox1)
+            if h_overlap:
+                if other_center_y < center_y - epsilon:
+                    has_above = True
+                elif other_center_y > center_y + epsilon:
+                    has_below = True
+
+        # Determine horizontal label
+        if has_left and has_right:
+            h_pos = "center"
+        elif has_left:
+            h_pos = "right"
+        elif has_right:
+            h_pos = "left"
+        else:
+            h_pos = "center"
+
+        # Determine vertical label (only if monitors above/below with overlap)
+        if has_above and has_below:
+            v_pos = "middle"
+        elif has_above:
+            v_pos = "bottom"
+        elif has_below:
+            v_pos = "top"
+        else:
+            v_pos = None
+
+        # Combine positions
+        if v_pos is None:
+            position = h_pos
+        elif h_pos == "center":
+            position = v_pos
+        else:
+            position = f"{h_pos}-{v_pos}"
+
+        m["position"] = position
+
+    return monitors

src/solstone_linux/observer.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""
+Standalone Linux desktop observer — screen + audio capture.
+
+Continuously captures audio and manages screencast recording based on activity.
+Creates 5-minute segments in a local cache directory. The sync service handles
+all uploads — the observer only writes locally.
+
+Key architectural change from monorepo version:
+- Capture writes completed segments to local cache only
+- No ObserverClient usage in boundary handling — no network calls in capture loop
+- Sync service picks up completed segments and uploads asynchronously
+
+State machine:
+    SCREENCAST: Screen is active, recording video
+    IDLE: Screen is inactive
+"""
+
+import asyncio
+import datetime
+import logging
+import os
+import platform
+import signal
+import socket
+import time
+from pathlib import Path
+
+import numpy as np
+from dbus_next.aio import MessageBus
+from dbus_next.constants import BusType
+
+from .activity import get_idle_time_ms, is_power_save_active, is_screen_locked
+from .audio_mute import is_sink_muted
+from .audio_recorder import AudioRecorder
+from .config import Config
+from .recovery import write_segment_metadata
+from .screencast import Screencaster, StreamInfo
+from .streams import stream_name
+from .sync import SyncService
+from .upload import UploadClient
+
+logger = logging.getLogger(__name__)
+
+# Host identification
+HOST = socket.gethostname()
+PLATFORM = platform.system().lower()
+
+# Constants
+IDLE_THRESHOLD_MS = 5 * 60 * 1000  # 5 minutes
+RMS_THRESHOLD = 0.01
+MIN_HITS_FOR_SAVE = 3
+CHUNK_DURATION = 5  # seconds
+
+# Capture modes
+MODE_IDLE = "idle"
+MODE_SCREENCAST = "screencast"
+
+# Audio detection retry
+DETECT_RETRIES = 3
+DETECT_RETRY_DELAY = 5  # seconds
+
+
+def _get_timestamp_parts(timestamp: float | None = None) -> tuple[str, str]:
+    """Get date and time parts from timestamp."""
+    if timestamp is None:
+        timestamp = time.time()
+    dt = datetime.datetime.fromtimestamp(timestamp)
+    return dt.strftime("%Y%m%d"), dt.strftime("%H%M%S")
+
+
+class Observer:
+    """Unified audio and screencast observer with local cache + sync."""
+
+    def __init__(self, config: Config):
+        self.config = config
+        self.interval = config.segment_interval
+        self.audio_recorder = AudioRecorder()
+        self.screencaster = Screencaster(config.restore_token_path)
+        self.bus: MessageBus | None = None
+        self.running = True
+        self.stream = config.stream
+
+        self._client: UploadClient | None = None
+        self._sync: SyncService | None = None
+
+        # State tracking
+        self.start_at = time.time()
+        self.start_at_mono = time.monotonic()
+        self.threshold_hits = 0
+        self.accumulated_audio_buffer = np.array([], dtype=np.float32).reshape(0, 2)
+
+        # Mode tracking
+        self.current_mode = MODE_IDLE
+
+        # Segment directory (HHMMSS.incomplete/)
+        self.segment_dir: Path | None = None
+
+        # Multi-file screencast tracking
+        self.current_streams: list[StreamInfo] = []
+
+        # Activity status cache (updated each loop)
+        self.cached_is_active = False
+        self.cached_idle_time_ms = 0
+        self.cached_screen_locked = False
+        self.cached_is_muted = False
+        self.cached_power_save = False
+
+        # Mute state at segment start (determines save format)
+        self.segment_is_muted = False
+
+    async def setup(self) -> bool:
+        """Initialize audio devices, DBus connection, and sync service."""
+        # Detect audio devices with retry (devices may still be initializing)
+        detected = False
+        for attempt in range(DETECT_RETRIES):
+            if self.audio_recorder.detect():
+                detected = True
+                break
+            if attempt < DETECT_RETRIES - 1:
+                logger.info(
+                    "Audio detection attempt %d/%d failed, retrying in %ds",
+                    attempt + 1,
+                    DETECT_RETRIES,
+                    DETECT_RETRY_DELAY,
+                )
+                await asyncio.sleep(DETECT_RETRY_DELAY)
+        if not detected:
+            logger.error("Failed to detect audio devices")
+            return False
+
+        self.audio_recorder.start_recording()
+        logger.info("Audio recording started")
+
+        # Connect to DBus for idle/lock detection
+        self.bus = await MessageBus(bus_type=BusType.SESSION).connect()
+        logger.info("DBus connection established")
+
+        # Verify portal is available (exit if not)
+        if not await self.screencaster.connect():
+            logger.error("Screencast portal not available")
+            return False
+        logger.info("Screencast portal connected")
+
+        # Initialize upload client and sync service
+        self._client = UploadClient(self.config)
+        if self.config.server_url:
+            self._client.ensure_registered(self.config)
+        self._sync = SyncService(self.config, self._client)
+        logger.info("Sync service initialized")
+
+        return True
+
+    async def check_activity_status(self) -> str:
+        """Check system activity status and determine capture mode."""
+        idle_time = await get_idle_time_ms(self.bus)
+        screen_locked = await is_screen_locked(self.bus)
+        power_save = await is_power_save_active(self.bus)
+        sink_muted = await is_sink_muted()
+
+        # Cache values for status events
+        self.cached_idle_time_ms = idle_time
+        self.cached_screen_locked = screen_locked
+        self.cached_is_muted = sink_muted
+        self.cached_power_save = power_save
+
+        # Determine screen activity
+        screen_idle = (idle_time > IDLE_THRESHOLD_MS) or screen_locked or power_save
+        screen_active = not screen_idle
+
+        # Determine mode
+        if screen_active:
+            mode = MODE_SCREENCAST
+        else:
+            mode = MODE_IDLE
+
+        # Cache legacy is_active for audio threshold logic
+        has_audio_activity = self.threshold_hits >= MIN_HITS_FOR_SAVE
+        self.cached_is_active = screen_active or has_audio_activity
+
+        return mode
+
+    def compute_rms(self, audio_buffer: np.ndarray) -> float:
+        """Compute per-channel RMS and return maximum (stereo: mic=left, sys=right)."""
+        if audio_buffer.size == 0:
+            return 0.0
+        rms_left = float(np.sqrt(np.mean(audio_buffer[:, 0] ** 2)))
+        rms_right = float(np.sqrt(np.mean(audio_buffer[:, 1] ** 2)))
+        return max(rms_left, rms_right)
+
+    def _save_audio_segment(self, segment_dir: Path, is_muted: bool) -> list[str]:
+        """Save accumulated audio buffer to segment directory."""
+        if self.accumulated_audio_buffer.size == 0:
+            logger.warning("No audio buffer to save")
+            return []
+
+        if is_muted:
+            # Split mode: save mic and sys as separate mono files
+            mic_data = self.accumulated_audio_buffer[:, 0]
+            sys_data = self.accumulated_audio_buffer[:, 1]
+
+            mic_bytes = self.audio_recorder.create_mono_flac_bytes(mic_data)
+            sys_bytes = self.audio_recorder.create_mono_flac_bytes(sys_data)
+
+            (segment_dir / "mic_audio.flac").write_bytes(mic_bytes)
+            (segment_dir / "sys_audio.flac").write_bytes(sys_bytes)
+
+            logger.info(f"Saved split audio (muted): {segment_dir}")
+            return ["mic_audio.flac", "sys_audio.flac"]
+        else:
+            # Normal mode: save combined stereo file
+            flac_bytes = self.audio_recorder.create_flac_bytes(
+                self.accumulated_audio_buffer
+            )
+            (segment_dir / "audio.flac").write_bytes(flac_bytes)
+
+            logger.info(f"Saved audio to {segment_dir}/audio.flac")
+            return ["audio.flac"]
+
+    def _start_segment(self) -> Path:
+        """Start a new segment with .incomplete directory."""
+        self.start_at = time.time()
+        self.start_at_mono = time.monotonic()
+
+        date_part, time_part = _get_timestamp_parts(self.start_at)
+        captures_dir = self.config.captures_dir
+
+        # Create YYYYMMDD/stream/HHMMSS.incomplete/
+        segment_dir = captures_dir / date_part / self.stream / f"{time_part}.incomplete"
+        segment_dir.mkdir(parents=True, exist_ok=True)
+        self.segment_dir = segment_dir
+
+        # Write metadata for recovery
+        write_segment_metadata(segment_dir, self.start_at)
+
+        return segment_dir
+
+    def _finalize_segment(self) -> str | None:
+        """Rename .incomplete to HHMMSS_DDD/ and return segment key."""
+        if not self.segment_dir or not self.segment_dir.exists():
+            return None
+
+        # Remove .metadata before finalizing
+        meta_path = self.segment_dir / ".metadata"
+        if meta_path.exists():
+            try:
+                meta_path.unlink()
+            except OSError:
+                pass
+
+        # Check if there are any actual files
+        contents = [f for f in self.segment_dir.iterdir() if f.is_file()]
+        if not contents:
+            # Empty segment, remove it
+            try:
+                os.rmdir(str(self.segment_dir))
+            except OSError:
+                pass
+            return None
+
+        _, time_part = _get_timestamp_parts(self.start_at)
+        duration = int(time.time() - self.start_at)
+        segment_key = f"{time_part}_{duration}"
+        final_dir = self.segment_dir.parent / segment_key
+
+        try:
+            os.rename(str(self.segment_dir), str(final_dir))
+            logger.info(f"Segment finalized: {segment_key}")
+            return segment_key
+        except OSError as e:
+            logger.error(f"Failed to finalize segment: {e}")
+            return None
+
+    async def handle_boundary(self, new_mode: str):
+        """Handle window boundary rollover.
+
+        Closes the current segment, writes audio, finalizes to local cache,
+        and triggers sync. No network calls in the capture loop.
+        """
+        # Stop screencast first (closes file handles)
+        if self.current_mode == MODE_SCREENCAST:
+            logger.info("Stopping previous screencast")
+            await self.screencaster.stop()
+            self.current_streams = []
+
+        # Save audio if we have enough threshold hits
+        did_save_audio = self.threshold_hits >= MIN_HITS_FOR_SAVE
+        if did_save_audio and self.segment_dir:
+            audio_files = self._save_audio_segment(
+                self.segment_dir, self.segment_is_muted
+            )
+            if audio_files:
+                logger.info(
+                    f"Saved {len(audio_files)} audio file(s) ({self.threshold_hits} hits)"
+                )
+        else:
+            logger.debug(
+                f"Skipping audio save (only {self.threshold_hits}/{MIN_HITS_FOR_SAVE} hits)"
+            )
+
+        # Reset audio state
+        self.accumulated_audio_buffer = np.array([], dtype=np.float32).reshape(0, 2)
+        self.threshold_hits = 0
+
+        # Finalize segment (rename .incomplete -> HHMMSS_DDD/)
+        segment_key = self._finalize_segment()
+        self.segment_dir = None
+
+        # Trigger sync to upload the completed segment
+        if segment_key and self._sync:
+            self._sync.trigger()
+
+        # Update segment mute state for new segment
+        self.segment_is_muted = self.cached_is_muted
+
+        # Update mode
+        old_mode = self.current_mode
+        self.current_mode = new_mode
+
+        # Start new capture based on mode
+        if new_mode == MODE_SCREENCAST and not self.cached_screen_locked:
+            await self.initialize_screencast()
+        else:
+            self._start_segment()
+
+        logger.info(f"Mode transition: {old_mode} -> {new_mode}")
+
+    async def initialize_screencast(self) -> bool:
+        """Start a new screencast recording.
+
+        Creates a segment directory and starts GStreamer recording to it.
+        """
+        segment_dir = self._start_segment()
+
+        try:
+            streams = await self.screencaster.start(
+                str(segment_dir), framerate=1, draw_cursor=True
+            )
+        except RuntimeError as e:
+            logger.error(f"Failed to start screencast: {e}")
+            raise
+
+        if not streams:
+            logger.error("No streams returned from screencast start")
+            raise RuntimeError("No streams available")
+
+        self.current_streams = streams
+
+        logger.info(f"Started screencast with {len(streams)} stream(s)")
+        for stream in streams:
+            logger.info(f"  {stream.position} ({stream.connector}): {stream.file_path}")
+
+        return True
+
+    def emit_status(self):
+        """Emit observe.status event with current state (fire-and-forget)."""
+        if not self._client:
+            return
+
+        elapsed = int(time.monotonic() - self.start_at_mono)
+
+        # Screencast info
+        if self.current_mode == MODE_SCREENCAST and self.current_streams:
+            streams_info = [
+                {
+                    "position": stream.position,
+                    "connector": stream.connector,
+                    "file": stream.file_path,
+                }
+                for stream in self.current_streams
+            ]
+            screencast_info = {
+                "recording": True,
+                "streams": streams_info,
+                "window_elapsed_seconds": elapsed,
+            }
+        else:
+            screencast_info = {"recording": False}
+
+        # Audio info
+        audio_info = {
+            "threshold_hits": self.threshold_hits,
+            "will_save": self.threshold_hits >= MIN_HITS_FOR_SAVE,
+        }
+
+        # Activity info
+        activity_info = {
+            "active": self.cached_is_active,
+            "idle_time_ms": self.cached_idle_time_ms,
+            "screen_locked": self.cached_screen_locked,
+            "sink_muted": self.cached_is_muted,
+            "power_save": self.cached_power_save,
+        }
+
+        self._client.relay_event(
+            "observe",
+            "status",
+            mode=self.current_mode,
+            screencast=screencast_info,
+            audio=audio_info,
+            activity=activity_info,
+            host=HOST,
+            platform=PLATFORM,
+            stream=self.stream,
+        )
+
+    async def main_loop(self):
+        """Run the main observer loop with background sync."""
+        logger.info(f"Starting observer loop (interval={self.interval}s)")
+
+        # Start sync service as background task
+        sync_task = None
+        if self._sync:
+            sync_task = asyncio.create_task(self._sync.run())
+
+        # Determine initial mode
+        new_mode = await self.check_activity_status()
+        self.segment_is_muted = self.cached_is_muted
+        self.current_mode = new_mode
+
+        # Start initial capture based on mode
+        if new_mode == MODE_SCREENCAST and not self.cached_screen_locked:
+            try:
+                await self.initialize_screencast()
+            except RuntimeError:
+                self.running = False
+                if sync_task:
+                    if self._sync:
+                        self._sync.stop()
+                    sync_task.cancel()
+                    try:
+                        await sync_task
+                    except asyncio.CancelledError:
+                        pass
+                return
+        else:
+            self._start_segment()
+
+        logger.info(f"Initial mode: {self.current_mode}")
+
+        try:
+            while self.running:
+                await asyncio.sleep(CHUNK_DURATION)
+
+                # Check activity status and determine new mode
+                new_mode = await self.check_activity_status()
+
+                # Check for GStreamer failure mid-recording
+                if (
+                    self.current_mode == MODE_SCREENCAST
+                    and not self.screencaster.is_healthy()
+                ):
+                    logger.warning("Screencast recording failed, stopping gracefully")
+                    await self.screencaster.stop()
+                    self.current_streams = []
+                    self.current_mode = MODE_IDLE
+
+                # Detect mode change
+                mode_changed = new_mode != self.current_mode
+                if mode_changed:
+                    logger.info(f"Mode changing: {self.current_mode} -> {new_mode}")
+
+                # Only trigger segment boundary on screencast transitions
+                screencast_transition = mode_changed and (
+                    self.current_mode == MODE_SCREENCAST or new_mode == MODE_SCREENCAST
+                )
+
+                # Detect mute state transition
+                mute_transition = self.cached_is_muted != self.segment_is_muted
+                if mute_transition:
+                    logger.info(
+                        f"Mute state changed: "
+                        f"{'muted' if self.segment_is_muted else 'unmuted'} -> "
+                        f"{'muted' if self.cached_is_muted else 'unmuted'}"
+                    )
+
+                # Capture audio buffer for this chunk
+                audio_chunk = self.audio_recorder.get_buffers()
+
+                if audio_chunk.size > 0:
+                    self.accumulated_audio_buffer = np.vstack(
+                        (self.accumulated_audio_buffer, audio_chunk)
+                    )
+                    rms = self.compute_rms(audio_chunk)
+                    if rms > RMS_THRESHOLD:
+                        self.threshold_hits += 1
+                        logger.debug(
+                            f"RMS {rms:.4f} > threshold (hit {self.threshold_hits})"
+                        )
+                    else:
+                        logger.debug(f"RMS {rms:.4f} below threshold")
+                else:
+                    logger.debug("No audio data in chunk")
+
+                # Check for window boundary (monotonic to avoid DST/clock jumps)
+                elapsed = time.monotonic() - self.start_at_mono
+                is_boundary = (
+                    (elapsed >= self.interval) or screencast_transition or mute_transition
+                )
+
+                if is_boundary:
+                    logger.info(
+                        f"Boundary: elapsed={elapsed:.1f}s screencast_change={screencast_transition} "
+                        f"mute_change={mute_transition} "
+                        f"hits={self.threshold_hits}/{MIN_HITS_FOR_SAVE}"
+                    )
+                    await self.handle_boundary(new_mode)
+
+                # Emit status event
+                self.emit_status()
+        finally:
+            # Cleanup on exit
+            logger.info("Observer loop stopped, cleaning up...")
+            await self.shutdown()
+            if sync_task:
+                if self._sync:
+                    self._sync.stop()
+                sync_task.cancel()
+                try:
+                    await sync_task
+                except asyncio.CancelledError:
+                    pass
+
+    async def shutdown(self):
+        """Clean shutdown of observer."""
+        # Stop screencast first (closes file handles)
+        if self.current_mode == MODE_SCREENCAST:
+            logger.info("Stopping screencast for shutdown")
+            await self.screencaster.stop()
+            await asyncio.sleep(0.5)
+
+        # Save final audio if threshold met
+        if self.threshold_hits >= MIN_HITS_FOR_SAVE and self.segment_dir:
+            audio_files = self._save_audio_segment(
+                self.segment_dir, self.segment_is_muted
+            )
+            if audio_files:
+                logger.info(f"Saved final audio: {len(audio_files)} file(s)")
+
+        # Finalize segment locally
+        segment_key = self._finalize_segment()
+        self.segment_dir = None
+
+        if segment_key:
+            logger.info(f"Finalized segment locally: {segment_key} (shutdown)")
+
+        # Stop audio recorder
+        self.audio_recorder.stop_recording()
+        logger.info("Audio recording stopped")
+
+        if self._client:
+            self._client.stop()
+            self._client = None
+        logger.info("Client stopped")
+
+
+async def async_run(config: Config) -> int:
+    """Async entry point for the observer."""
+    from .session_env import check_session_ready
+
+    # Pre-flight: check session prerequisites
+    not_ready = check_session_ready()
+    if not_ready:
+        logger.warning("Session not ready: %s", not_ready)
+        return 75  # EXIT_TEMPFAIL
+
+    observer = Observer(config)
+
+    loop = asyncio.get_running_loop()
+
+    def signal_handler():
+        logger.info("Received shutdown signal")
+        observer.running = False
+
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        loop.add_signal_handler(sig, signal_handler)
+
+    if not await observer.setup():
+        logger.error("Observer setup failed")
+        return 1
+
+    try:
+        await observer.main_loop()
+    except RuntimeError as e:
+        logger.error(f"Observer runtime error: {e}")
+        return 1
+    except Exception as e:
+        logger.error(f"Observer error: {e}", exc_info=True)
+        return 1
+
+    return 0

src/solstone_linux/recovery.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Crash recovery for orphaned .incomplete segment directories.
+
+Modeled on solstone-macos's IncompleteSegmentRecovery.swift.
+Runs on startup before the capture loop begins.
+
+Improvement over tmux baseline: reads .metadata JSON file for accurate
+start timestamp instead of relying on brittle filesystem timestamps.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# Segments newer than this are assumed to be actively recording
+MINIMUM_AGE_SECONDS = 120  # 2 minutes
+
+METADATA_FILENAME = ".metadata"
+
+
+def write_segment_metadata(segment_dir: Path, start_timestamp: float) -> None:
+    """Write metadata file inside a segment directory.
+
+    Called when creating a new .incomplete segment so recovery can
+    use the actual start timestamp instead of filesystem timestamps.
+    """
+    meta_path = segment_dir / METADATA_FILENAME
+    try:
+        data = {"start_timestamp": start_timestamp}
+        with open(meta_path, "w", encoding="utf-8") as f:
+            json.dump(data, f)
+            f.write("\n")
+    except OSError as e:
+        logger.warning(f"Failed to write segment metadata: {e}")
+
+
+def _read_segment_metadata(segment_dir: Path) -> dict | None:
+    """Read metadata file from a segment directory."""
+    meta_path = segment_dir / METADATA_FILENAME
+    if not meta_path.exists():
+        return None
+    try:
+        with open(meta_path, encoding="utf-8") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, OSError):
+        return None
+
+
+def recover_incomplete_segments(captures_dir: Path) -> int:
+    """Scan captures dir for orphaned .incomplete directories and finalize them.
+
+    For each .incomplete directory older than 2 minutes:
+    - Read .metadata for start timestamp if available, else fall back to
+      filesystem timestamps (mtime - ctime)
+    - Rename to HHMMSS_DDD/ format
+    - If recovery fails, rename to HHMMSS.failed/ to prevent infinite retry
+
+    Returns the number of successfully recovered segments.
+    """
+    if not captures_dir.exists():
+        return 0
+
+    recovered = 0
+    now = time.time()
+
+    for day_dir in sorted(captures_dir.iterdir()):
+        if not day_dir.is_dir():
+            continue
+
+        for stream_dir in sorted(day_dir.iterdir()):
+            if not stream_dir.is_dir():
+                continue
+
+            for segment_dir in sorted(stream_dir.iterdir()):
+                if not segment_dir.is_dir():
+                    continue
+
+                dir_name = segment_dir.name
+                if not dir_name.endswith(".incomplete"):
+                    continue
+
+                # Check age
+                try:
+                    dir_stat = segment_dir.stat()
+                    age = now - dir_stat.st_mtime
+                    if age < MINIMUM_AGE_SECONDS:
+                        logger.debug(f"Skipping recent incomplete: {dir_name}")
+                        continue
+                except OSError:
+                    continue
+
+                logger.info(f"Recovering incomplete segment: {dir_name}")
+                if _recover_segment(segment_dir):
+                    recovered += 1
+
+    if recovered:
+        logger.info(f"Recovered {recovered} incomplete segment(s)")
+    return recovered
+
+
+def _recover_segment(segment_dir: Path) -> bool:
+    """Recover a single incomplete segment directory.
+
+    Returns True on success.
+    """
+    dir_name = segment_dir.name
+    time_prefix = dir_name.removesuffix(".incomplete")
+
+    # Try .metadata first for accurate duration
+    metadata = _read_segment_metadata(segment_dir)
+    if metadata and "start_timestamp" in metadata:
+        start_ts = metadata["start_timestamp"]
+        duration = max(1, int(time.time() - start_ts))
+    else:
+        # Fall back to filesystem timestamps
+        try:
+            st = segment_dir.stat()
+            duration = max(1, int(st.st_mtime - st.st_ctime))
+        except OSError:
+            return _mark_failed(segment_dir)
+
+    # Check there are actual files inside (ignore .metadata)
+    try:
+        contents = [
+            f for f in segment_dir.iterdir()
+            if f.name != METADATA_FILENAME
+        ]
+        if not contents:
+            logger.warning(f"Empty incomplete segment: {dir_name}")
+            return _mark_failed(segment_dir)
+    except OSError:
+        return _mark_failed(segment_dir)
+
+    # Build final segment key with duration
+    segment_key = f"{time_prefix}_{duration}"
+    final_dir = segment_dir.parent / segment_key
+
+    # Remove .metadata before finalizing (not a capture artifact)
+    meta_path = segment_dir / METADATA_FILENAME
+    if meta_path.exists():
+        try:
+            meta_path.unlink()
+        except OSError:
+            pass
+
+    try:
+        os.rename(str(segment_dir), str(final_dir))
+        logger.info(f"Recovered: {dir_name} -> {segment_key}")
+        return True
+    except OSError as e:
+        logger.warning(f"Failed to rename {dir_name}: {e}")
+        return _mark_failed(segment_dir)
+
+
+def _mark_failed(segment_dir: Path) -> bool:
+    """Rename from .incomplete to .failed to prevent infinite retry."""
+    dir_name = segment_dir.name
+    if not dir_name.endswith(".incomplete"):
+        return False
+
+    failed_name = dir_name.removesuffix(".incomplete") + ".failed"
+    failed_dir = segment_dir.parent / failed_name
+
+    try:
+        os.rename(str(segment_dir), str(failed_dir))
+        logger.warning(f"Marked as failed: {dir_name} -> {failed_name}")
+    except OSError as e:
+        logger.error(f"Failed to mark as failed: {e}")
+
+    return False

src/solstone_linux/screencast.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""
+Portal-based multi-monitor screencast recording.
+
+Uses xdg-desktop-portal ScreenCast API with PipeWire + GStreamer to record
+each monitor as a separate file. This replaces the old GNOME Shell D-Bus approach.
+
+Extracted from solstone's observe/linux/screencast.py.
+
+Changes from monorepo version:
+- Replaces `from think.utils import get_journal` with config-based restore token path
+- Replaces `from observe.gnome.activity import get_monitor_geometries` with local activity module
+
+Runtime deps:
+  - xdg-desktop-portal with org.freedesktop.portal.ScreenCast
+  - Portal backend: xdg-desktop-portal-gnome (or -kde, -wlr, etc.)
+  - PipeWire running
+  - GStreamer with PipeWire plugin: gst-launch-1.0 pipewiresrc
+"""
+
+import asyncio
+import logging
+import os
+import signal
+import subprocess
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+
+from dbus_next import Variant, introspection
+from dbus_next.aio import MessageBus
+from dbus_next.constants import BusType
+
+# Workaround for dbus-next issue #122: portal has properties with hyphens
+# (e.g., "power-saver-enabled") which violate strict D-Bus naming validation.
+introspection.assert_member_name_valid = lambda name: None
+
+logger = logging.getLogger(__name__)
+
+# Portal D-Bus constants
+PORTAL_BUS = "org.freedesktop.portal.Desktop"
+PORTAL_PATH = "/org/freedesktop/portal/desktop"
+SC_IFACE = "org.freedesktop.portal.ScreenCast"
+REQ_IFACE = "org.freedesktop.portal.Request"
+SESSION_IFACE = "org.freedesktop.portal.Session"
+
+
+@dataclass
+class StreamInfo:
+    """Information about a single monitor's recording stream."""
+
+    node_id: int
+    position: str
+    connector: str
+    x: int
+    y: int
+    width: int
+    height: int
+    file_path: str  # Final path in segment directory
+
+    @property
+    def filename(self) -> str:
+        """Return just the filename for event payloads."""
+        return os.path.basename(self.file_path)
+
+
+def _load_restore_token(token_path: Path) -> str | None:
+    """Load restore token from disk."""
+    try:
+        data = token_path.read_text(encoding="utf-8").strip()
+        return data or None
+    except (FileNotFoundError, OSError):
+        return None
+
+
+def _save_restore_token(token: str, token_path: Path) -> None:
+    """Save restore token to disk."""
+    try:
+        token_path.parent.mkdir(parents=True, exist_ok=True)
+        token_path.write_text(token.strip() + "\n", encoding="utf-8")
+        logger.debug(f"Saved restore token to {token_path}")
+    except OSError as e:
+        logger.warning(f"Failed to save restore token: {e}")
+
+
+def _make_request_handle(bus: MessageBus, token: str) -> str:
+    """Compute expected Request object path for a handle_token."""
+    sender = bus.unique_name.lstrip(":").replace(".", "_")
+    return f"/org/freedesktop/portal/desktop/request/{sender}/{token}"
+
+
+def _prepare_request_handler(bus: MessageBus, handle: str) -> asyncio.Future:
+    """Set up signal handler for Request::Response before calling portal method."""
+    loop = asyncio.get_running_loop()
+    fut: asyncio.Future = loop.create_future()
+
+    def _message_handler(msg):
+        if (
+            msg.message_type.name == "SIGNAL"
+            and msg.path == handle
+            and msg.interface == REQ_IFACE
+            and msg.member == "Response"
+        ):
+            response = msg.body[0]
+            results = msg.body[1] if len(msg.body) > 1 else {}
+            if not fut.done():
+                fut.set_result((int(response), results))
+            bus.remove_message_handler(_message_handler)
+
+    bus.add_message_handler(_message_handler)
+    return fut
+
+
+def _variant_or_value(val):
+    """Extract value from Variant if needed."""
+    if isinstance(val, Variant):
+        return val.value
+    return val
+
+
+def _match_streams_to_monitors(streams: list[dict], monitors: list[dict]) -> list[dict]:
+    """
+    Match portal stream geometries to GDK monitor info.
+
+    Portal streams have position (x, y) and size (width, height).
+    GDK monitors have connector IDs and box coordinates.
+
+    Returns streams augmented with connector and position labels.
+    """
+    matched = []
+
+    for stream in streams:
+        props = stream.get("props", {})
+
+        # Extract stream geometry from portal properties
+        stream_pos = _variant_or_value(props.get("position", (0, 0)))
+        stream_size = _variant_or_value(props.get("size", (0, 0)))
+
+        if isinstance(stream_pos, (tuple, list)) and len(stream_pos) >= 2:
+            sx, sy = int(stream_pos[0]), int(stream_pos[1])
+        else:
+            sx, sy = 0, 0
+
+        if isinstance(stream_size, (tuple, list)) and len(stream_size) >= 2:
+            sw, sh = int(stream_size[0]), int(stream_size[1])
+        else:
+            sw, sh = 0, 0
+
+        # Find matching monitor by geometry
+        best_match = None
+        best_overlap = 0
+
+        for monitor in monitors:
+            mx1, my1, mx2, my2 = monitor["box"]
+            mw, mh = mx2 - mx1, my2 - my1
+
+            # Check if geometries match (within tolerance for scaling)
+            if abs(sx - mx1) < 10 and abs(sy - my1) < 10:
+                overlap = min(sw, mw) * min(sh, mh)
+                if overlap > best_overlap:
+                    best_overlap = overlap
+                    best_match = monitor
+
+        if best_match:
+            stream["connector"] = best_match["id"]
+            stream["position_label"] = best_match.get("position", "unknown")
+            stream["x"] = best_match["box"][0]
+            stream["y"] = best_match["box"][1]
+            stream["width"] = best_match["box"][2] - best_match["box"][0]
+            stream["height"] = best_match["box"][3] - best_match["box"][1]
+        else:
+            # Fallback: use stream index as identifier
+            stream["connector"] = f"monitor-{stream['idx']}"
+            stream["position_label"] = "unknown"
+            stream["x"] = sx
+            stream["y"] = sy
+            stream["width"] = sw
+            stream["height"] = sh
+
+        matched.append(stream)
+
+    return matched
+
+
+class Screencaster:
+    """Portal-based multi-monitor screencast manager."""
+
+    def __init__(self, restore_token_path: Path):
+        self.bus: MessageBus | None = None
+        self.session_handle: str | None = None
+        self.pw_fd: int | None = None
+        self.gst_process: subprocess.Popen | None = None
+        self.streams: list[StreamInfo] = []
+        self._started = False
+        self._restore_token_path = restore_token_path
+
+    async def connect(self) -> bool:
+        """
+        Establish D-Bus connection and verify portal availability.
+
+        Returns:
+            True if portal is available, False otherwise.
+        """
+        if self.bus is not None:
+            return True
+
+        try:
+            self.bus = await MessageBus(
+                bus_type=BusType.SESSION,
+                negotiate_unix_fd=True,
+            ).connect()
+
+            # Verify portal interface exists
+            root_intro = await self.bus.introspect(PORTAL_BUS, PORTAL_PATH)
+            root_obj = self.bus.get_proxy_object(PORTAL_BUS, PORTAL_PATH, root_intro)
+            root_obj.get_interface(SC_IFACE)
+            return True
+
+        except Exception as e:
+            logger.error(f"Portal not available: {e}")
+            self.bus = None
+            return False
+
+    async def start(
+        self,
+        output_dir: str,
+        framerate: int = 1,
+        draw_cursor: bool = True,
+    ) -> list[StreamInfo]:
+        """
+        Start screencast recording for all monitors.
+
+        Files are written directly to output_dir with final names (position_connector_screen.webm).
+        The output_dir is typically a segment directory that will be renamed on completion.
+
+        Args:
+            output_dir: Directory for output files (e.g., YYYYMMDD/stream/HHMMSS.incomplete/)
+            framerate: Frames per second (default: 1)
+            draw_cursor: Whether to draw mouse cursor (default: True)
+
+        Returns:
+            List of StreamInfo for each monitor being recorded.
+
+        Raises:
+            RuntimeError: If recording fails to start.
+        """
+        if not await self.connect():
+            raise RuntimeError("Portal not available")
+
+        # Get monitor info from GDK for connector IDs
+        from .activity import get_monitor_geometries
+        try:
+            monitors = get_monitor_geometries()
+        except Exception as e:
+            logger.warning(f"Failed to get monitor geometries: {e}")
+            monitors = []
+
+        # Get portal interface
+        root_intro = await self.bus.introspect(PORTAL_BUS, PORTAL_PATH)
+        root_obj = self.bus.get_proxy_object(PORTAL_BUS, PORTAL_PATH, root_intro)
+        screencast = root_obj.get_interface(SC_IFACE)
+
+        # 1) CreateSession
+        create_token = "h_" + uuid.uuid4().hex
+        create_handle = _make_request_handle(self.bus, create_token)
+        create_fut = _prepare_request_handler(self.bus, create_handle)
+
+        create_opts = {
+            "handle_token": Variant("s", create_token),
+            "session_handle_token": Variant("s", "s_" + uuid.uuid4().hex),
+        }
+
+        await screencast.call_create_session(create_opts)
+        resp, results = await create_fut
+        if resp != 0:
+            raise RuntimeError(f"CreateSession failed with code {resp}")
+
+        self.session_handle = str(_variant_or_value(results.get("session_handle")))
+        if not self.session_handle:
+            raise RuntimeError("CreateSession returned no session_handle")
+
+        logger.debug(f"Portal session: {self.session_handle}")
+
+        # 2) SelectSources
+        restore_token = _load_restore_token(self._restore_token_path)
+        if restore_token:
+            logger.debug("Using saved restore token")
+
+        cursor_mode = 1 if draw_cursor else 0
+
+        select_token = "h_" + uuid.uuid4().hex
+        select_handle = _make_request_handle(self.bus, select_token)
+        select_fut = _prepare_request_handler(self.bus, select_handle)
+
+        select_opts = {
+            "handle_token": Variant("s", select_token),
+            "types": Variant("u", 1),  # 1 = MONITOR
+            "multiple": Variant("b", True),
+            "cursor_mode": Variant("u", cursor_mode),
+            "persist_mode": Variant("u", 2),  # Persist until revoked
+        }
+        if restore_token:
+            select_opts["restore_token"] = Variant("s", restore_token)
+
+        await screencast.call_select_sources(self.session_handle, select_opts)
+        resp, _ = await select_fut
+        if resp != 0:
+            await self._close_session()
+            raise RuntimeError(f"SelectSources failed with code {resp}")
+
+        # 3) Start
+        start_token = "h_" + uuid.uuid4().hex
+        start_handle = _make_request_handle(self.bus, start_token)
+        start_fut = _prepare_request_handler(self.bus, start_handle)
+
+        start_opts = {"handle_token": Variant("s", start_token)}
+        await screencast.call_start(self.session_handle, "", start_opts)
+        resp, results = await start_fut
+        if resp != 0:
+            await self._close_session()
+            raise RuntimeError(f"Start failed with code {resp}")
+
+        portal_streams = _variant_or_value(results.get("streams")) or []
+        if not portal_streams:
+            await self._close_session()
+            raise RuntimeError("Start returned no streams")
+
+        # Save new restore token if provided
+        new_token = _variant_or_value(results.get("restore_token"))
+        if isinstance(new_token, str) and new_token.strip():
+            _save_restore_token(new_token, self._restore_token_path)
+
+        # Parse streams
+        stream_info = []
+        for idx, stream in enumerate(portal_streams):
+            try:
+                node_id = int(stream[0])
+                props = stream[1] if len(stream) > 1 else {}
+                stream_info.append({"idx": idx, "node_id": node_id, "props": props})
+            except Exception as e:
+                logger.warning(f"Could not parse stream {idx}: {e}")
+
+        if not stream_info:
+            await self._close_session()
+            raise RuntimeError("No valid streams found")
+
+        # Match streams to monitors
+        stream_info = _match_streams_to_monitors(stream_info, monitors)
+
+        logger.info(f"Portal returned {len(stream_info)} stream(s)")
+
+        # 4) OpenPipeWireRemote
+        fd_obj = await screencast.call_open_pipe_wire_remote(self.session_handle, {})
+        if hasattr(fd_obj, "take"):
+            self.pw_fd = fd_obj.take()
+        else:
+            self.pw_fd = int(fd_obj)
+
+        # 5) Build GStreamer pipeline
+        self.streams = []
+        pipeline_parts = []
+
+        for info in stream_info:
+            node_id = info["node_id"]
+            position = info["position_label"]
+            connector = info["connector"]
+
+            # Final file path: position_connector_screen.webm
+            file_path = os.path.join(output_dir, f"{position}_{connector}_screen.webm")
+
+            stream_obj = StreamInfo(
+                node_id=node_id,
+                position=position,
+                connector=connector,
+                x=info["x"],
+                y=info["y"],
+                width=info["width"],
+                height=info["height"],
+                file_path=file_path,
+            )
+            self.streams.append(stream_obj)
+
+            # GStreamer branch for this stream
+            branch = (
+                f"pipewiresrc fd={self.pw_fd} path={node_id} ! "
+                f"videorate ! video/x-raw,framerate={framerate}/1 ! "
+                f"videoconvert ! vp8enc end-usage=cq cq-level=4 max-quantizer=15 "
+                f"keyframe-max-dist=30 static-threshold=100 ! webmmux ! "
+                f"filesink location={file_path}"
+            )
+            pipeline_parts.append(branch)
+
+            logger.info(f"  Stream {node_id}: {position} ({connector}) -> {file_path}")
+
+        pipeline_str = " ".join(pipeline_parts)
+        cmd = ["gst-launch-1.0", "-e"] + pipeline_str.split()
+
+        try:
+            self.gst_process = subprocess.Popen(
+                cmd,
+                pass_fds=(self.pw_fd,),
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+            )
+        except FileNotFoundError:
+            await self._close_session()
+            raise RuntimeError("gst-launch-1.0 not found")
+        except Exception as e:
+            await self._close_session()
+            raise RuntimeError(f"Failed to start GStreamer: {e}")
+
+        # Brief delay to check for immediate failure
+        await asyncio.sleep(0.2)
+        if self.gst_process.poll() is not None:
+            stderr = (
+                self.gst_process.stderr.read().decode()
+                if self.gst_process.stderr
+                else ""
+            )
+            await self._close_session()
+            raise RuntimeError(f"GStreamer exited immediately: {stderr[:200]}")
+
+        self._started = True
+        return self.streams
+
+    async def stop(self) -> list[StreamInfo]:
+        """
+        Stop screencast recording gracefully.
+
+        Returns:
+            List of StreamInfo with file_path for the recorded files.
+        """
+        streams = self.streams.copy()
+
+        # Stop GStreamer with SIGINT for clean EOS
+        if self.gst_process and self.gst_process.poll() is None:
+            try:
+                self.gst_process.send_signal(signal.SIGINT)
+                try:
+                    await asyncio.wait_for(
+                        asyncio.to_thread(self.gst_process.wait),
+                        timeout=5.0,
+                    )
+                except asyncio.TimeoutError:
+                    logger.warning("GStreamer did not exit cleanly, killing")
+                    self.gst_process.kill()
+                    self.gst_process.wait()
+            except Exception as e:
+                logger.warning(f"Error stopping GStreamer: {e}")
+
+        self.gst_process = None
+
+        # Close PipeWire fd
+        if self.pw_fd is not None:
+            try:
+                os.close(self.pw_fd)
+            except OSError:
+                pass
+            self.pw_fd = None
+
+        # Close portal session
+        await self._close_session()
+
+        self.streams = []
+        self._started = False
+
+        return streams
+
+    async def _close_session(self):
+        """Close the portal session."""
+        if self.session_handle and self.bus:
+            try:
+                session_intro = await self.bus.introspect(
+                    PORTAL_BUS, self.session_handle
+                )
+                session_obj = self.bus.get_proxy_object(
+                    PORTAL_BUS, self.session_handle, session_intro
+                )
+                session_iface = session_obj.get_interface(SESSION_IFACE)
+                await session_iface.call_close()
+            except Exception:
+                pass
+        self.session_handle = None
+
+    def is_healthy(self) -> bool:
+        """Check if recording is still running."""
+        if not self._started:
+            return False
+        if self.gst_process is None:
+            return False
+        return self.gst_process.poll() is None

src/solstone_linux/session_env.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Desktop session environment checks and recovery.
+
+Extracted from solstone's observe/linux/observer.py (lines 598-666).
+
+_recover_session_env() is kept as fallback for manual CLI launch.
+For systemd service launch, PassEnvironment= in the unit file is
+the primary mechanism.
+"""
+
+import logging
+import os
+import shutil
+import subprocess
+
+logger = logging.getLogger(__name__)
+
+# Exit codes
+EXIT_TEMPFAIL = 75  # EX_TEMPFAIL: session not ready, retry later
+
+
+def _recover_session_env() -> None:
+    """Try to recover desktop session env vars from the systemd user manager.
+
+    On GNOME Wayland, gnome-shell pushes DISPLAY, WAYLAND_DISPLAY, and
+    DBUS_SESSION_BUS_ADDRESS into the systemd user environment on startup.
+    When the observer is launched from a non-desktop shell, these vars may be missing
+    from the inherited environment — but systemctl --user show-environment
+    has them.
+    """
+    needed = {"DISPLAY", "WAYLAND_DISPLAY", "DBUS_SESSION_BUS_ADDRESS"}
+    missing = {v for v in needed if not os.environ.get(v)}
+    if not missing:
+        return
+
+    # Ensure XDG_RUNTIME_DIR is set (required for systemctl --user to connect)
+    if not os.environ.get("XDG_RUNTIME_DIR"):
+        os.environ["XDG_RUNTIME_DIR"] = f"/run/user/{os.getuid()}"
+
+    try:
+        result = subprocess.run(
+            ["systemctl", "--user", "show-environment"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+        if result.returncode != 0:
+            return
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        return
+
+    recovered = []
+    for line in result.stdout.splitlines():
+        key, _, value = line.partition("=")
+        if key in missing and value:
+            os.environ[key] = value
+            recovered.append(f"{key}={value}")
+
+    if recovered:
+        logger.info("Recovered session env from systemd: %s", ", ".join(recovered))
+
+
+def check_session_ready() -> str | None:
+    """Check if the desktop session is ready for observation.
+
+    Returns None if ready, or a description of what's missing.
+    """
+    # Try to recover missing session vars from systemd user manager
+    _recover_session_env()
+
+    # Display server
+    if not os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"):
+        return "no display server (DISPLAY/WAYLAND_DISPLAY not set)"
+
+    # DBus session bus
+    if not os.environ.get("DBUS_SESSION_BUS_ADDRESS"):
+        return "no DBus session bus (DBUS_SESSION_BUS_ADDRESS not set)"
+
+    # PulseAudio / PipeWire audio
+    pactl = shutil.which("pactl")
+    if pactl:
+        try:
+            subprocess.run(
+                [pactl, "info"],
+                capture_output=True,
+                timeout=5,
+            ).check_returncode()
+        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
+            return "audio server not responding (pactl info failed)"
+    return None

src/solstone_linux/streams.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Stream identity for observer segments.
+
+Extracted from solstone's think/streams.py — only the pure naming functions
+needed by standalone observers.
+
+Naming convention (separator is '.'):
+    Local Linux:  {hostname}         e.g. "archon"
+    Remote:       {remote_name}      e.g. "desktop"
+"""
+
+from __future__ import annotations
+
+import re
+
+_STREAM_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$")
+
+
+def _strip_hostname(name: str) -> str:
+    """Strip domain suffix from a hostname, keeping only the first label.
+
+    Dots in stream names are reserved for qualifiers (e.g., '.tmux').
+    Hostnames like 'ja1r.local' or '192.168.1.1' must be reduced to a
+    dot-free base name.
+
+    Examples: 'ja1r.local' -> 'ja1r', '192.168.1.1' -> '192-168-1-1',
+    'archon' -> 'archon', 'my.host.example.com' -> 'my'
+    """
+    name = name.strip()
+    if not name:
+        return name
+    parts = name.split(".")
+    if all(p.isdigit() for p in parts if p):
+        return "-".join(p for p in parts if p)
+    return parts[0]
+
+
+def stream_name(
+    *,
+    host: str | None = None,
+    remote: str | None = None,
+    qualifier: str | None = None,
+) -> str:
+    """Derive canonical stream name from source characteristics.
+
+    Parameters
+    ----------
+    host : str, optional
+        Local hostname (e.g., "archon").
+    remote : str, optional
+        Remote observer name (e.g., "desktop").
+    qualifier : str, optional
+        Sub-stream qualifier. Appended with dot separator.
+
+    Returns
+    -------
+    str
+        Canonical stream name.
+
+    Raises
+    ------
+    ValueError
+        If no source is provided, or the resulting name is invalid.
+    """
+    if host:
+        base = _strip_hostname(host)
+    elif remote:
+        base = _strip_hostname(remote)
+    else:
+        raise ValueError("stream_name requires host or remote")
+
+    name = base.lower().strip()
+    name = re.sub(r"[\s/\\]+", "-", name)
+
+    if qualifier:
+        qualifier = qualifier.lower().strip()
+        qualifier = re.sub(r"[\s/\\]+", "-", qualifier)
+        name = f"{name}.{qualifier}"
+
+    if not name or ".." in name:
+        raise ValueError(f"Invalid stream name: {name!r}")
+    if not _STREAM_NAME_RE.match(name):
+        raise ValueError(f"Invalid stream name: {name!r}")
+
+    return name

src/solstone_linux/sync.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Background sync service for uploading captured segments.
+
+Modeled on solstone-macos's SyncService.swift. Runs as an asyncio
+background task in the same event loop as capture. Walks cache days
+newest-to-oldest, queries server for existing segments, uploads missing ones.
+
+Refinements over tmux baseline:
+- Respects configured sync_max_retries (no hard min(config,3) cap)
+- Circuit breaker tuned by error type: auth=immediate, transient=5-10
+- Synced-days pruning at 90 days to prevent unbounded cache growth
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import time
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+from .config import Config
+from .upload import ErrorType, UploadClient
+
+logger = logging.getLogger(__name__)
+
+# Circuit breaker thresholds by error type
+CIRCUIT_THRESHOLD_AUTH = 1       # Auth failures open immediately
+CIRCUIT_THRESHOLD_TRANSIENT = 5  # Transient failures need 5 consecutive
+
+# Synced days older than this are pruned from the cache
+SYNCED_DAYS_MAX_AGE = 90
+
+
+class SyncService:
+    """Background sync service that uploads completed segments to the server."""
+
+    def __init__(self, config: Config, client: UploadClient):
+        self._config = config
+        self._client = client
+        self._synced_days: set[str] = set()
+        self._consecutive_failures = 0
+        self._last_error_type: ErrorType | None = None
+        self._circuit_open = False
+        self._last_full_sync: float = 0
+        self._running = True
+        self._trigger = asyncio.Event()
+
+        # Load synced days cache
+        self._load_synced_days()
+
+    def _synced_days_path(self) -> Path:
+        return self._config.state_dir / "synced_days.json"
+
+    def _load_synced_days(self) -> None:
+        path = self._synced_days_path()
+        if not path.exists():
+            return
+        try:
+            with open(path, encoding="utf-8") as f:
+                data = json.load(f)
+            self._synced_days = set(data) if isinstance(data, list) else set()
+        except (json.JSONDecodeError, OSError):
+            self._synced_days = set()
+
+    def _save_synced_days(self) -> None:
+        self._config.state_dir.mkdir(parents=True, exist_ok=True)
+        path = self._synced_days_path()
+        tmp = path.with_suffix(f".{os.getpid()}.tmp")
+        try:
+            with open(tmp, "w", encoding="utf-8") as f:
+                json.dump(sorted(self._synced_days), f)
+                f.write("\n")
+            os.rename(str(tmp), str(path))
+        except OSError as e:
+            logger.warning(f"Failed to save synced days: {e}")
+
+    def _prune_synced_days(self) -> None:
+        """Remove synced-days entries older than 90 days."""
+        if not self._synced_days:
+            return
+        cutoff = (datetime.now() - timedelta(days=SYNCED_DAYS_MAX_AGE)).strftime("%Y%m%d")
+        before = len(self._synced_days)
+        self._synced_days = {d for d in self._synced_days if d >= cutoff}
+        pruned = before - len(self._synced_days)
+        if pruned:
+            logger.info(f"Pruned {pruned} synced-days entries older than {SYNCED_DAYS_MAX_AGE} days")
+            self._save_synced_days()
+
+    def _circuit_threshold(self) -> int:
+        """Get circuit breaker threshold based on last error type."""
+        if self._last_error_type == ErrorType.AUTH:
+            return CIRCUIT_THRESHOLD_AUTH
+        return CIRCUIT_THRESHOLD_TRANSIENT
+
+    def trigger(self) -> None:
+        """Trigger a sync pass (called by observer on segment completion)."""
+        self._trigger.set()
+
+    def stop(self) -> None:
+        """Stop the sync service."""
+        self._running = False
+        self._trigger.set()
+
+    async def run(self) -> None:
+        """Main sync loop — waits for triggers, then syncs."""
+        # Prune on startup
+        self._prune_synced_days()
+
+        while self._running:
+            try:
+                # Wait for trigger or periodic check (60s timeout)
+                try:
+                    await asyncio.wait_for(self._trigger.wait(), timeout=60)
+                except asyncio.TimeoutError:
+                    pass
+
+                self._trigger.clear()
+
+                if not self._running:
+                    break
+
+                if self._circuit_open:
+                    logger.warning("Circuit breaker open — skipping sync")
+                    continue
+
+                # Force full sync daily
+                now = time.time()
+                force_full = (now - self._last_full_sync) > 86400
+
+                await self._sync(force_full=force_full)
+
+                if force_full:
+                    self._last_full_sync = now
+
+            except Exception as e:
+                logger.error(f"Sync error: {e}", exc_info=True)
+                await asyncio.sleep(5)
+
+    async def _sync(self, force_full: bool = False) -> None:
+        """Walk days newest-to-oldest and upload missing segments."""
+        captures_dir = self._config.captures_dir
+        if not captures_dir.exists():
+            return
+
+        today = datetime.now().strftime("%Y%m%d")
+
+        # Collect segments by day
+        segments_by_day = self._collect_segments(captures_dir)
+        if not segments_by_day:
+            return
+
+        for day in sorted(segments_by_day.keys(), reverse=True):
+            if not self._running:
+                break
+
+            if self._circuit_open:
+                break
+
+            # Skip past days already fully synced (unless forcing)
+            if day != today and day in self._synced_days and not force_full:
+                continue
+
+            local_segments = segments_by_day[day]
+
+            # Query server for existing segments
+            server_segments = await asyncio.to_thread(
+                self._client.get_server_segments, day
+            )
+            if server_segments is None:
+                logger.warning(f"Failed to query server for day {day}")
+                continue
+
+            # Build lookup
+            server_keys: set[str] = set()
+            for seg in server_segments:
+                server_keys.add(seg.get("key", ""))
+                if "original_key" in seg:
+                    server_keys.add(seg["original_key"])
+
+            any_needed_upload = False
+
+            for segment_dir in local_segments:
+                if not self._running or self._circuit_open:
+                    break
+
+                segment_key = segment_dir.name
+                if segment_key in server_keys:
+                    continue
+
+                any_needed_upload = True
+                success = await self._upload_segment(day, segment_dir)
+
+                if not success:
+                    self._consecutive_failures += 1
+                    threshold = self._circuit_threshold()
+                    if self._consecutive_failures >= threshold:
+                        self._circuit_open = True
+                        logger.error(
+                            f"Circuit breaker OPEN: {self._consecutive_failures} consecutive "
+                            f"{self._last_error_type.value if self._last_error_type else 'unknown'} "
+                            f"failures (threshold: {threshold})"
+                        )
+                        break
+                else:
+                    self._consecutive_failures = 0
+                    self._last_error_type = None
+
+            # Mark past days as synced if nothing needed upload
+            if day != today and not any_needed_upload:
+                self._synced_days.add(day)
+                self._save_synced_days()
+
+    def _collect_segments(self, captures_dir: Path) -> dict[str, list[Path]]:
+        """Collect completed segments grouped by day."""
+        result: dict[str, list[Path]] = {}
+
+        for day_dir in sorted(captures_dir.iterdir(), reverse=True):
+            if not day_dir.is_dir():
+                continue
+
+            day = day_dir.name
+
+            for stream_dir in day_dir.iterdir():
+                if not stream_dir.is_dir():
+                    continue
+
+                segments = []
+                for seg_dir in sorted(stream_dir.iterdir(), reverse=True):
+                    if not seg_dir.is_dir():
+                        continue
+                    name = seg_dir.name
+                    # Skip incomplete and failed
+                    if name.endswith(".incomplete") or name.endswith(".failed"):
+                        continue
+                    segments.append(seg_dir)
+
+                if segments:
+                    result.setdefault(day, []).extend(segments)
+
+        return result
+
+    async def _upload_segment(self, day: str, segment_dir: Path) -> bool:
+        """Upload a single segment with retry logic."""
+        segment_key = segment_dir.name
+        files = [f for f in segment_dir.iterdir() if f.is_file()]
+        if not files:
+            return True  # Nothing to upload
+
+        meta: dict[str, Any] = {"stream": self._config.stream}
+
+        result = await asyncio.to_thread(
+            self._client.upload_segment, day, segment_key, files, meta
+        )
+
+        if result.success:
+            logger.info(f"Uploaded: {day}/{segment_key} ({len(files)} files)")
+            return True
+
+        # Track error type for circuit breaker
+        self._last_error_type = result.error_type
+
+        # Non-retryable errors
+        if self._client.is_revoked:
+            logger.error("Client revoked — disabling sync")
+            self._circuit_open = True
+            return False
+
+        logger.error(f"Upload failed: {day}/{segment_key}")
+        return False

src/solstone_linux/upload.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""HTTP upload client for solstone ingest server.
+
+Extracted from solstone's observe/remote_client.py. Accepts Config
+as constructor parameter instead of reading config internally.
+
+Refinements over tmux baseline:
+- Respects configured sync_max_retries without hard cap
+- Error classification: auth (401/403) vs transient (5xx/network)
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from enum import Enum
+from pathlib import Path
+from typing import Any, NamedTuple
+
+import requests
+
+from .config import Config
+
+logger = logging.getLogger(__name__)
+
+UPLOAD_TIMEOUT = 300
+EVENT_TIMEOUT = 30
+
+
+class ErrorType(Enum):
+    """Classification of upload errors for circuit breaker tuning."""
+    AUTH = "auth"           # 401, 403 — open circuit immediately
+    CLIENT = "client"       # 400 — non-retryable, don't count for circuit
+    TRANSIENT = "transient" # 5xx, network, timeout — allow more failures
+
+
+class UploadResult(NamedTuple):
+    success: bool
+    duplicate: bool = False
+    error_type: ErrorType | None = None
+
+
+class UploadClient:
+    """HTTP client for uploading observer segments to the ingest server."""
+
+    def __init__(self, config: Config):
+        self._url = config.server_url.rstrip("/") if config.server_url else ""
+        self._key = config.key
+        self._stream = config.stream
+        self._revoked = False
+        self._session = requests.Session()
+        self._retry_backoff = config.sync_retry_delays or [5, 30, 120, 300]
+        # Respect configured retry cap — no hard min(config, 3)
+        self._max_retries = config.sync_max_retries
+
+    @property
+    def is_revoked(self) -> bool:
+        return self._revoked
+
+    def _persist_key(self, config: Config, key: str) -> None:
+        """Save auto-registered key back to config."""
+        from .config import save_config
+
+        config.key = key
+        save_config(config)
+
+    def ensure_registered(self, config: Config) -> bool:
+        """Ensure the client has a valid key, auto-registering if needed.
+
+        Returns True if a key is available.
+        """
+        if self._key:
+            return True
+        if not self._url:
+            return False
+
+        url = f"{self._url}/app/remote/api/create"
+        name = self._stream or "solstone-linux"
+
+        retries = min(3, len(self._retry_backoff))
+        for attempt in range(retries):
+            delay = self._retry_backoff[min(attempt, len(self._retry_backoff) - 1)]
+            try:
+                resp = self._session.post(
+                    url, json={"name": name}, timeout=EVENT_TIMEOUT
+                )
+                if resp.status_code == 200:
+                    data = resp.json()
+                    self._key = data["key"]
+                    self._persist_key(config, self._key)
+                    logger.info(f"Auto-registered as '{name}' (key: {self._key[:8]}...)")
+                    return True
+                elif resp.status_code == 403:
+                    self._revoked = True
+                    logger.error("Registration rejected (403)")
+                    return False
+                else:
+                    logger.warning(
+                        f"Registration attempt {attempt + 1} failed: {resp.status_code}"
+                    )
+            except requests.RequestException as e:
+                logger.warning(f"Registration attempt {attempt + 1} failed: {e}")
+            if attempt < retries - 1:
+                time.sleep(delay)
+
+        logger.error(f"Registration failed after {retries} attempts")
+        return False
+
+    @staticmethod
+    def classify_error(status_code: int | None, is_network_error: bool = False) -> ErrorType:
+        """Classify an error for circuit breaker and retry decisions."""
+        if is_network_error:
+            return ErrorType.TRANSIENT
+        if status_code is None:
+            return ErrorType.TRANSIENT
+        if status_code in (401, 403):
+            return ErrorType.AUTH
+        if status_code == 400:
+            return ErrorType.CLIENT
+        # 5xx and anything else
+        return ErrorType.TRANSIENT
+
+    def upload_segment(
+        self,
+        day: str,
+        segment: str,
+        files: list[Path],
+        meta: dict[str, Any] | None = None,
+    ) -> UploadResult:
+        """Upload a segment's files to the ingest server."""
+        if self._revoked or not self._key or not self._url:
+            return UploadResult(False, error_type=ErrorType.AUTH if self._revoked else None)
+
+        url = f"{self._url}/app/remote/ingest/{self._key}"
+
+        for attempt in range(self._max_retries):
+            file_handles = []
+            files_data = []
+            error_type = None
+            try:
+                for path in files:
+                    if not path.exists():
+                        logger.warning(f"File not found, skipping: {path}")
+                        continue
+                    fh = open(path, "rb")
+                    file_handles.append(fh)
+                    files_data.append(
+                        ("files", (path.name, fh, "application/octet-stream"))
+                    )
+
+                if not files_data:
+                    return UploadResult(False)
+
+                data: dict[str, Any] = {"day": day, "segment": segment}
+                if meta:
+                    data["meta"] = json.dumps(meta)
+
+                response = self._session.post(
+                    url, data=data, files=files_data, timeout=UPLOAD_TIMEOUT
+                )
+
+                if response.status_code == 200:
+                    resp_data = response.json()
+                    is_duplicate = resp_data.get("status") == "duplicate"
+                    return UploadResult(True, duplicate=is_duplicate)
+
+                error_type = self.classify_error(response.status_code)
+
+                if error_type == ErrorType.AUTH:
+                    if response.status_code == 403:
+                        self._revoked = True
+                    logger.error(
+                        f"Upload rejected ({response.status_code}): {response.text}"
+                    )
+                    return UploadResult(False, error_type=error_type)
+
+                if error_type == ErrorType.CLIENT:
+                    logger.error(
+                        f"Upload rejected ({response.status_code}): {response.text}"
+                    )
+                    return UploadResult(False, error_type=error_type)
+
+                logger.warning(
+                    f"Upload attempt {attempt + 1} failed: "
+                    f"{response.status_code} {response.text}"
+                )
+            except requests.RequestException as e:
+                error_type = ErrorType.TRANSIENT
+                logger.warning(f"Upload attempt {attempt + 1} failed: {e}")
+            finally:
+                for fh in file_handles:
+                    try:
+                        fh.close()
+                    except Exception:
+                        pass
+
+            if attempt < self._max_retries - 1:
+                delay = self._retry_backoff[min(attempt, len(self._retry_backoff) - 1)]
+                time.sleep(delay)
+
+        logger.error(f"Upload failed after {self._max_retries} attempts: {day}/{segment}")
+        return UploadResult(False, error_type=error_type)
+
+    def get_server_segments(self, day: str) -> list[dict] | None:
+        """Query server for segments on a given day.
+
+        Returns list of segment dicts, or None on failure.
+        """
+        if self._revoked or not self._key or not self._url:
+            return None
+
+        url = f"{self._url}/app/remote/ingest/{self._key}/segments/{day}"
+        params = {}
+        if self._stream:
+            params["stream"] = self._stream
+
+        try:
+            resp = self._session.get(url, params=params, timeout=EVENT_TIMEOUT)
+            if resp.status_code == 200:
+                return resp.json()
+            if resp.status_code in (401, 403):
+                if resp.status_code == 403:
+                    self._revoked = True
+                logger.error(f"Segments query rejected ({resp.status_code})")
+                return None
+            logger.warning(f"Segments query failed: {resp.status_code}")
+            return None
+        except requests.RequestException as e:
+            logger.debug(f"Segments query failed: {e}")
+            return None
+
+    def relay_event(self, tract: str, event: str, **fields: Any) -> bool:
+        """Fire-and-forget event relay."""
+        if self._revoked or not self._key or not self._url:
+            return False
+
+        url = f"{self._url}/app/remote/ingest/{self._key}/event"
+        payload = {"tract": tract, "event": event, **fields}
+        try:
+            resp = self._session.post(url, json=payload, timeout=EVENT_TIMEOUT)
+            if resp.status_code == 200:
+                return True
+            if resp.status_code == 403:
+                self._revoked = True
+            return False
+        except requests.RequestException:
+            return False
+
+    def stop(self) -> None:
+        self._session.close()

tests/test_config.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from pathlib import Path
+
+from solstone_linux.config import Config, load_config, save_config
+
+
+class TestConfig:
+    def test_defaults(self):
+        config = Config()
+        assert config.server_url == ""
+        assert config.key == ""
+        assert config.segment_interval == 300
+
+    def test_captures_dir(self):
+        config = Config()
+        assert config.captures_dir == config.base_dir / "captures"
+
+    def test_restore_token_path(self):
+        config = Config()
+        assert config.restore_token_path == config.base_dir / "config" / "restore_token"
+
+    def test_round_trip(self, tmp_path: Path):
+        config = Config(base_dir=tmp_path)
+        config.server_url = "https://example.com"
+        config.key = "test-key-123"
+        config.stream = "archon"
+        config.segment_interval = 600
+
+        save_config(config)
+
+        loaded = load_config(tmp_path)
+        assert loaded.server_url == "https://example.com"
+        assert loaded.key == "test-key-123"
+        assert loaded.stream == "archon"
+        assert loaded.segment_interval == 600
+
+    def test_load_missing(self, tmp_path: Path):
+        config = load_config(tmp_path)
+        assert config.server_url == ""
+        assert config.key == ""
+
+    def test_load_corrupt(self, tmp_path: Path):
+        config_dir = tmp_path / "config"
+        config_dir.mkdir(parents=True)
+        (config_dir / "config.json").write_text("not json!")
+
+        config = load_config(tmp_path)
+        assert config.server_url == ""
+
+    def test_permissions(self, tmp_path: Path):
+        config = Config(base_dir=tmp_path)
+        config.server_url = "https://example.com"
+        config.key = "secret"
+        save_config(config)
+
+        mode = config.config_path.stat().st_mode & 0o777
+        assert mode == 0o600
+
+    def test_sync_config_roundtrip(self, tmp_path: Path):
+        config = Config(base_dir=tmp_path)
+        config.sync_retry_delays = [10, 60, 300]
+        config.sync_max_retries = 5
+        save_config(config)
+
+        loaded = load_config(tmp_path)
+        assert loaded.sync_retry_delays == [10, 60, 300]
+        assert loaded.sync_max_retries == 5

tests/test_monitor_positions.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+from solstone_linux.monitor_positions import assign_monitor_positions
+
+
+class TestAssignMonitorPositions:
+    def test_single_monitor(self):
+        monitors = [{"id": "DP-1", "box": [0, 0, 1920, 1080]}]
+        result = assign_monitor_positions(monitors)
+        assert result[0]["position"] == "center"
+
+    def test_two_horizontal(self):
+        monitors = [
+            {"id": "DP-1", "box": [0, 0, 1920, 1080]},
+            {"id": "DP-2", "box": [1920, 0, 3840, 1080]},
+        ]
+        result = assign_monitor_positions(monitors)
+        positions = {m["id"]: m["position"] for m in result}
+        assert positions["DP-1"] == "left"
+        assert positions["DP-2"] == "right"
+
+    def test_three_horizontal(self):
+        monitors = [
+            {"id": "DP-1", "box": [0, 0, 1920, 1080]},
+            {"id": "DP-2", "box": [1920, 0, 3840, 1080]},
+            {"id": "DP-3", "box": [3840, 0, 5760, 1080]},
+        ]
+        result = assign_monitor_positions(monitors)
+        positions = {m["id"]: m["position"] for m in result}
+        assert positions["DP-1"] == "left"
+        assert positions["DP-2"] == "center"
+        assert positions["DP-3"] == "right"
+
+    def test_stacked_vertical(self):
+        monitors = [
+            {"id": "DP-1", "box": [0, 0, 1920, 1080]},
+            {"id": "DP-2", "box": [0, 1080, 1920, 2160]},
+        ]
+        result = assign_monitor_positions(monitors)
+        positions = {m["id"]: m["position"] for m in result}
+        assert positions["DP-1"] == "top"
+        assert positions["DP-2"] == "bottom"
+
+    def test_empty(self):
+        assert assign_monitor_positions([]) == []
+
+    def test_offset_monitors_no_phantom_vertical(self):
+        # Two side-by-side monitors that don't overlap horizontally
+        # should NOT get vertical labels
+        monitors = [
+            {"id": "DP-1", "box": [0, 0, 1920, 1080]},
+            {"id": "DP-2", "box": [1920, 200, 3840, 1280]},
+        ]
+        result = assign_monitor_positions(monitors)
+        positions = {m["id"]: m["position"] for m in result}
+        assert positions["DP-1"] == "left"
+        assert positions["DP-2"] == "right"

tests/test_observer.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Tests for the observer module — segment lifecycle and local cache."""
+
+from pathlib import Path
+
+from solstone_linux.config import Config
+from solstone_linux.recovery import write_segment_metadata
+
+
+class TestSegmentMetadata:
+    """Test .metadata file creation for recovery."""
+
+    def test_writes_metadata(self, tmp_path: Path):
+        import json
+
+        seg_dir = tmp_path / "test.incomplete"
+        seg_dir.mkdir()
+        write_segment_metadata(seg_dir, 1712160000.0)
+
+        meta_path = seg_dir / ".metadata"
+        assert meta_path.exists()
+
+        data = json.loads(meta_path.read_text())
+        assert data["start_timestamp"] == 1712160000.0
+
+
+class TestSegmentDirStructure:
+    """Test that config directories follow the expected structure."""
+
+    def test_captures_dir_path(self, tmp_path: Path):
+        config = Config(base_dir=tmp_path)
+        assert str(config.captures_dir).endswith("captures")
+
+    def test_restore_token_path(self, tmp_path: Path):
+        config = Config(base_dir=tmp_path)
+        assert str(config.restore_token_path).endswith("restore_token")
+        assert "config" in str(config.restore_token_path)

tests/test_session_env.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Tests for session environment checks."""
+
+import os
+from unittest.mock import patch
+
+from solstone_linux.session_env import check_session_ready
+
+
+class TestCheckSessionReady:
+    """Test desktop session readiness checks."""
+
+    def test_no_display_server(self):
+        env = {
+            k: v for k, v in os.environ.items()
+            if k not in ("DISPLAY", "WAYLAND_DISPLAY")
+        }
+        with patch.dict(os.environ, env, clear=True):
+            with patch("solstone_linux.session_env._recover_session_env"):
+                result = check_session_ready()
+                assert result is not None
+                assert "display server" in result
+
+    def test_no_dbus(self):
+        env = {
+            k: v for k, v in os.environ.items()
+            if k != "DBUS_SESSION_BUS_ADDRESS"
+        }
+        env["DISPLAY"] = ":0"
+        with patch.dict(os.environ, env, clear=True):
+            with patch("solstone_linux.session_env._recover_session_env"):
+                result = check_session_ready()
+                assert result is not None
+                assert "DBus" in result
+
+    def test_ready_with_display_and_dbus(self):
+        env = dict(os.environ)
+        env["DISPLAY"] = ":0"
+        env["DBUS_SESSION_BUS_ADDRESS"] = "unix:path=/run/user/1000/bus"
+        with patch.dict(os.environ, env, clear=True):
+            with patch("solstone_linux.session_env._recover_session_env"):
+                with patch("solstone_linux.session_env.shutil") as mock_shutil:
+                    mock_shutil.which.return_value = None  # No pactl
+                    result = check_session_ready()
+                    assert result is None  # Ready

tests/test_streams.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+import pytest
+
+from solstone_linux.streams import _strip_hostname, stream_name
+
+
+class TestStripHostname:
+    def test_simple(self):
+        assert _strip_hostname("archon") == "archon"
+
+    def test_with_domain(self):
+        assert _strip_hostname("ja1r.local") == "ja1r"
+
+    def test_ip_address(self):
+        assert _strip_hostname("192.168.1.1") == "192-168-1-1"
+
+    def test_fqdn(self):
+        assert _strip_hostname("my.host.example.com") == "my"
+
+    def test_empty(self):
+        assert _strip_hostname("") == ""
+
+
+class TestStreamName:
+    def test_host_only(self):
+        assert stream_name(host="archon") == "archon"
+
+    def test_host_with_qualifier(self):
+        assert stream_name(host="archon", qualifier="tmux") == "archon.tmux"
+
+    def test_host_no_qualifier(self):
+        # Linux observer uses host without qualifier
+        assert stream_name(host="archon") == "archon"
+
+    def test_remote(self):
+        assert stream_name(remote="desktop") == "desktop"
+
+    def test_rejects_empty(self):
+        with pytest.raises(ValueError):
+            stream_name()
+
+    def test_rejects_invalid_chars(self):
+        with pytest.raises(ValueError):
+            stream_name(host="!invalid")

tests/test_sync.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+import json
+import os
+import time
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from solstone_linux.config import Config
+from solstone_linux.recovery import recover_incomplete_segments
+from solstone_linux.upload import ErrorType, UploadClient, UploadResult
+
+
+class TestRecovery:
+    """Test crash recovery for incomplete segments."""
+
+    def _make_incomplete(
+        self, captures_dir: Path, day: str, stream: str, time_prefix: str, age: int = 300
+    ) -> Path:
+        """Create an incomplete segment directory with a dummy file."""
+        seg_dir = captures_dir / day / stream / f"{time_prefix}.incomplete"
+        seg_dir.mkdir(parents=True)
+        (seg_dir / "center_DP-3_screen.webm").write_bytes(b"\x00" * 100)
+
+        # Set timestamps to simulate age
+        old_time = time.time() - age
+        os.utime(seg_dir, (old_time, old_time))
+        return seg_dir
+
+    def test_recovers_old_incomplete(self, tmp_path: Path):
+        captures_dir = tmp_path / "captures"
+        self._make_incomplete(captures_dir, "20260403", "archon", "140000", age=300)
+
+        recovered = recover_incomplete_segments(captures_dir)
+        assert recovered == 1
+
+        stream_dir = captures_dir / "20260403" / "archon"
+        dirs = [d.name for d in stream_dir.iterdir() if d.is_dir()]
+        assert len(dirs) == 1
+        assert dirs[0].startswith("140000_")
+        assert not dirs[0].endswith(".incomplete")
+
+    def test_recovers_with_metadata(self, tmp_path: Path):
+        """Recovery uses .metadata start_timestamp for accurate duration."""
+        captures_dir = tmp_path / "captures"
+        seg_dir = captures_dir / "20260403" / "archon" / "140000.incomplete"
+        seg_dir.mkdir(parents=True)
+        (seg_dir / "center_DP-3_screen.webm").write_bytes(b"\x00" * 100)
+
+        # Write metadata with known start timestamp (60 seconds ago)
+        start_ts = time.time() - 60
+        meta = {"start_timestamp": start_ts}
+        (seg_dir / ".metadata").write_text(json.dumps(meta))
+
+        # Age the directory
+        old_time = time.time() - 300
+        os.utime(seg_dir, (old_time, old_time))
+
+        recovered = recover_incomplete_segments(captures_dir)
+        assert recovered == 1
+
+        stream_dir = captures_dir / "20260403" / "archon"
+        dirs = [d.name for d in stream_dir.iterdir() if d.is_dir()]
+        assert len(dirs) == 1
+        # Duration should be based on metadata start timestamp, not mtime-ctime
+        duration = int(dirs[0].split("_")[1])
+        assert 55 <= duration <= 65  # ~60 seconds
+
+    def test_skips_recent_incomplete(self, tmp_path: Path):
+        captures_dir = tmp_path / "captures"
+        seg_dir = captures_dir / "20260403" / "archon" / "140000.incomplete"
+        seg_dir.mkdir(parents=True)
+        (seg_dir / "test.webm").write_bytes(b"\x00")
+
+        recovered = recover_incomplete_segments(captures_dir)
+        assert recovered == 0
+        assert seg_dir.exists()
+
+    def test_marks_empty_as_failed(self, tmp_path: Path):
+        captures_dir = tmp_path / "captures"
+        seg_dir = captures_dir / "20260403" / "archon" / "140000.incomplete"
+        seg_dir.mkdir(parents=True)
+        # No files inside — should fail
+
+        old_time = time.time() - 300
+        os.utime(seg_dir, (old_time, old_time))
+
+        recovered = recover_incomplete_segments(captures_dir)
+        assert recovered == 0
+
+        failed_dir = captures_dir / "20260403" / "archon" / "140000.failed"
+        assert failed_dir.exists()
+
+    def test_metadata_removed_on_recovery(self, tmp_path: Path):
+        """The .metadata file should be removed during recovery."""
+        captures_dir = tmp_path / "captures"
+        seg_dir = captures_dir / "20260403" / "archon" / "140000.incomplete"
+        seg_dir.mkdir(parents=True)
+        (seg_dir / "screen.webm").write_bytes(b"\x00")
+        (seg_dir / ".metadata").write_text('{"start_timestamp": 1000}')
+
+        old_time = time.time() - 300
+        os.utime(seg_dir, (old_time, old_time))
+
+        recover_incomplete_segments(captures_dir)
+
+        stream_dir = captures_dir / "20260403" / "archon"
+        for d in stream_dir.iterdir():
+            if d.is_dir() and not d.name.endswith((".incomplete", ".failed")):
+                # .metadata should not be in the recovered dir
+                assert not (d / ".metadata").exists()
+
+    def test_no_captures_dir(self, tmp_path: Path):
+        assert recover_incomplete_segments(tmp_path / "nonexistent") == 0
+
+
+class TestSyncServiceCollect:
+    """Test segment collection logic."""
+
+    def test_skips_incomplete_and_failed(self, tmp_path: Path):
+        from solstone_linux.sync import SyncService
+
+        config = Config(base_dir=tmp_path)
+        config.ensure_dirs()
+
+        captures = config.captures_dir
+        stream_dir = captures / "20260403" / "archon"
+        stream_dir.mkdir(parents=True)
+
+        (stream_dir / "140000_300").mkdir()
+        (stream_dir / "140000_300" / "screen.webm").write_bytes(b"\x00")
+        (stream_dir / "145000.incomplete").mkdir()
+        (stream_dir / "143000.failed").mkdir()
+        (stream_dir / "150000_300").mkdir()
+        (stream_dir / "150000_300" / "audio.flac").write_bytes(b"\x00")
+
+        client = UploadClient(config)
+        sync = SyncService(config, client)
+
+        segments = sync._collect_segments(captures)
+        assert "20260403" in segments
+        names = [s.name for s in segments["20260403"]]
+        assert "140000_300" in names
+        assert "150000_300" in names
+        assert "145000.incomplete" not in names
+        assert "143000.failed" not in names
+
+
+class TestSyncedDaysPruning:
+    """Test that synced-days cache is pruned to 90 days."""
+
+    def test_prunes_old_entries(self, tmp_path: Path):
+        from solstone_linux.sync import SyncService
+
+        config = Config(base_dir=tmp_path)
+        config.ensure_dirs()
+
+        client = UploadClient(config)
+        sync = SyncService(config, client)
+
+        # Add entries spanning 100 days
+        from datetime import datetime, timedelta
+        today = datetime.now()
+        for i in range(100):
+            day = (today - timedelta(days=i)).strftime("%Y%m%d")
+            sync._synced_days.add(day)
+
+        sync._prune_synced_days()
+
+        # Should have ~90 entries (not 100)
+        assert len(sync._synced_days) <= 91  # Allow 1 day tolerance
+
+
+class TestErrorClassification:
+    """Test HTTP error classification for circuit breaker tuning."""
+
+    def test_auth_errors(self):
+        assert UploadClient.classify_error(401) == ErrorType.AUTH
+        assert UploadClient.classify_error(403) == ErrorType.AUTH
+
+    def test_client_errors(self):
+        assert UploadClient.classify_error(400) == ErrorType.CLIENT
+
+    def test_transient_errors(self):
+        assert UploadClient.classify_error(500) == ErrorType.TRANSIENT
+        assert UploadClient.classify_error(502) == ErrorType.TRANSIENT
+        assert UploadClient.classify_error(503) == ErrorType.TRANSIENT
+
+    def test_network_errors(self):
+        assert UploadClient.classify_error(None, is_network_error=True) == ErrorType.TRANSIENT
+
+    def test_unknown_status(self):
+        assert UploadClient.classify_error(418) == ErrorType.TRANSIENT
+
+
+class TestCircuitBreakerThresholds:
+    """Test circuit breaker state transitions with error-type tuning."""
+
+    def test_auth_opens_immediately(self, tmp_path: Path):
+        from solstone_linux.sync import SyncService, CIRCUIT_THRESHOLD_AUTH
+
+        config = Config(base_dir=tmp_path)
+        config.ensure_dirs()
+        client = UploadClient(config)
+        sync = SyncService(config, client)
+
+        sync._last_error_type = ErrorType.AUTH
+        assert sync._circuit_threshold() == CIRCUIT_THRESHOLD_AUTH
+        assert CIRCUIT_THRESHOLD_AUTH == 1
+
+    def test_transient_allows_more_failures(self, tmp_path: Path):
+        from solstone_linux.sync import SyncService, CIRCUIT_THRESHOLD_TRANSIENT
+
+        config = Config(base_dir=tmp_path)
+        config.ensure_dirs()
+        client = UploadClient(config)
+        sync = SyncService(config, client)
+
+        sync._last_error_type = ErrorType.TRANSIENT
+        assert sync._circuit_threshold() == CIRCUIT_THRESHOLD_TRANSIENT
+        assert CIRCUIT_THRESHOLD_TRANSIENT >= 5
+
+
+class TestRetryCapRespected:
+    """Test that upload respects configured retry cap (no hard min(config,3))."""
+
+    def test_respects_configured_max_retries(self):
+        """Upload client should use the configured max_retries, not cap at 3."""
+        config = Config()
+        config.sync_max_retries = 10
+        client = UploadClient(config)
+        assert client._max_retries == 10
+
+    def test_low_max_retries_respected(self):
+        config = Config()
+        config.sync_max_retries = 1
+        client = UploadClient(config)
+        assert client._max_retries == 1