From 6feca2ba989eacb999c74dfb43c4791f8220daab Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Wed, 19 Aug 2026 00:30:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(I):=20I-WORK-BASE/I-WORK-ACTOR/I-WORK-HAND?= =?UTF-8?q?OFF=20=E2=80=94=20workstream=20contract=20(typed=20blocks,=20un?= =?UTF-8?q?ified=20posting=20path,=20human-agent=20handoff=20with=20task?= =?UTF-8?q?=20creation),=2025=20tests=20passing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/ai/workstream_contract.py | 100 ++++++++++++++++++++++++++++++ tests/test_phase_i_integration.py | 93 +++++++++++++++++++++++++++ 2 files changed, 193 insertions(+) create mode 100644 app/ai/workstream_contract.py diff --git a/app/ai/workstream_contract.py b/app/ai/workstream_contract.py new file mode 100644 index 0000000..a480a93 --- /dev/null +++ b/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"] diff --git a/tests/test_phase_i_integration.py b/tests/test_phase_i_integration.py index dce0fb1..c6752db 100644 --- a/tests/test_phase_i_integration.py +++ b/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"