From df261b1b2c2dfacecb93e0ed8c6d83456365d306 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Wed, 19 Aug 2026 01:09:08 +0200 Subject: [PATCH] =?UTF-8?q?feat(I):=20I-WORK-PROACTIVE=20=E2=80=94=20proac?= =?UTF-8?q?tive=20workstream=20feed=20(suggestions,=20cooldown,=20dedupe,?= =?UTF-8?q?=20user=20settings,=20priority=20filtering),=2034=20tests=20pas?= =?UTF-8?q?sing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/ai/proactive_feed.py | 236 ++++++++++++++++++++++++++++++ tests/test_phase_i_integration.py | 83 +++++++++++ 2 files changed, 319 insertions(+) create mode 100644 app/ai/proactive_feed.py diff --git a/app/ai/proactive_feed.py b/app/ai/proactive_feed.py new file mode 100644 index 0000000..d08ff9f --- /dev/null +++ b/app/ai/proactive_feed.py @@ -0,0 +1,236 @@ +"""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", +] diff --git a/tests/test_phase_i_integration.py b/tests/test_phase_i_integration.py index c6752db..fb085d9 100644 --- a/tests/test_phase_i_integration.py +++ b/tests/test_phase_i_integration.py @@ -297,3 +297,86 @@ class TestWorkstreamContract: # Verify task_type is 'handoff' call_args = mock_create.call_args assert call_args[0][3]["task_type"] == "handoff" + + +# ─── I-WORK-PROACTIVE: Proactive Workstream Feed ───────────────────────────── + + +class TestProactiveFeed: + """Test the proactive feed module (I-WORK-PROACTIVE).""" + + def test_proactive_suggestion_dataclass(self): + """ProactiveSuggestion dataclass works correctly.""" + from app.ai.proactive_feed import ProactiveSuggestion + s = ProactiveSuggestion(trigger="mail.received", title="Test", priority="medium") + assert s.trigger == "mail.received" + assert s.priority == "medium" + d = s.to_dict() + assert d["trigger"] == "mail.received" + + def test_get_cooldown_known_trigger(self): + """get_cooldown returns correct cooldown for known triggers.""" + from app.ai.proactive_feed import get_cooldown + assert get_cooldown("mail.received") == 300 + assert get_cooldown("contact.created") == 600 + assert get_cooldown("workflow.completed") == 60 + + def test_get_cooldown_unknown_trigger(self): + """get_cooldown returns default for unknown triggers.""" + from app.ai.proactive_feed import get_cooldown + assert get_cooldown("unknown.trigger") == 300 + + def test_is_cooled_down_initial(self): + """is_cooled_down returns False for first-time trigger.""" + from app.ai.proactive_feed import is_cooled_down + assert is_cooled_down(uuid.uuid4(), "mail.received") is False + + def test_mark_suggested_sets_cooldown(self): + """mark_suggested sets cooldown for the trigger.""" + from app.ai.proactive_feed import is_cooled_down, mark_suggested + tid = uuid.uuid4() + mark_suggested(tid, "mail.received", "msg-123") + assert is_cooled_down(tid, "mail.received", "msg-123") is True + + def test_filter_by_user_settings_enabled(self): + """filter_by_user_settings filters by enabled flag.""" + from app.ai.proactive_feed import ProactiveSuggestion, filter_by_user_settings + suggestions = [ProactiveSuggestion(trigger="test", priority="medium")] + assert len(filter_by_user_settings(suggestions, {"enabled": True})) == 1 + assert len(filter_by_user_settings(suggestions, {"enabled": False})) == 0 + + def test_filter_by_user_settings_min_priority(self): + """filter_by_user_settings filters by min_priority.""" + from app.ai.proactive_feed import ProactiveSuggestion, filter_by_user_settings + suggestions = [ + ProactiveSuggestion(trigger="test", priority="low"), + ProactiveSuggestion(trigger="test", priority="medium"), + ProactiveSuggestion(trigger="test", priority="high"), + ] + filtered = filter_by_user_settings(suggestions, {"enabled": True, "min_priority": "medium"}) + assert len(filtered) == 2 # medium + high + + @pytest.mark.asyncio + async def test_generate_suggestions_mail_received(self): + """generate_suggestions creates suggestion for mail.received trigger.""" + from app.ai.proactive_feed import generate_suggestions + suggestions = await generate_suggestions( + db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(), + trigger="mail.received", payload={"entity_id": str(uuid.uuid4()), "sender": "test@example.com"}, + ) + assert len(suggestions) == 1 + assert suggestions[0].trigger == "mail.received" + assert suggestions[0].priority == "medium" + + @pytest.mark.asyncio + async def test_generate_suggestions_cooldown_blocks(self): + """generate_suggestions returns empty list when in cooldown.""" + from app.ai.proactive_feed import generate_suggestions, mark_suggested + tid = uuid.uuid4() + eid = str(uuid.uuid4()) + mark_suggested(tid, "mail.received", eid) + suggestions = await generate_suggestions( + db=MagicMock(), tenant_id=tid, user_id=uuid.uuid4(), + trigger="mail.received", payload={"entity_id": eid}, + ) + assert len(suggestions) == 0