personal memory agent
Align tmux observer output with screen.jsonl format
- Output per-session files: tmux_{session}_screen.jsonl
- Entry format now matches screen.jsonl (frame_id, timestamp, analysis, etc.)
- Add tmux category with formatter for markdown rendering (strips ANSI)
- Existing *_screen.jsonl pattern handles indexing automatically
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
+207
-43
+3
observe/categories/tmux.json
+3
observe/categories/tmux.json
+68
observe/categories/tmux.py
+68
observe/categories/tmux.py
···
1
+
"""Formatter for tmux category content.
2
+
3
+
Renders tmux terminal capture to markdown with pane contents.
4
+
"""
5
+
6
+
import re
7
+
from typing import Any
8
+
9
+
# ANSI escape code pattern
10
+
ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]")
11
+
12
+
13
+
def _strip_ansi(text: str) -> str:
14
+
"""Remove ANSI escape codes from text."""
15
+
return ANSI_ESCAPE.sub("", text)
16
+
17
+
18
+
def format(content: Any, context: dict) -> str:
19
+
"""Format tmux capture to markdown.
20
+
21
+
Args:
22
+
content: Tmux capture dict with session, window, panes
23
+
context: Dict with frame, file_path, timestamp_str
24
+
25
+
Returns:
26
+
Formatted markdown string
27
+
"""
28
+
if not isinstance(content, dict):
29
+
return ""
30
+
31
+
lines = []
32
+
33
+
# Session and window header
34
+
session = content.get("session", "unknown")
35
+
window = content.get("window", {})
36
+
window_name = window.get("name", "unknown") if isinstance(window, dict) else "unknown"
37
+
38
+
lines.append(f"**Tmux** ({session}:{window_name})")
39
+
lines.append("")
40
+
41
+
# Pane contents
42
+
panes = content.get("panes", [])
43
+
for pane in panes:
44
+
if not isinstance(pane, dict):
45
+
continue
46
+
47
+
pane_content = pane.get("content", "")
48
+
if not pane_content:
49
+
continue
50
+
51
+
# Strip ANSI codes for clean markdown
52
+
clean_content = _strip_ansi(pane_content).rstrip()
53
+
if not clean_content:
54
+
continue
55
+
56
+
# Label pane if multiple panes
57
+
if len(panes) > 1:
58
+
pane_idx = pane.get("index", 0)
59
+
active = " (active)" if pane.get("active") else ""
60
+
lines.append(f"**Pane {pane_idx}{active}:**")
61
+
lines.append("")
62
+
63
+
lines.append("```")
64
+
lines.append(clean_content)
65
+
lines.append("```")
66
+
lines.append("")
67
+
68
+
return "\n".join(lines)
+86
-42
observe/tmux/observer.py
+86
-42
observe/tmux/observer.py
···
3
3
Tmux terminal observer for journaling.
4
4
5
5
Polls tmux for active sessions and captures terminal content from the active
6
-
window's panes. Creates 5-minute segments with JSONL output.
6
+
window's panes. Outputs tmux_{session}_screen.jsonl files in screen.jsonl-
7
+
compatible format for unified formatting and indexing.
7
8
"""
8
9
9
10
import argparse
···
247
248
return hashlib.md5(content.encode()).hexdigest()
248
249
249
250
def result_to_dict(self, result: CaptureResult, ts: float) -> dict:
250
-
"""Convert CaptureResult to JSON-serializable dict."""
251
+
"""Convert CaptureResult to JSON-serializable dict.
252
+
253
+
Output format matches screen.jsonl structure from observe-describe:
254
+
- frame_id: Capture sequence number
255
+
- timestamp: Seconds since segment start
256
+
- requests: Empty list (no AI processing)
257
+
- analysis: Category info with templated visual_description
258
+
- tmux: All terminal-specific data
259
+
"""
251
260
self.capture_id += 1
261
+
262
+
# Calculate relative timestamp from segment start
263
+
relative_ts = ts - self.start_at
264
+
265
+
# Build visual description from tmux info
266
+
pane_count = len(result.panes)
267
+
pane_word = "pane" if pane_count == 1 else "panes"
268
+
visual_description = (
269
+
f"Terminal session '{result.session}' with {pane_count} {pane_word} "
270
+
f"in window '{result.window.name}'"
271
+
)
272
+
252
273
return {
253
-
"id": self.capture_id,
254
-
"ts": ts,
255
-
"session": result.session,
256
-
"window": {
257
-
"id": result.window.id,
258
-
"index": result.window.index,
259
-
"name": result.window.name,
274
+
"frame_id": self.capture_id,
275
+
"timestamp": relative_ts,
276
+
"requests": [],
277
+
"analysis": {
278
+
"visual_description": visual_description,
279
+
"primary": "tmux",
280
+
"secondary": "none",
281
+
"overlap": False,
282
+
},
283
+
"tmux": {
284
+
"session": result.session,
285
+
"window": {
286
+
"id": result.window.id,
287
+
"index": result.window.index,
288
+
"name": result.window.name,
289
+
},
290
+
"windows": [
291
+
{"id": w.id, "index": w.index, "name": w.name, "active": w.active}
292
+
for w in result.windows
293
+
],
294
+
"panes": [
295
+
{
296
+
"id": p.id,
297
+
"index": p.index,
298
+
"left": p.left,
299
+
"top": p.top,
300
+
"width": p.width,
301
+
"height": p.height,
302
+
"active": p.active,
303
+
"content": p.content,
304
+
}
305
+
for p in result.panes
306
+
],
260
307
},
261
-
"windows": [
262
-
{"id": w.id, "index": w.index, "name": w.name, "active": w.active}
263
-
for w in result.windows
264
-
],
265
-
"panes": [
266
-
{
267
-
"id": p.id,
268
-
"index": p.index,
269
-
"left": p.left,
270
-
"top": p.top,
271
-
"width": p.width,
272
-
"height": p.height,
273
-
"active": p.active,
274
-
"content": p.content,
275
-
}
276
-
for p in result.panes
277
-
],
278
308
}
279
309
280
310
def poll_and_capture(self) -> list[dict]:
···
311
341
return new_captures
312
342
313
343
def handle_boundary(self):
314
-
"""Write accumulated captures to segment JSONL and reset."""
344
+
"""Write accumulated captures to segment JSONL files and reset.
345
+
346
+
Writes one file per session: tmux_{session}_screen.jsonl
347
+
Format matches screen.jsonl for unified formatting/indexing.
348
+
"""
315
349
if not self.captures:
316
350
logger.info("No captures in segment, skipping write")
317
351
self.reset_segment()
···
327
361
segment_dir = day_path(date_part) / segment_key
328
362
segment_dir.mkdir(parents=True, exist_ok=True)
329
363
330
-
output_path = segment_dir / "tmux.jsonl"
364
+
# Group captures by session
365
+
by_session: dict[str, list[dict]] = {}
366
+
for capture in self.captures:
367
+
session = capture.get("tmux", {}).get("session", "unknown")
368
+
if session not in by_session:
369
+
by_session[session] = []
370
+
by_session[session].append(capture)
331
371
332
-
# Write JSONL: metadata header + captures
333
-
with open(output_path, "w") as f:
334
-
# Header with metadata
335
-
header = {
336
-
"captures": len(self.captures),
337
-
"sessions": sorted(self.sessions_seen),
338
-
}
339
-
f.write(json.dumps(header) + "\n")
372
+
# Write one file per session
373
+
files_written = []
374
+
for session, captures in by_session.items():
375
+
# Sanitize session name for filename
376
+
safe_session = session.replace("/", "_").replace(" ", "_")
377
+
filename = f"tmux_{safe_session}_screen.jsonl"
378
+
output_path = segment_dir / filename
379
+
380
+
with open(output_path, "w") as f:
381
+
# Header matching screen.jsonl format
382
+
header = {"raw": filename}
383
+
f.write(json.dumps(header) + "\n")
340
384
341
-
# Write each capture
342
-
for capture in self.captures:
343
-
f.write(json.dumps(capture) + "\n")
385
+
# Write each capture
386
+
for capture in captures:
387
+
f.write(json.dumps(capture) + "\n")
344
388
345
-
logger.info(
346
-
f"Wrote {len(self.captures)} captures to {output_path}"
347
-
)
389
+
files_written.append(filename)
390
+
logger.info(f"Wrote {len(captures)} captures to {output_path}")
348
391
349
392
# Emit event
350
393
if self.callosum:
···
354
397
segment=segment_key,
355
398
captures=len(self.captures),
356
399
sessions=sorted(self.sessions_seen),
400
+
files=files_written,
357
401
)
358
402
359
403
self.reset_segment()
+50
-1
tests/test_screen_formatter.py
+50
-1
tests/test_screen_formatter.py
···
421
421
"media",
422
422
"gaming",
423
423
"productivity",
424
+
"tmux",
424
425
]
425
426
for cat in expected:
426
427
assert cat in CATEGORIES, f"Expected category {cat} not found"
427
-
assert len(CATEGORIES) == 9
428
+
assert len(CATEGORIES) == 10
429
+
430
+
431
+
def test_tmux_formatter_output():
432
+
"""Test that tmux formatter produces expected markdown."""
433
+
from observe.categories.tmux import format as tmux_format
434
+
435
+
content = {
436
+
"session": "main",
437
+
"window": {"id": "@1", "index": 0, "name": "bash"},
438
+
"panes": [
439
+
{
440
+
"id": "%1",
441
+
"index": 0,
442
+
"active": True,
443
+
"content": "$ ls -la\n\x1b[32mtotal 42\x1b[0m\ndrwxr-xr-x 2 user",
444
+
},
445
+
],
446
+
}
447
+
448
+
result = tmux_format(content, {})
449
+
450
+
assert "**Tmux** (main:bash)" in result
451
+
assert "```" in result
452
+
# ANSI codes should be stripped
453
+
assert "\x1b[32m" not in result
454
+
assert "total 42" in result
455
+
assert "$ ls -la" in result
456
+
457
+
458
+
def test_tmux_formatter_multiple_panes():
459
+
"""Test tmux formatter labels multiple panes."""
460
+
from observe.categories.tmux import format as tmux_format
461
+
462
+
content = {
463
+
"session": "dev",
464
+
"window": {"name": "work"},
465
+
"panes": [
466
+
{"index": 0, "active": True, "content": "pane zero"},
467
+
{"index": 1, "active": False, "content": "pane one"},
468
+
],
469
+
}
470
+
471
+
result = tmux_format(content, {})
472
+
473
+
assert "**Pane 0 (active):**" in result
474
+
assert "**Pane 1:**" in result
475
+
assert "pane zero" in result
476
+
assert "pane one" in result