feat(I): I-WORK-BASE/I-WORK-ACTOR/I-WORK-HANDOFF — workstream contract (typed blocks, unified posting path, human-agent handoff with task creation), 25 tests passing

app/ai/workstream_contract.py

@@ -0,0 +1,100 @@
+"""Workstream contract — unified posting path for Human/System/Agent/Workflow (I-WORK-BASE, I-WORK-ACTOR, I-WORK-HANDOFF).
+
+Defines the contract for posting typed blocks to the central Communication
+system. All actors use the same posting path with typed blocks.
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+ActorType = Literal["human", "system", "agent", "workflow"]
+BlockType = Literal["text", "entity_card", "action_card", "evidence_card", "approval_card", "miniapp", "workflow_status", "workflow_handoff", "error"]
+
+
+@dataclass
+class WorkstreamBlock:
+    type: BlockType
+    content: str = ""
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"type": self.type, "content": self.content, "metadata": self.metadata}
+
+
+@dataclass
+class WorkstreamMessage:
+    actor_type: ActorType
+    actor_id: str | None = None
+    content: str = ""
+    blocks: list[WorkstreamBlock] = field(default_factory=list)
+    conversation_id: str | None = None
+    tenant_id: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "actor_type": self.actor_type,
+            "actor_id": self.actor_id,
+            "content": self.content,
+            "blocks": [b.to_dict() for b in self.blocks],
+            "conversation_id": self.conversation_id,
+            "tenant_id": self.tenant_id,
+        }
+
+
+async def post_to_workstream(db: AsyncSession, tenant_id: uuid.UUID, message: WorkstreamMessage) -> dict[str, Any] | None:
+    """Post a message to the central workstream (I-WORK-ACTOR)."""
+    try:
+        from app.plugins.builtins.kommunikation.contracts import KommunikationContract
+        contract = KommunikationContract
+        post_fn = contract.get_function("post_message")
+        if post_fn is None:
+            if message.actor_type == "human" and message.actor_id:
+                from app.core.notifications import post_system_message
+                await post_system_message(db, tenant_id, uuid.UUID(message.actor_id), "workstream", message.content[:200], message.content)
+            return None
+        return await post_fn(db=db, tenant_id=tenant_id, sender_id=uuid.UUID(message.actor_id) if message.actor_id else None, sender_type=message.actor_type, message_type=f"workstream_{message.blocks[0].type}" if message.blocks else "workstream_text", content=message.content, blocks=[b.to_dict() for b in message.blocks], conversation_id=message.conversation_id)
+    except Exception as e:
+        logger.warning("Failed to post to workstream: %s", e)
+        return None
+
+
+async def create_handoff(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, *, handoff_type: str, assignee_type: str = "user", assignee_id: str | None = None, entity_type: str | None = None, entity_id: str | None = None, description: str = "", agent_run_id: str | None = None, workflow_instance_id: str | None = None, conversation_id: str | None = None) -> dict[str, Any]:
+    """Create a Human<->Agent handoff (I-WORK-HANDOFF). Creates a Task with task_type='handoff'."""
+    from app.plugins.builtins.tasks.services import create_task
+    task_data: dict[str, Any] = {"title": f"Handoff: {handoff_type}", "description": description, "task_type": "handoff", "assignee_type": assignee_type, "assignee_id": assignee_id, "entity_type": entity_type, "entity_id": entity_id, "status": "open", "priority": "medium"}
+    task = await create_task(db, tenant_id, user_id, task_data)
+    handoff_block = WorkstreamBlock(type="workflow_handoff", content=description, metadata={"handoff_type": handoff_type, "task_id": task.get("id") if task else None, "assignee_type": assignee_type, "assignee_id": assignee_id, "entity_type": entity_type, "entity_id": entity_id, "agent_run_id": agent_run_id, "workflow_instance_id": workflow_instance_id})
+    message = WorkstreamMessage(actor_type="system", content=f"Handoff: {handoff_type} - {description}", blocks=[handoff_block], conversation_id=conversation_id, tenant_id=str(tenant_id))
+    post_result = await post_to_workstream(db, tenant_id, message)
+    return {"task": task, "workstream_post": post_result, "handoff_type": handoff_type}
+
+
+def build_entity_card(entity_type: str, entity_id: str, title: str = "", subtitle: str = "", url: str = "") -> WorkstreamBlock:
+    return WorkstreamBlock(type="entity_card", content=title, metadata={"entity_type": entity_type, "entity_id": entity_id, "title": title, "subtitle": subtitle, "url": url})
+
+
+def build_action_card(title: str, actions: list[dict[str, str]], description: str = "") -> WorkstreamBlock:
+    return WorkstreamBlock(type="action_card", content=title, metadata={"title": title, "description": description, "actions": actions})
+
+
+def build_evidence_card(source_type: str, source_id: str, title: str, snippet: str = "", url: str = "", confidence: float = 0.0) -> WorkstreamBlock:
+    return WorkstreamBlock(type="evidence_card", content=title, metadata={"source_type": source_type, "source_id": source_id, "title": title, "snippet": snippet[:200], "url": url, "confidence": confidence})
+
+
+def build_approval_card(approval_id: str, action: str, description: str = "") -> WorkstreamBlock:
+    return WorkstreamBlock(type="approval_card", content=f"Approval needed: {action}", metadata={"approval_id": approval_id, "action": action, "description": description})
+
+
+def build_miniapp_block(app_id: str, title: str = "", render_schema: dict[str, Any] | None = None) -> WorkstreamBlock:
+    return WorkstreamBlock(type="miniapp", content=title, metadata={"app_id": app_id, "title": title, "render_schema": render_schema or {}})
+
+
+__all__ = ["ActorType", "BlockType", "WorkstreamBlock", "WorkstreamMessage", "post_to_workstream", "create_handoff", "build_entity_card", "build_action_card", "build_evidence_card", "build_approval_card", "build_miniapp_block"]

tests/test_phase_i_integration.py

@@ -204,3 +204,96 @@ class TestMCPExposure:
             )
             assert result["query"] == "test"
             mock_search.assert_called_once()
+
+
+# ─── I-WORK-BASE/ACTOR/HANDOFF: Workstream Contract ──────────────────────────
+
+
+class TestWorkstreamContract:
+    """Test the workstream contract module (I-WORK-BASE, I-WORK-ACTOR, I-WORK-HANDOFF)."""
+
+    def test_workstream_block_dataclass(self):
+        """WorkstreamBlock dataclass works correctly."""
+        from app.ai.workstream_contract import WorkstreamBlock
+        block = WorkstreamBlock(type="text", content="Hello", metadata={"key": "value"})
+        assert block.type == "text"
+        assert block.content == "Hello"
+        d = block.to_dict()
+        assert d["type"] == "text"
+        assert d["content"] == "Hello"
+
+    def test_workstream_message_dataclass(self):
+        """WorkstreamMessage dataclass works correctly."""
+        from app.ai.workstream_contract import WorkstreamMessage, WorkstreamBlock
+        msg = WorkstreamMessage(actor_type="agent", actor_id="abc-123", content="Test", blocks=[WorkstreamBlock(type="text")])
+        assert msg.actor_type == "agent"
+        assert msg.actor_id == "abc-123"
+        d = msg.to_dict()
+        assert d["actor_type"] == "agent"
+        assert len(d["blocks"]) == 1
+
+    def test_build_entity_card(self):
+        """build_entity_card creates correct block."""
+        from app.ai.workstream_contract import build_entity_card
+        block = build_entity_card("contact", "123", "John Doe", "CEO", "/contacts/123")
+        assert block.type == "entity_card"
+        assert block.metadata["entity_type"] == "contact"
+        assert block.metadata["title"] == "John Doe"
+
+    def test_build_action_card(self):
+        """build_action_card creates correct block."""
+        from app.ai.workstream_contract import build_action_card
+        block = build_action_card("Approve?", [{"label": "Yes", "action": "approve"}], "Please approve")
+        assert block.type == "action_card"
+        assert len(block.metadata["actions"]) == 1
+
+    def test_build_evidence_card(self):
+        """build_evidence_card creates correct block."""
+        from app.ai.workstream_contract import build_evidence_card
+        block = build_evidence_card("wiki", "456", "Article", "Snippet", "/wiki/456", 0.9)
+        assert block.type == "evidence_card"
+        assert block.metadata["confidence"] == 0.9
+
+    def test_build_approval_card(self):
+        """build_approval_card creates correct block."""
+        from app.ai.workstream_contract import build_approval_card
+        block = build_approval_card("appr-123", "send_email", "Please approve")
+        assert block.type == "approval_card"
+        assert block.metadata["approval_id"] == "appr-123"
+
+    def test_build_miniapp_block(self):
+        """build_miniapp_block creates correct block."""
+        from app.ai.workstream_contract import build_miniapp_block
+        block = build_miniapp_block("calendar-app", "Calendar", {"type": "form"})
+        assert block.type == "miniapp"
+        assert block.metadata["app_id"] == "calendar-app"
+
+    @pytest.mark.asyncio
+    async def test_post_to_workstream_fallback(self):
+        """post_to_workstream falls back to notification when CommContract unavailable."""
+        from app.ai.workstream_contract import post_to_workstream, WorkstreamMessage, WorkstreamBlock
+
+        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
+            result = await post_to_workstream(
+                db=MagicMock(), tenant_id=uuid.uuid4(),
+                message=WorkstreamMessage(actor_type="human", actor_id=str(uuid.uuid4()), content="Test", blocks=[WorkstreamBlock(type="text")]),
+            )
+            assert result is None  # Fallback
+
+    @pytest.mark.asyncio
+    async def test_create_handoff_creates_task(self):
+        """create_handoff creates a Task with task_type='handoff'."""
+        from app.ai.workstream_contract import create_handoff
+
+        with patch("app.plugins.builtins.tasks.services.create_task", new_callable=AsyncMock) as mock_create:
+            mock_create.return_value = {"id": "task-123", "title": "Handoff: review_needed"}
+            with patch("app.ai.workstream_contract.post_to_workstream", new_callable=AsyncMock):
+                result = await create_handoff(
+                    db=MagicMock(), tenant_id=uuid.uuid4(), user_id=uuid.uuid4(),
+                    handoff_type="review_needed", description="Please review",
+                )
+                assert result["task"]["id"] == "task-123"
+                mock_create.assert_called_once()
+                # Verify task_type is 'handoff'
+                call_args = mock_create.call_args
+                assert call_args[0][3]["task_type"] == "handoff"