tmux observer
Add tmux status bar indicator for observer state
New ☼ indicator in tmux status-left shows observer/sync state: yellow when sync is connected, grey when offline. Controlled by `status_indicator` config field (default true). Restores original status-left on clean shutdown.
+267
+1
AGENTS.md
+1
AGENTS.md
···
17
17
config.py Config loading/persistence (~/.local/share/solstone-tmux/)
18
18
capture.py Tmux capture library (polls sessions, panes, deduplication)
19
19
observer.py Main capture loop with segment rotation
20
+
indicator.py Tmux status-left indicator (☼ sync state display)
20
21
streams.py Stream name derivation (hostname.tmux convention)
21
22
sync.py Background sync service (uploads segments to server)
22
23
upload.py HTTP upload client for solstone ingest API
+22
INSTALL.md
+22
INSTALL.md
···
70
70
sol observer list
71
71
```
72
72
73
+
## status bar indicator
74
+
75
+
solstone-tmux shows a ☼ symbol at the left edge of your tmux status bar while running:
76
+
77
+
- **yellow ☼** — observer active, sync connected
78
+
- **grey ☼** — observer active, sync offline (server unreachable or not configured)
79
+
- **absent** — observer not running
80
+
81
+
the indicator is removed automatically on clean shutdown (SIGTERM, SIGINT). if the observer is killed with SIGKILL or the system crashes, the indicator may persist. to clear it manually:
82
+
83
+
```
84
+
tmux set -g @solstone ""
85
+
```
86
+
87
+
to disable the indicator entirely, add to config.json:
88
+
89
+
```json
90
+
{
91
+
"status_indicator": false
92
+
}
93
+
```
94
+
73
95
## notes
74
96
75
97
- if pipx is not installed: `pip install --user pipx` or install via your package manager.
+4
src/solstone_tmux/config.py
+4
src/solstone_tmux/config.py
···
39
39
)
40
40
sync_max_retries: int = DEFAULT_SYNC_MAX_RETRIES
41
41
cache_retention_days: int = 7
42
+
status_indicator: bool = True
42
43
base_dir: Path = DEFAULT_BASE_DIR
43
44
44
45
@property
···
95
96
config.cache_retention_days = int(data["cache_retention_days"])
96
97
except (ValueError, TypeError):
97
98
pass
99
+
if "status_indicator" in data:
100
+
config.status_indicator = bool(data["status_indicator"])
98
101
99
102
return config
100
103
···
112
115
"sync_retry_delays": config.sync_retry_delays,
113
116
"sync_max_retries": config.sync_max_retries,
114
117
"cache_retention_days": config.cache_retention_days,
118
+
"status_indicator": config.status_indicator,
115
119
}
116
120
117
121
config_path = config.config_path
+56
src/solstone_tmux/indicator.py
+56
src/solstone_tmux/indicator.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Tmux status-left indicator for observer and sync state."""
5
+
6
+
import logging
7
+
8
+
from .capture import run_tmux_command
9
+
10
+
logger = logging.getLogger(__name__)
11
+
12
+
_original_status_left: str | None = None
13
+
_SENTINEL = "#{?@solstone,"
14
+
_INDICATOR_FMT = (
15
+
"#{?@solstone,"
16
+
"#{?#{==:#{@solstone},syncing},#[fg=yellow]☼#[default],#[fg=colour245]☼#[default]},"
17
+
"}"
18
+
)
19
+
20
+
21
+
def install() -> None:
22
+
"""Install the status indicator into tmux's status-left."""
23
+
global _original_status_left
24
+
25
+
status_left = run_tmux_command(["show", "-gv", "status-left"])
26
+
if status_left is None:
27
+
logger.warning("Unable to install tmux status indicator: tmux unavailable")
28
+
return
29
+
30
+
status_left = status_left.rstrip("\n")
31
+
if _SENTINEL in status_left:
32
+
return
33
+
34
+
_original_status_left = status_left
35
+
new_value = f"{_INDICATOR_FMT}{status_left}"
36
+
run_tmux_command(["set", "-g", "status-left", new_value])
37
+
run_tmux_command(["set", "-g", "@solstone", "observing"])
38
+
39
+
40
+
def update(syncing: bool) -> None:
41
+
"""Update the indicator state user variable."""
42
+
value = "syncing" if syncing else "observing"
43
+
run_tmux_command(["set", "-g", "@solstone", value])
44
+
45
+
46
+
def remove() -> None:
47
+
"""Remove the status indicator and restore the original status-left."""
48
+
global _original_status_left
49
+
50
+
if _original_status_left is None:
51
+
run_tmux_command(["set", "-g", "@solstone", ""])
52
+
return
53
+
54
+
run_tmux_command(["set", "-g", "status-left", _original_status_left])
55
+
run_tmux_command(["set", "-g", "@solstone", ""])
56
+
_original_status_left = None
+8
src/solstone_tmux/observer.py
+8
src/solstone_tmux/observer.py
···
20
20
21
21
from .capture import TmuxCapture, write_captures_jsonl
22
22
from .config import Config
23
+
from . import indicator
23
24
from .streams import stream_name
24
25
from .sync import SyncService
25
26
from .upload import UploadClient
···
183
184
platform=PLATFORM,
184
185
stream=self.stream,
185
186
)
187
+
if self.config.status_indicator and self._sync:
188
+
indicator.update(self._sync.is_connected)
186
189
187
190
async def main_loop(self):
188
191
"""Run the capture loop with background sync."""
···
222
225
if self._client:
223
226
self._client.stop()
224
227
self._client = None
228
+
if self.config.status_indicator:
229
+
indicator.remove()
225
230
226
231
227
232
async def async_run(config: Config) -> int:
···
240
245
if not observer.setup():
241
246
logger.error("Tmux observer setup failed")
242
247
return 1
248
+
249
+
if config.status_indicator:
250
+
indicator.install()
243
251
244
252
try:
245
253
await observer.main_loop()
+5
src/solstone_tmux/sync.py
+5
src/solstone_tmux/sync.py
···
77
77
self._running = False
78
78
self._trigger.set()
79
79
80
+
@property
81
+
def is_connected(self) -> bool:
82
+
"""Whether sync is connected (server configured and circuit closed)."""
83
+
return bool(self._config.server_url) and not self._circuit_open
84
+
80
85
async def run(self) -> None:
81
86
"""Main sync loop — waits for triggers, then syncs."""
82
87
while self._running:
+17
tests/test_config.py
+17
tests/test_config.py
···
71
71
72
72
loaded = load_config(tmp_path)
73
73
assert loaded.cache_retention_days == 7
74
+
75
+
def test_status_indicator_roundtrip(self, tmp_path: Path):
76
+
config = Config(base_dir=tmp_path)
77
+
config.status_indicator = False
78
+
save_config(config)
79
+
80
+
loaded = load_config(tmp_path)
81
+
assert loaded.status_indicator is False
82
+
83
+
def test_status_indicator_default(self, tmp_path: Path):
84
+
"""Existing configs without status_indicator default to True."""
85
+
config_dir = tmp_path / "config"
86
+
config_dir.mkdir(parents=True)
87
+
(config_dir / "config.json").write_text('{"server_url": "http://test"}')
88
+
89
+
loaded = load_config(tmp_path)
90
+
assert loaded.status_indicator is True
+128
tests/test_indicator.py
+128
tests/test_indicator.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
import pytest
5
+
6
+
from solstone_tmux import indicator
7
+
8
+
9
+
@pytest.fixture(autouse=True)
10
+
def reset_original_status_left():
11
+
indicator._original_status_left = None
12
+
yield
13
+
indicator._original_status_left = None
14
+
15
+
16
+
def test_install_saves_and_prepends(monkeypatch):
17
+
calls = []
18
+
19
+
def fake_run_tmux_command(args):
20
+
calls.append(args)
21
+
if args == ["show", "-gv", "status-left"]:
22
+
return "existing-status\n"
23
+
return None
24
+
25
+
monkeypatch.setattr(indicator, "run_tmux_command", fake_run_tmux_command)
26
+
27
+
indicator.install()
28
+
29
+
assert calls == [
30
+
["show", "-gv", "status-left"],
31
+
["set", "-g", "status-left", indicator._INDICATOR_FMT + "existing-status"],
32
+
["set", "-g", "@solstone", "observing"],
33
+
]
34
+
assert indicator._original_status_left == "existing-status"
35
+
36
+
37
+
def test_install_idempotent(monkeypatch):
38
+
calls = []
39
+
indicator._original_status_left = "keep-me"
40
+
41
+
def fake_run_tmux_command(args):
42
+
calls.append(args)
43
+
if args == ["show", "-gv", "status-left"]:
44
+
return indicator._SENTINEL + "existing-status\n"
45
+
return None
46
+
47
+
monkeypatch.setattr(indicator, "run_tmux_command", fake_run_tmux_command)
48
+
49
+
indicator.install()
50
+
51
+
assert calls == [["show", "-gv", "status-left"]]
52
+
assert indicator._original_status_left == "keep-me"
53
+
54
+
55
+
def test_install_tmux_unavailable(monkeypatch):
56
+
calls = []
57
+
58
+
def fake_run_tmux_command(args):
59
+
calls.append(args)
60
+
return None
61
+
62
+
monkeypatch.setattr(indicator, "run_tmux_command", fake_run_tmux_command)
63
+
64
+
indicator.install()
65
+
66
+
assert calls == [["show", "-gv", "status-left"]]
67
+
assert indicator._original_status_left is None
68
+
69
+
70
+
def test_update_syncing(monkeypatch):
71
+
calls = []
72
+
73
+
def fake_run_tmux_command(args):
74
+
calls.append(args)
75
+
return None
76
+
77
+
monkeypatch.setattr(indicator, "run_tmux_command", fake_run_tmux_command)
78
+
79
+
indicator.update(True)
80
+
81
+
assert calls == [["set", "-g", "@solstone", "syncing"]]
82
+
83
+
84
+
def test_update_observing(monkeypatch):
85
+
calls = []
86
+
87
+
def fake_run_tmux_command(args):
88
+
calls.append(args)
89
+
return None
90
+
91
+
monkeypatch.setattr(indicator, "run_tmux_command", fake_run_tmux_command)
92
+
93
+
indicator.update(False)
94
+
95
+
assert calls == [["set", "-g", "@solstone", "observing"]]
96
+
97
+
98
+
def test_remove_restores(monkeypatch):
99
+
calls = []
100
+
indicator._original_status_left = "my-original"
101
+
102
+
def fake_run_tmux_command(args):
103
+
calls.append(args)
104
+
return None
105
+
106
+
monkeypatch.setattr(indicator, "run_tmux_command", fake_run_tmux_command)
107
+
108
+
indicator.remove()
109
+
110
+
assert calls == [
111
+
["set", "-g", "status-left", "my-original"],
112
+
["set", "-g", "@solstone", ""],
113
+
]
114
+
assert indicator._original_status_left is None
115
+
116
+
117
+
def test_remove_no_original(monkeypatch):
118
+
calls = []
119
+
120
+
def fake_run_tmux_command(args):
121
+
calls.append(args)
122
+
return None
123
+
124
+
monkeypatch.setattr(indicator, "run_tmux_command", fake_run_tmux_command)
125
+
126
+
indicator.remove()
127
+
128
+
assert calls == [["set", "-g", "@solstone", ""]]
+26
tests/test_sync.py
+26
tests/test_sync.py
···
309
309
await sync._cleanup_synced_segments()
310
310
311
311
assert not (captures / "20260101" / "archon.tmux" / "120000_300").exists()
312
+
313
+
314
+
class TestSyncServiceConnected:
315
+
"""Test is_connected property."""
316
+
317
+
def test_connected_when_server_and_circuit_closed(self, tmp_path: Path):
318
+
config = Config(base_dir=tmp_path, server_url="http://localhost:5015")
319
+
config.ensure_dirs()
320
+
client = UploadClient(config)
321
+
sync = SyncService(config, client)
322
+
assert sync.is_connected is True
323
+
324
+
def test_disconnected_when_no_server(self, tmp_path: Path):
325
+
config = Config(base_dir=tmp_path)
326
+
config.ensure_dirs()
327
+
client = UploadClient(config)
328
+
sync = SyncService(config, client)
329
+
assert sync.is_connected is False
330
+
331
+
def test_disconnected_when_circuit_open(self, tmp_path: Path):
332
+
config = Config(base_dir=tmp_path, server_url="http://localhost:5015")
333
+
config.ensure_dirs()
334
+
client = UploadClient(config)
335
+
sync = SyncService(config, client)
336
+
sync._circuit_open = True
337
+
assert sync.is_connected is False