personal memory agent
Add task queue counts to supervisor status and sol top display
Enrich supervisor.status event with per-command queue depths via new
TaskQueue.collect_queue_counts() method. Update sol top to consume
queue counts from status polls, providing immediate queue visibility
on fresh connect instead of waiting for real-time queue change events.
Display queued-but-not-running commands in the tasks section when no
task rows are visible (e.g. "queued: dream ×2, indexer ×1").
+38
-4
+11
think/supervisor.py
+11
think/supervisor.py
···
385
385
)
386
386
return tasks
387
387
388
+
def collect_queue_counts(self) -> dict[str, int]:
389
+
"""Snapshot per-command queue depths for status reporting."""
390
+
with self._lock:
391
+
return {
392
+
cmd_name: len(queue)
393
+
for cmd_name, queue in self._queues.items()
394
+
if queue
395
+
}
396
+
388
397
389
398
# Global task queue instance (initialized in main())
390
399
_task_queue: TaskQueue | None = None
···
760
769
761
770
# Running tasks
762
771
tasks = _task_queue.collect_task_status() if _task_queue else []
772
+
queues = _task_queue.collect_queue_counts() if _task_queue else {}
763
773
764
774
# Stale heartbeats
765
775
stale = check_health()
···
768
778
"services": services,
769
779
"crashed": crashed,
770
780
"tasks": tasks,
781
+
"queues": queues,
771
782
"stale_heartbeats": stale,
772
783
}
773
784
+27
-4
think/top.py
+27
-4
think/top.py
···
55
55
self.cpu_cache = {} # Maps pid -> last cpu_percent value
56
56
self.cpu_procs = {} # Maps pid -> Process object for cpu tracking
57
57
self.running_tasks = {} # Maps ref -> task info from logs tract
58
-
self.command_queues = (
59
-
{}
60
-
) # Maps command_name -> queued count from supervisor.queue
58
+
self.command_queues = {} # Maps command_name -> queued count
61
59
self.event_queue: queue.Queue = queue.Queue() # Callosum events for main loop
62
60
self.active_notifications = {} # Maps service_name -> notification_id
63
61
self.crash_history = {} # Maps service_name -> [crash_timestamps]
···
213
211
if event == "status":
214
212
self.services = message.get("services", [])
215
213
self.crashed = message.get("crashed", [])
214
+
# Seed queue counts from status (fills gaps on fresh connect)
215
+
queues = message.get("queues")
216
+
if queues is not None:
217
+
self.command_queues = dict(queues)
216
218
217
219
# Poll CPU for current services and tasks
218
220
all_pids = [svc["pid"] for svc in self.services]
···
603
605
output.append(t.bold + header + t.normal)
604
606
605
607
if not tasks_only:
606
-
output.append(t.dim + " -" + t.normal)
608
+
# Show queued commands even when no tasks are running
609
+
queued_only = {
610
+
cmd: count for cmd, count in self.command_queues.items() if count > 0
611
+
}
612
+
if queued_only:
613
+
parts = [
614
+
f"{cmd} ×{count}" for cmd, count in sorted(queued_only.items())
615
+
]
616
+
output.append(t.dim + " queued: " + ", ".join(parts) + t.normal)
617
+
else:
618
+
output.append(t.dim + " -" + t.normal)
607
619
return output
608
620
609
621
# Task rows (sorted by start time, oldest first)
···
624
636
625
637
line = f" {name:<15} {pid:<8} {runtime:<12} {memory:>7} {cpu:>5} {log_age:>5} "
626
638
output.append(line + log_color + log_display + t.normal)
639
+
640
+
# Show queued commands not visible as running tasks
641
+
visible_commands = {task["name"] for task in tasks_only}
642
+
queued_only = {
643
+
cmd: count
644
+
for cmd, count in self.command_queues.items()
645
+
if cmd not in visible_commands and count > 0
646
+
}
647
+
if queued_only:
648
+
parts = [f"{cmd} ×{count}" for cmd, count in sorted(queued_only.items())]
649
+
output.append(t.dim + " queued: " + ", ".join(parts) + t.normal)
627
650
628
651
return output
629
652