personal memory agent
Add per-service headers to sol health logs output
When stdout is a TTY, print dim separator headers (── service ──) each
time the log output switches between services. In follow mode (-f),
track service changes and emit headers the same way. When piped, output
remains unchanged — no headers, no ANSI codes.
+78
+52
tests/test_logs_cli.py
+52
tests/test_logs_cli.py
···
172
172
assert "line 9" in output[1]
173
173
174
174
175
+
def test_collect_headers_on_tty(tmp_path, monkeypatch, capsys):
176
+
"""Service headers appear when stdout is a TTY."""
177
+
from think import logs_cli
178
+
179
+
day = datetime.now().strftime("%Y%m%d")
180
+
echo_lines = [
181
+
"2026-02-09T10:00:00 [echo:stdout] line a",
182
+
"2026-02-09T10:02:00 [echo:stdout] line c",
183
+
]
184
+
observer_lines = [
185
+
"2026-02-09T10:01:00 [observer:stdout] line b",
186
+
]
187
+
make_journal(tmp_path, day, {"echo": echo_lines, "observer": observer_lines})
188
+
monkeypatch.setenv("JOURNAL_PATH", str(tmp_path))
189
+
monkeypatch.setattr("sys.stdout.isatty", lambda: True)
190
+
191
+
logs_cli.collect_and_print(_args(c=5))
192
+
193
+
raw = capsys.readouterr().out
194
+
output = raw.strip().splitlines()
195
+
# Should have: header + line a, blank + header + line b, blank + header + line c
196
+
# That's 3 headers + 3 content lines + 2 blank lines = 8 lines total
197
+
assert any("── echo ──" in line for line in output)
198
+
assert any("── observer ──" in line for line in output)
199
+
# Content lines are still present
200
+
assert any("line a" in line for line in output)
201
+
assert any("line b" in line for line in output)
202
+
assert any("line c" in line for line in output)
203
+
204
+
205
+
def test_collect_no_headers_when_piped(tmp_path, monkeypatch, capsys):
206
+
"""No headers when stdout is not a TTY (piped)."""
207
+
from think import logs_cli
208
+
209
+
day = datetime.now().strftime("%Y%m%d")
210
+
echo_lines = [
211
+
"2026-02-09T10:00:00 [echo:stdout] line a",
212
+
]
213
+
observer_lines = [
214
+
"2026-02-09T10:01:00 [observer:stdout] line b",
215
+
]
216
+
make_journal(tmp_path, day, {"echo": echo_lines, "observer": observer_lines})
217
+
monkeypatch.setenv("JOURNAL_PATH", str(tmp_path))
218
+
monkeypatch.setattr("sys.stdout.isatty", lambda: False)
219
+
220
+
logs_cli.collect_and_print(_args(c=5))
221
+
222
+
output = capsys.readouterr().out.strip().splitlines()
223
+
assert len(output) == 2
224
+
assert not any("──" in line for line in output)
225
+
226
+
175
227
def test_filter_service(tmp_path, monkeypatch, capsys):
176
228
from think import logs_cli
177
229
+26
think/logs_cli.py
+26
think/logs_cli.py
···
25
25
26
26
from think.utils import get_journal, setup_cli
27
27
28
+
_DIM = "\033[2m"
29
+
_RESET = "\033[0m"
30
+
28
31
29
32
class LogLine(NamedTuple):
30
33
timestamp: datetime
···
32
35
stream: str
33
36
message: str
34
37
raw: str
38
+
39
+
40
+
def _service_header(service: str, use_color: bool) -> str:
41
+
header = f"── {service} ──"
42
+
if use_color:
43
+
return f"{_DIM}{header}{_RESET}"
44
+
return header
35
45
36
46
37
47
def parse_log_line(line: str) -> LogLine | None:
···
175
185
lines.sort(key=lambda line: line.timestamp)
176
186
if has_filters and args.c:
177
187
lines = lines[-args.c:]
188
+
use_color = sys.stdout.isatty()
189
+
last_service = None
178
190
for line in lines:
191
+
if use_color and line.service != last_service:
192
+
if last_service is not None:
193
+
print()
194
+
print(_service_header(line.service, use_color))
195
+
last_service = line.service
179
196
print(line.raw)
180
197
181
198
···
186
203
print("No health directory found.", file=sys.stderr)
187
204
return
188
205
206
+
last_service = None
207
+
use_color = sys.stdout.isatty()
189
208
tracked: dict[Path, tuple[Path | None, object]] = {}
190
209
191
210
def open_logs() -> None:
···
213
232
while line:
214
233
line = line.rstrip("\n")
215
234
if line:
235
+
parsed = parse_log_line(line)
236
+
current_service = parsed.service if parsed else None
237
+
if use_color and current_service and current_service != last_service:
238
+
if last_service is not None:
239
+
print(flush=True)
240
+
print(_service_header(current_service, use_color), flush=True)
241
+
last_service = current_service
216
242
print(line, flush=True)
217
243
line = fh.readline()
218
244