personal memory agent
Replace sol callosum server-start CLI with send and listen subcommands
The server is always started in-process by supervisor, so the standalone
server-start CLI was unused. Replace it with two useful subcommands:
- listen: stream events as JSONL with --tract/--event/--pretty filters
(also the default when running bare `sol callosum`)
- send: emit messages via positional args, JSON string, or stdin/heredoc
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+349
-11
+17
-1
docs/CALLOSUM.md
+17
-1
docs/CALLOSUM.md
···
175
175
callosum_send("observe", "described", day="20251102", segment="143045_300")
176
176
```
177
177
178
-
**`CallosumServer`** - Broadcast server (run via `sol callosum` or supervisor)
178
+
**`CallosumServer`** - Broadcast server (started in-process by supervisor)
179
179
180
180
### Convey Integration
181
181
···
183
183
- `apps.events` - Server-side event handlers via `@on_event` decorator
184
184
185
185
See [APPS.md](APPS.md) for app event handler patterns.
186
+
187
+
### CLI Tools
188
+
189
+
**`sol callosum`** / **`sol callosum listen`** - Listen to events on the message bus
190
+
```bash
191
+
sol callosum # Stream all events as JSONL
192
+
sol callosum listen --tract cortex # Filter to cortex tract
193
+
sol callosum listen --event finish -p # Pretty-print finish events
194
+
```
195
+
196
+
**`sol callosum send`** - Send a message to the bus
197
+
```bash
198
+
sol callosum send observe described day=20250101 segment=143045_300
199
+
sol callosum send '{"tract":"test","event":"ping","data":42}'
200
+
echo '{"tract":"test","event":"ping"}' | sol callosum send
201
+
```
186
202
187
203
---
188
204
+1
-1
muse/help.md
+1
-1
muse/help.md
···
28
28
- `sol detect-created` - Detect newly created content artifacts.
29
29
- `sol top` - Show runtime/service activity status.
30
30
- `sol health` - Show service health status. Use `sol health logs` to view service logs.
31
-
- `sol callosum` - Interact with Callosum message bus tooling.
31
+
- `sol callosum [listen] [--tract NAME] [--event NAME] [-p]` - Listen to Callosum events. `sol callosum send <tract> <event> [key=value ...]` to send.
32
32
- `sol streams` - Manage or inspect stream-related state.
33
33
- `sol journal-stats` - Show journal statistics.
34
34
- `sol config` - Inspect or manage configuration.
+171
tests/test_callosum.py
+171
tests/test_callosum.py
···
269
269
270
270
# Should fail gracefully (no server listening)
271
271
assert result is False
272
+
273
+
274
+
# --- CLI helper tests ---
275
+
276
+
277
+
class TestParseValue:
278
+
"""Tests for _parse_value auto-type detection."""
279
+
280
+
def test_integer(self):
281
+
from think.callosum import _parse_value
282
+
283
+
assert _parse_value("42") == 42
284
+
285
+
def test_float(self):
286
+
from think.callosum import _parse_value
287
+
288
+
assert _parse_value("3.14") == 3.14
289
+
290
+
def test_boolean_true(self):
291
+
from think.callosum import _parse_value
292
+
293
+
assert _parse_value("true") is True
294
+
295
+
def test_boolean_false(self):
296
+
from think.callosum import _parse_value
297
+
298
+
assert _parse_value("false") is False
299
+
300
+
def test_null(self):
301
+
from think.callosum import _parse_value
302
+
303
+
assert _parse_value("null") is None
304
+
305
+
def test_plain_string(self):
306
+
from think.callosum import _parse_value
307
+
308
+
assert _parse_value("hello") == "hello"
309
+
310
+
def test_string_with_spaces(self):
311
+
from think.callosum import _parse_value
312
+
313
+
assert _parse_value("hello world") == "hello world"
314
+
315
+
def test_json_array(self):
316
+
from think.callosum import _parse_value
317
+
318
+
assert _parse_value("[1,2,3]") == [1, 2, 3]
319
+
320
+
321
+
class TestParseKvFields:
322
+
"""Tests for _parse_kv_fields key=value parsing."""
323
+
324
+
def test_basic_fields(self):
325
+
from think.callosum import _parse_kv_fields
326
+
327
+
result = _parse_kv_fields(["day=20250101", "count=5", "active=true"])
328
+
assert result == {"day": 20250101, "count": 5, "active": True}
329
+
330
+
def test_empty_list(self):
331
+
from think.callosum import _parse_kv_fields
332
+
333
+
assert _parse_kv_fields([]) == {}
334
+
335
+
def test_value_with_equals(self):
336
+
from think.callosum import _parse_kv_fields
337
+
338
+
# Value containing '=' should keep everything after first '='
339
+
result = _parse_kv_fields(["expr=a=b"])
340
+
assert result == {"expr": "a=b"}
341
+
342
+
def test_missing_equals_exits(self):
343
+
from think.callosum import _parse_kv_fields
344
+
345
+
with pytest.raises(SystemExit):
346
+
_parse_kv_fields(["no_equals_here"])
347
+
348
+
349
+
class TestParseJsonMessage:
350
+
"""Tests for _parse_json_message validation."""
351
+
352
+
def test_valid_json(self):
353
+
from think.callosum import _parse_json_message
354
+
355
+
result = _parse_json_message('{"tract":"test","event":"ping","data":1}')
356
+
assert result == {"tract": "test", "event": "ping", "data": 1}
357
+
358
+
def test_missing_tract(self):
359
+
from think.callosum import _parse_json_message
360
+
361
+
with pytest.raises(SystemExit):
362
+
_parse_json_message('{"event":"ping"}')
363
+
364
+
def test_missing_event(self):
365
+
from think.callosum import _parse_json_message
366
+
367
+
with pytest.raises(SystemExit):
368
+
_parse_json_message('{"tract":"test"}')
369
+
370
+
def test_invalid_json(self):
371
+
from think.callosum import _parse_json_message
372
+
373
+
with pytest.raises(SystemExit):
374
+
_parse_json_message("not json")
375
+
376
+
def test_json_array_rejected(self):
377
+
from think.callosum import _parse_json_message
378
+
379
+
with pytest.raises(SystemExit):
380
+
_parse_json_message("[1,2,3]")
381
+
382
+
383
+
class TestCmdSendInputModes:
384
+
"""Tests for _cmd_send input mode detection."""
385
+
386
+
def test_positional_mode(self):
387
+
"""Test tract event key=value positional syntax."""
388
+
from types import SimpleNamespace
389
+
390
+
from think.callosum import _cmd_send
391
+
392
+
args = SimpleNamespace(args=["test", "ping", "data=42"])
393
+
with patch("think.callosum.callosum_send", return_value=True) as mock_send:
394
+
_cmd_send(args)
395
+
mock_send.assert_called_once_with("test", "ping", data=42)
396
+
397
+
def test_json_arg_mode(self):
398
+
"""Test JSON string argument mode."""
399
+
from types import SimpleNamespace
400
+
401
+
from think.callosum import _cmd_send
402
+
403
+
args = SimpleNamespace(args=['{"tract":"test","event":"ping","n":1}'])
404
+
with patch("think.callosum.callosum_send", return_value=True) as mock_send:
405
+
_cmd_send(args)
406
+
mock_send.assert_called_once_with("test", "ping", n=1)
407
+
408
+
def test_stdin_mode(self, monkeypatch):
409
+
"""Test reading JSON from stdin."""
410
+
import io
411
+
from types import SimpleNamespace
412
+
413
+
from think.callosum import _cmd_send
414
+
415
+
args = SimpleNamespace(args=[])
416
+
fake_stdin = io.StringIO('{"tract":"test","event":"ping"}')
417
+
monkeypatch.setattr("think.callosum.sys.stdin", fake_stdin)
418
+
419
+
with patch("think.callosum.callosum_send", return_value=True) as mock_send:
420
+
_cmd_send(args)
421
+
mock_send.assert_called_once_with("test", "ping")
422
+
423
+
def test_too_few_positional_args_exits(self):
424
+
"""Test that a single positional arg (not JSON) exits with usage."""
425
+
from types import SimpleNamespace
426
+
427
+
from think.callosum import _cmd_send
428
+
429
+
args = SimpleNamespace(args=["only_one"])
430
+
with pytest.raises(SystemExit):
431
+
_cmd_send(args)
432
+
433
+
def test_send_failure_exits(self):
434
+
"""Test that failed send exits with code 1."""
435
+
from types import SimpleNamespace
436
+
437
+
from think.callosum import _cmd_send
438
+
439
+
args = SimpleNamespace(args=["test", "ping"])
440
+
with patch("think.callosum.callosum_send", return_value=False):
441
+
with pytest.raises(SystemExit):
442
+
_cmd_send(args)
+160
-9
think/callosum.py
+160
-9
think/callosum.py
···
11
11
import logging
12
12
import queue
13
13
import socket
14
+
import sys
14
15
import threading
15
16
import time
16
17
from pathlib import Path
···
437
438
return False
438
439
439
440
441
+
def _parse_value(value: str) -> Any:
442
+
"""Parse a string value, auto-detecting JSON types.
443
+
444
+
Tries json.loads first (handles numbers, booleans, null, arrays, objects).
445
+
Falls back to raw string.
446
+
"""
447
+
try:
448
+
return json.loads(value)
449
+
except (json.JSONDecodeError, ValueError):
450
+
return value
451
+
452
+
453
+
def _parse_kv_fields(pairs: list[str]) -> dict[str, Any]:
454
+
"""Parse key=value pairs into a dict with auto-typed values."""
455
+
fields: dict[str, Any] = {}
456
+
for pair in pairs:
457
+
if "=" not in pair:
458
+
print(
459
+
f"Error: Invalid field '{pair}' (expected key=value)", file=sys.stderr
460
+
)
461
+
sys.exit(1)
462
+
key, value = pair.split("=", 1)
463
+
fields[key] = _parse_value(value)
464
+
return fields
465
+
466
+
467
+
def _parse_json_message(text: str) -> dict[str, Any]:
468
+
"""Parse a JSON string into a message dict, validating required fields."""
469
+
try:
470
+
message = json.loads(text)
471
+
except json.JSONDecodeError as e:
472
+
print(f"Error: Invalid JSON: {e}", file=sys.stderr)
473
+
sys.exit(1)
474
+
475
+
if not isinstance(message, dict):
476
+
print("Error: JSON must be an object", file=sys.stderr)
477
+
sys.exit(1)
478
+
479
+
if "tract" not in message or "event" not in message:
480
+
print("Error: JSON must contain 'tract' and 'event' fields", file=sys.stderr)
481
+
sys.exit(1)
482
+
483
+
return message
484
+
485
+
486
+
def _cmd_listen(args) -> None:
487
+
"""Listen to Callosum events and print to stdout."""
488
+
conn = CallosumConnection()
489
+
490
+
def on_message(message: dict[str, Any]) -> None:
491
+
# Apply filters
492
+
if args.tract and message.get("tract") != args.tract:
493
+
return
494
+
if args.event and message.get("event") != args.event:
495
+
return
496
+
497
+
if args.pretty:
498
+
print(json.dumps(message, indent=2))
499
+
else:
500
+
print(json.dumps(message), flush=True)
501
+
502
+
conn.start(callback=on_message)
503
+
504
+
try:
505
+
# Block until Ctrl+C
506
+
import signal
507
+
508
+
signal.pause()
509
+
except KeyboardInterrupt:
510
+
pass
511
+
finally:
512
+
conn.stop()
513
+
514
+
515
+
def _cmd_send(args) -> None:
516
+
"""Send a message to Callosum."""
517
+
positional = args.args or []
518
+
519
+
# Determine input mode:
520
+
# 1. First positional starts with '{' → JSON string arg
521
+
# 2. No positional args and stdin is not a TTY → read JSON from stdin
522
+
# 3. Otherwise → tract event [key=value ...] positional syntax
523
+
if positional and positional[0].lstrip().startswith("{"):
524
+
# JSON string argument
525
+
raw = " ".join(positional)
526
+
message = _parse_json_message(raw)
527
+
elif not positional and not sys.stdin.isatty():
528
+
# Read JSON from stdin (supports piping and heredoc)
529
+
raw = sys.stdin.read().strip()
530
+
if not raw:
531
+
print("Error: Empty input on stdin", file=sys.stderr)
532
+
sys.exit(1)
533
+
message = _parse_json_message(raw)
534
+
elif len(positional) >= 2:
535
+
# Positional: tract event [key=value ...]
536
+
tract, event = positional[0], positional[1]
537
+
fields = _parse_kv_fields(positional[2:])
538
+
message = {"tract": tract, "event": event, **fields}
539
+
else:
540
+
print(
541
+
"Usage: sol callosum send <tract> <event> [key=value ...]\n"
542
+
' sol callosum send \'{"tract":"x","event":"y",...}\'\n'
543
+
" echo '{...}' | sol callosum send",
544
+
file=sys.stderr,
545
+
)
546
+
sys.exit(1)
547
+
548
+
ok = callosum_send(message.pop("tract"), message.pop("event"), **message)
549
+
if ok:
550
+
print("Sent", file=sys.stderr)
551
+
else:
552
+
print("Failed to send (is callosum running?)", file=sys.stderr)
553
+
sys.exit(1)
554
+
555
+
440
556
def main() -> None:
441
-
"""CLI entry point."""
557
+
"""CLI entry point for Callosum message bus tools."""
442
558
import argparse
443
559
444
560
from think.utils import setup_cli
445
561
446
-
parser = argparse.ArgumentParser(description="Callosum message bus")
447
-
setup_cli(parser) # Handles logging setup based on -v/-d flags
562
+
parser = argparse.ArgumentParser(
563
+
description="Callosum message bus tools",
564
+
epilog="Run 'sol callosum' with no subcommand to listen to all events.",
565
+
)
566
+
subparsers = parser.add_subparsers(dest="subcommand")
448
567
449
-
server = CallosumServer()
568
+
# --- listen subcommand ---
569
+
listen_parser = subparsers.add_parser(
570
+
"listen", help="Listen to events on the message bus"
571
+
)
572
+
listen_parser.add_argument("--tract", help="Filter to a specific tract")
573
+
listen_parser.add_argument("--event", help="Filter to a specific event type")
574
+
listen_parser.add_argument(
575
+
"-p", "--pretty", action="store_true", help="Pretty-print JSON output"
576
+
)
450
577
451
-
try:
452
-
server.start()
453
-
except KeyboardInterrupt:
454
-
logger.info("Shutting down Callosum")
455
-
server.stop()
578
+
# --- send subcommand ---
579
+
send_parser = subparsers.add_parser(
580
+
"send",
581
+
help="Send a message to the bus",
582
+
epilog=(
583
+
"Examples:\n"
584
+
" sol callosum send observe described day=20250101 segment=143045_300\n"
585
+
' sol callosum send \'{"tract":"test","event":"ping"}\'\n'
586
+
" echo '{...}' | sol callosum send"
587
+
),
588
+
formatter_class=argparse.RawDescriptionHelpFormatter,
589
+
)
590
+
send_parser.add_argument(
591
+
"args", nargs="*", help="tract event [key=value ...] or JSON string"
592
+
)
593
+
594
+
args = setup_cli(parser)
595
+
596
+
if args.subcommand == "send":
597
+
_cmd_send(args)
598
+
else:
599
+
# Default: listen (both bare 'sol callosum' and 'sol callosum listen')
600
+
if not hasattr(args, "tract"):
601
+
args.tract = None
602
+
if not hasattr(args, "event"):
603
+
args.event = None
604
+
if not hasattr(args, "pretty"):
605
+
args.pretty = False
606
+
_cmd_listen(args)
456
607
457
608
458
609
if __name__ == "__main__":