operator
153 lines
4.9 KiB
Python
153 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Side-chat to main-chat work siphon — detection rules.
|
|
|
|
Monitors side chat messages and identifies "siphon-worthy" content:
|
|
work that should surface in main chat for visibility.
|
|
|
|
Categories:
|
|
COMPLETED - work finished, results ready
|
|
BLOCKER - something is stuck, needs intervention
|
|
DECISION - a decision is needed from the user/operator
|
|
ALERT - health/security/urgency signal
|
|
MILESTONE - significant progress checkpoint
|
|
|
|
Detection is purely pattern-based (raw Python, no AI).
|
|
Each rule returns (category, confidence, summary) or None.
|
|
"""
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
|
|
@dataclass
|
|
class SiphonHit:
|
|
category: str # COMPLETED, BLOCKER, DECISION, ALERT, MILESTONE
|
|
confidence: float # 0.0 - 1.0
|
|
summary: str # one-line summary for main chat
|
|
thread_id: str # source side chat
|
|
message_id: str # source message
|
|
# Full message text is NOT stored here — main chat gets a summary
|
|
# plus a link back, never the full content (safety: no sensitive
|
|
# data siphoned verbatim).
|
|
|
|
|
|
# --- Keyword sets (configurable) ---
|
|
|
|
COMPLETED_PATTERNS = [
|
|
re.compile(r'\b(done|completed|finished|deployed|shipped|live|verified)\b', re.I),
|
|
re.compile(r'\b(all|tests?)\s+(pass|green|passing)\b', re.I),
|
|
re.compile(r'\[RESULT[^\]]*\]\s*OK', re.I),
|
|
re.compile(r'\b(merged|committed|pushed|published)\b', re.I),
|
|
]
|
|
|
|
BLOCKER_PATTERNS = [
|
|
re.compile(r'\b(blocked|stuck|failing|broken|down|error|failed)\b', re.I),
|
|
re.compile(r'\b(need|needs|waiting)\s+(your|approval|input|decision)\b', re.I),
|
|
re.compile(r'\b(can\'t|cannot|unable to)\b', re.I),
|
|
re.compile(r'\[RESULT[^\]]*\]\s*(FAIL|ERROR)', re.I),
|
|
]
|
|
|
|
DECISION_PATTERNS = [
|
|
re.compile(r'\b(should (i|we)|shall i|want me to)\b', re.I),
|
|
re.compile(r'\b(your call|needs? your|awaiting your)\b', re.I),
|
|
re.compile(r'\b(approve|approval)\b.*\?', re.I),
|
|
re.compile(r'^(yes|no)\s*\?\s*$', re.I),
|
|
]
|
|
|
|
ALERT_PATTERNS = [
|
|
re.compile(r'\b(security|vulnerability|breach|compromised|exploit)\b', re.I),
|
|
re.compile(r'\b(urgent|critical|emergency|asap)\b', re.I),
|
|
re.compile(r'\b(502|503|500)\b.*\b(error|down)\b', re.I),
|
|
re.compile(r'\b(ssh|tunnel).*\b(down|broken|failed)\b', re.I),
|
|
]
|
|
|
|
MILESTONE_PATTERNS = [
|
|
re.compile(r'\b(milestone|phase \d+ (complete|done)|shipped v)\b', re.I),
|
|
re.compile(r'\b(all \d+ (items? )?done)\b', re.I),
|
|
]
|
|
|
|
# Patterns that suppress siphoning (safety)
|
|
SUPPRESS_PATTERNS = [
|
|
re.compile(r'\b(password|secret|token|key|pin)\s*[:=]', re.I),
|
|
re.compile(r'-----BEGIN', re.I), # never siphon key material
|
|
re.compile(r'\[do not siphon\]', re.I), # explicit opt-out marker
|
|
]
|
|
|
|
|
|
def _match_score(text: str, patterns) -> float:
|
|
"""Return confidence based on how many patterns match."""
|
|
hits = sum(1 for p in patterns if p.search(text))
|
|
if hits == 0:
|
|
return 0.0
|
|
# Diminishing returns: 1 hit = 0.6, 2 = 0.8, 3+ = 0.95
|
|
return min(0.95, 0.6 + (hits - 1) * 0.2)
|
|
|
|
|
|
def _extract_summary(text: str, max_len: int = 120) -> str:
|
|
"""Extract a safe one-line summary. Strips to first meaningful line."""
|
|
# Take first non-empty line, truncate
|
|
for line in text.strip().split('\n'):
|
|
line = line.strip()
|
|
if line and len(line) > 10:
|
|
if len(line) > max_len:
|
|
return line[:max_len - 3] + '...'
|
|
return line
|
|
return text[:max_len]
|
|
|
|
|
|
def detect(text: str, thread_id: str, message_id: str,
|
|
min_confidence: float = 0.6) -> Optional[SiphonHit]:
|
|
"""
|
|
Check a side chat message for siphon-worthy content.
|
|
Returns SiphonHit or None.
|
|
"""
|
|
# Safety: suppress sensitive content
|
|
for p in SUPPRESS_PATTERNS:
|
|
if p.search(text):
|
|
return None
|
|
|
|
candidates = [
|
|
("COMPLETED", _match_score(text, COMPLETED_PATTERNS)),
|
|
("BLOCKER", _match_score(text, BLOCKER_PATTERNS)),
|
|
("DECISION", _match_score(text, DECISION_PATTERNS)),
|
|
("ALERT", _match_score(text, ALERT_PATTERNS)),
|
|
("MILESTONE", _match_score(text, MILESTONE_PATTERNS)),
|
|
]
|
|
|
|
# Sort by confidence descending; ALERT wins ties (safety: urgency first)
|
|
# Use negative confidence for descending, and ALERT as tiebreaker
|
|
candidates.sort(key=lambda x: (-x[1], 0 if x[0] == "ALERT" else 1))
|
|
best_cat, best_conf = candidates[0]
|
|
|
|
if best_conf < min_confidence:
|
|
return None
|
|
|
|
return SiphonHit(
|
|
category=best_cat,
|
|
confidence=best_conf,
|
|
summary=_extract_summary(text),
|
|
thread_id=thread_id,
|
|
message_id=message_id,
|
|
)
|
|
|
|
|
|
# --- Opt-out registry ---
|
|
|
|
_opt_out_threads: set = set()
|
|
|
|
|
|
def opt_out(thread_id: str):
|
|
"""Agent opts a side chat out of siphoning."""
|
|
_opt_out_threads.add(thread_id)
|
|
|
|
|
|
def opt_in(thread_id: str):
|
|
"""Re-enable siphoning for a side chat."""
|
|
_opt_out_threads.discard(thread_id)
|
|
|
|
|
|
def is_opted_out(thread_id: str) -> bool:
|
|
return thread_id in _opt_out_threads
|