box/bin/siphon.py

#!/usr/bin/env python3
"""
Side-chat to main-chat work siphon — siphon action.

When detection fires, post a summary to main chat with:
  - Category badge
  - One-line summary (never full message text)
  - Link back to the source side chat thread
  - Confidence score (for transparency)

Safety:
  - Rate limited (max N siphons per hour per thread)
  - Never posts full message content
  - Respects opt-out registry
  - Deduplicates (same message_id never siphoned twice)
"""

import time
from dataclasses import dataclass, field
from typing import Callable, Optional

from detect import SiphonHit, is_opted_out


# --- Follow-up modulation ---
#
# Wire the follow-up modulation table into the siphon so each hit gets
# the right follow-up policy:
#   ALERT / BLOCKER / DECISION -> tracked, fast fuse for ALERT/BLOCKER
#   COMPLETED / MILESTONE      -> untracked (no nudge budget burned)
#
# modulate.py must be landed on bl before this runs (rollout step 1).
# If the import fails we degrade to the old behavior: post the summary
# with no follow-up tags (fail-closed toward visibility, not tracking).
try:
    from modulate import for_siphon_hit, render_tags
    _MODULATION_AVAILABLE = True
except ImportError:  # pragma: no cover - deploy keeps modulate.py present
    _MODULATION_AVAILABLE = False
    for_siphon_hit = None
    render_tags = None


def policy_for_hit(hit: SiphonHit):
    """Follow-up policy for a siphon hit, or None when untracked.

    COMPLETED / MILESTONE hits return None (post the summary, create no
    follow-up record). ALERT / BLOCKER / DECISION return a Policy whose
    tags render into the canonical bracket vocabulary.
    """
    if not _MODULATION_AVAILABLE:
        return None
    return for_siphon_hit(hit.category)


def is_tracked(hit: SiphonHit) -> bool:
    """True when this hit should create a follow-up record.

    Callers that route tracked posts through dm.py --expect-reply (so a
    dm_followup record is actually created) can use this to choose the
    post path. Untracked hits post as plain summaries.
    """
    return policy_for_hit(hit) is not None


# --- Rate limiting ---

@dataclass
class RateLimiter:
    max_per_hour: int = 5
    _timestamps: dict = field(default_factory=dict)  # thread_id -> [ts, ...]

    def allow(self, thread_id: str) -> bool:
        now = time.time()
        stamps = self._timestamps.get(thread_id, [])
        # Prune older than 1 hour
        stamps = [s for s in stamps if now - s < 3600]
        if len(stamps) >= self.max_per_hour:
            return False
        stamps.append(now)
        self._timestamps[thread_id] = stamps
        return True


# --- Deduplication ---

_siphoned_ids: set = set()


def already_siphoned(message_id: str) -> bool:
    return message_id in _siphoned_ids


def mark_siphoned(message_id: str):
    _siphoned_ids.add(message_id)


# --- Siphon action ---

CATEGORY_EMOJI = {
    "COMPLETED": "✅",
    "BLOCKER": "🚧",
    "DECISION": "❓",
    "ALERT": "🚨",
    "MILESTONE": "🎯",
}


def format_siphon(hit: SiphonHit, agent_name: str = "sidechat") -> str:
    """
    Format a siphon message for main chat.
    Never includes full message text — summary + link only.
    """
    emoji = CATEGORY_EMOJI.get(hit.category, "📋")
    thread_url = f"https://muse.ai/thread/{hit.thread_id}"
    return (
        f"{emoji} [{hit.category}] from {agent_name} side chat\n"
        f"{hit.summary}\n"
        f"→ {thread_url}\n"
        f"(confidence {hit.confidence:.0%})"
    )


def siphon(hit: SiphonHit,
           agent_name: str,
           post_to_main: Callable[[str], bool],
           limiter: Optional[RateLimiter] = None) -> bool:
    """
    Execute the siphon: post summary to main chat.

    post_to_main: callable that posts text to main chat, returns True on success.
    Returns True if siphoned, False if suppressed.
    """
    # Safety checks
    if is_opted_out(hit.thread_id):
        return False
    if already_siphoned(hit.message_id):
        return False

    lim = limiter or RateLimiter()
    if not lim.allow(hit.thread_id):
        return False

    text = format_siphon(hit, agent_name)

    # Follow-up modulation: tracked hits (ALERT/BLOCKER/DECISION) get
    # the canonical follow-up tags appended — [reply:expected],
    # [reply:timeout=N], [reply:nudges=N], [reply:escalate=X], and
    # [input:siphon] for the audit trail. Untracked hits
    # (COMPLETED/MILESTONE) post as plain summaries.
    policy = policy_for_hit(hit)
    if policy is not None:
        text = text + "\n" + render_tags(policy)

    ok = post_to_main(text)
    if ok:
        mark_siphoned(hit.message_id)
    return ok