#!/usr/bin/env python3 """ Side-chat work digest — INTEGRATOR minimal version (agent 4 of 5 never delivered its digest/throttle design within the window). Purpose: stop the per-message ✅ COMPLETED relay flood. COMPLETED hits are batched here and emitted as ONE periodic digest instead of N main-chat messages. ALERT / BLOCKER / DECISION still relay individually via siphon() — urgency is never batched. Interface (what agent 4's full design should remain compatible with): - buffer = DigestBuffer(max_items=20, max_age_s=3600) - buffer.add(hit) -> None - buffer.flush() -> Optional[str] (formatted digest, clears buffer) - flush_digest() -> Optional[str] (module-level singleton convenience) Reversible: to restore per-message COMPLETED relays, route COMPLETED back through siphon() in monitor.py and ignore this module. """ import time from typing import List, Optional try: from detect import SiphonHit except ImportError: # pragma: no cover SiphonHit = object class DigestBuffer: """Batch COMPLETED hits; flush() renders one digest message.""" def __init__(self, max_items: int = 20, max_age_s: int = 3600): self.max_items = max_items self.max_age_s = max_age_s self._items: List[tuple] = [] # (ts, SiphonHit) def add(self, hit) -> None: now = time.time() # Prune items older than max_age_s on every add (bounded memory). self._items = [(ts, h) for ts, h in self._items if now - ts < self.max_age_s] self._items.append((now, hit)) # Bound the buffer; oldest evicted first. self._items = self._items[-self.max_items:] def __len__(self) -> int: return len(self._items) def flush(self) -> Optional[str]: """Render and clear. Returns None when there's nothing to digest.""" if not self._items: return None lines = ["📦 [DIGEST] completions from side chats " f"({len(self._items)} item(s))"] for _, hit in self._items: author = getattr(hit, "author", "") or "?" url = f"https://muse.ai/thread/{hit.thread_id}" lines.append(f"• {hit.summary} — {author} ({url})") self._items = [] return "\n".join(lines) # Module-level singleton: the monitor loop shares one buffer per process. _default_buffer = DigestBuffer() def get_buffer() -> DigestBuffer: return _default_buffer def flush_digest() -> Optional[str]: """Flush the process-wide digest buffer. None if empty.""" return _default_buffer.flush()