"""Proactive workstream feed — contextual suggestions and actions (I-WORK-PROACTIVE). UI-/Domain-Trigger erzeugen kontextuelle Vorschläge/Actions im Workstream mit Priority, Dedupe, Cooldown und User-Einstellungen. Kein störendes Popup-/Clippy-Verhalten. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any, Literal from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) Priority = Literal["low", "medium", "high", "urgent"] @dataclass class ProactiveSuggestion: """A proactive suggestion/action for the workstream feed.""" id: str = field(default_factory=lambda: str(uuid.uuid4())) trigger: str = "" # What triggered this (e.g. "mail.received", "contact.created") title: str = "" description: str = "" priority: Priority = "medium" action_type: str = "" # suggestion, action_required, info action_url: str = "" # Deep link to action entity_type: str | None = None entity_id: str | None = None blocks: list[dict[str, Any]] = field(default_factory=list) created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) expires_at: datetime | None = None metadata: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "id": self.id, "trigger": self.trigger, "title": self.title, "description": self.description, "priority": self.priority, "action_type": self.action_type, "action_url": self.action_url, "entity_type": self.entity_type, "entity_id": self.entity_id, "blocks": self.blocks, "created_at": self.created_at.isoformat(), "expires_at": self.expires_at.isoformat() if self.expires_at else None, "metadata": self.metadata, } # ─── Dedupe + Cooldown ─────────────────────────────────────────────────────── # In-memory dedupe cache (per-tenant). In production, use Redis. _dedupe_cache: dict[str, dict[str, datetime]] = {} # Default cooldown per trigger type (seconds) DEFAULT_COOLDOWNS: dict[str, int] = { "mail.received": 300, # 5 min between suggestions for same mail "contact.created": 600, # 10 min "workflow.completed": 60, # 1 min "agent.result": 120, # 2 min "default": 300, # 5 min default } def get_cooldown(trigger: str) -> int: """Get cooldown period for a trigger type.""" return DEFAULT_COOLDOWNS.get(trigger, DEFAULT_COOLDOWNS["default"]) def _dedupe_key(tenant_id: uuid.UUID, trigger: str, entity_id: str | None) -> str: """Build a dedupe key for a suggestion.""" return f"{tenant_id}:{trigger}:{entity_id or 'none'}" def is_cooled_down(tenant_id: uuid.UUID, trigger: str, entity_id: str | None = None) -> bool: """Check if a trigger is still in cooldown (should not produce new suggestions).""" key = _dedupe_key(tenant_id, trigger, entity_id) tenant_cache = _dedupe_cache.get(str(tenant_id), {}) last_seen = tenant_cache.get(key) if last_seen is None: return False cooldown = get_cooldown(trigger) return datetime.now(UTC) - last_seen < timedelta(seconds=cooldown) def mark_suggested(tenant_id: uuid.UUID, trigger: str, entity_id: str | None = None) -> None: """Mark a trigger as having produced a suggestion (for cooldown tracking).""" key = _dedupe_key(tenant_id, trigger, entity_id) tenant_id_str = str(tenant_id) if tenant_id_str not in _dedupe_cache: _dedupe_cache[tenant_id_str] = {} _dedupe_cache[tenant_id_str][key] = datetime.now(UTC) # ─── Suggestion Generators ─────────────────────────────────────────────────── async def generate_suggestions( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, trigger: str, payload: dict[str, Any], ) -> list[ProactiveSuggestion]: """Generate proactive suggestions for a trigger event (I-WORK-PROACTIVE). Checks cooldown, generates suggestions, and marks them as suggested. Returns a list of ProactiveSuggestion objects. """ entity_id = payload.get("entity_id") or payload.get("contact_id") or payload.get("message_id") # Check cooldown — don't spam if is_cooled_down(tenant_id, trigger, entity_id): return [] suggestions: list[ProactiveSuggestion] = [] # Generate based on trigger type if trigger == "mail.received": suggestions.append(ProactiveSuggestion( trigger=trigger, title="New email received", description=f"You received a new email from {payload.get('sender', 'unknown')}", priority="medium", action_type="info", action_url=f"/mail/messages/{entity_id}" if entity_id else "", entity_type="mail", entity_id=entity_id, blocks=[], )) elif trigger == "contact.created": suggestions.append(ProactiveSuggestion( trigger=trigger, title="New contact created", description=f"New contact: {payload.get('name', 'Unknown')}", priority="low", action_type="suggestion", action_url=f"/contacts/{entity_id}" if entity_id else "", entity_type="contact", entity_id=entity_id, )) elif trigger == "workflow.completed": suggestions.append(ProactiveSuggestion( trigger=trigger, title="Workflow completed", description=f"Workflow '{payload.get('workflow_name', 'Unknown')}' has been completed.", priority="medium", action_type="info", action_url=f"/workflows/instances/{entity_id}" if entity_id else "", entity_type="workflow_instance", entity_id=entity_id, )) elif trigger == "agent.result": suggestions.append(ProactiveSuggestion( trigger=trigger, title="Agent completed task", description=f"Agent finished: {payload.get('summary', 'Task completed')}", priority="medium", action_type="action_required", action_url=f"/agents/runs/{entity_id}" if entity_id else "", entity_type="agent_run", entity_id=entity_id, )) # Mark as suggested (cooldown) if suggestions: mark_suggested(tenant_id, trigger, entity_id) return suggestions # ─── User Settings ─────────────────────────────────────────────────────────── def get_user_proactive_settings(user_id: uuid.UUID) -> dict[str, Any]: """Get proactive feed settings for a user. In production, this would load from DB/user preferences. For now, returns defaults. """ return { "enabled": True, "min_priority": "low", # Don't show suggestions below this priority "max_per_hour": 20, # Rate limit suggestions per hour "triggers_enabled": { "mail.received": True, "contact.created": True, "workflow.completed": True, "agent.result": True, }, } def filter_by_user_settings( suggestions: list[ProactiveSuggestion], settings: dict[str, Any], ) -> list[ProactiveSuggestion]: """Filter suggestions by user settings.""" if not settings.get("enabled", True): return [] min_priority = settings.get("min_priority", "low") priority_order = {"low": 0, "medium": 1, "high": 2, "urgent": 3} min_level = priority_order.get(min_priority, 0) triggers_enabled = settings.get("triggers_enabled", {}) return [ s for s in suggestions if priority_order.get(s.priority, 0) >= min_level and triggers_enabled.get(s.trigger, True) ] __all__ = [ "ProactiveSuggestion", "generate_suggestions", "is_cooled_down", "mark_suggested", "get_cooldown", "get_user_proactive_settings", "filter_by_user_settings", "DEFAULT_COOLDOWNS", ]