"""Agent → Workstream integration. Posts agent messages to the central communication system (kommunikation plugin). Supports text, action_card, entity_card, task_card, approval, and miniapp block types. All agent messages are marked as AI-generated via transparency metadata. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.ai.transparency import mark_as_ai_generated logger = logging.getLogger(__name__) async def post_agent_message( db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID, agent_run_id: uuid.UUID, content: str, block_type: str = "text", block_data: dict[str, Any] | None = None, conversation_id: uuid.UUID | None = None, ) -> uuid.UUID: """Post a message from an agent to the communication system. Creates a CommMessage in the agent's conversation channel. If no conversation_id is provided, uses the agent's default channel. Args: db: Database session. tenant_id: Tenant ID. agent_id: Agent definition ID. agent_run_id: Agent run ID for traceability. content: Message text content. block_type: Block type (text, action_card, entity_card, task_card, approval, miniapp). block_data: Additional block data (e.g. action buttons, entity reference). conversation_id: Optional conversation to post to. If None, uses agent channel. Returns: Message ID. """ from app.plugins.builtins.kommunikation.models import CommMessage, CommMessageBlock from app.plugins.builtins.kommunikation.services import send_message # Mark as AI-generated ai_metadata = mark_as_ai_generated(content, { "agent_id": str(agent_id), "agent_run_id": str(agent_run_id), }) # Build block if not plain text blocks: list[dict[str, Any]] = [] if block_type != "text" and block_data: blocks.append({ "type": block_type, "data": block_data, }) # Post via kommunikation service message_id = await send_message( db=db, tenant_id=tenant_id, sender_id=agent_id, sender_type="agent", conversation_id=conversation_id or await _get_or_create_agent_channel(db, tenant_id, agent_id), content=content, blocks=blocks, metadata=ai_metadata, ) logger.info( "Agent %s posted message %s (block_type=%s, run=%s)", agent_id, message_id, block_type, agent_run_id, ) return message_id async def _get_or_create_agent_channel( db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID ) -> uuid.UUID: """Get or create a dedicated conversation channel for an agent.""" from app.plugins.builtins.kommunikation.models import CommConversation # Try to find existing agent channel result = await db.execute( select(CommConversation).where( CommConversation.tenant_id == tenant_id, CommConversation.entity_type == "agent", CommConversation.entity_id == agent_id, CommConversation.deleted_at.is_(None), ) ) conv = result.scalar_one_or_none() if conv: return conv.id # Create new channel conv = CommConversation( tenant_id=tenant_id, entity_type="agent", entity_id=agent_id, title=f"Agent Channel", conversation_type="channel", is_system=False, ) db.add(conv) await db.flush() return conv.id async def post_agent_step( db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID, agent_run_id: uuid.UUID, step_number: int, thought: str, action: str | None = None, observation: str | None = None, ) -> uuid.UUID | None: """Post a ReAct step as an action_card to the workstream. Only posts if the agent's trace_mode is 'extended'. """ block_data = { "step_number": step_number, "thought": thought[:500], # Truncate for display "action": action, "observation": (observation or "")[:500], } return await post_agent_message( db=db, tenant_id=tenant_id, agent_id=agent_id, agent_run_id=agent_run_id, content=f"Step {step_number}: {action or 'Thinking...'}", block_type="action_card", block_data=block_data, ) async def post_agent_result( db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID, agent_run_id: uuid.UUID, final_content: str, total_cost_usd: float, steps_taken: int, status: str, ) -> uuid.UUID: """Post the final result of an agent run to the workstream.""" block_data = { "status": status, "steps_taken": steps_taken, "total_cost_usd": round(total_cost_usd, 6), "run_id": str(agent_run_id), } return await post_agent_message( db=db, tenant_id=tenant_id, agent_id=agent_id, agent_run_id=agent_run_id, content=final_content, block_type="action_card", block_data=block_data, ) async def post_approval_request( db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID, agent_run_id: uuid.UUID, approval_id: uuid.UUID, action: str, description: str, ) -> uuid.UUID: """Post an approval request card to the workstream.""" block_data = { "approval_id": str(approval_id), "action": action, "description": description, "status": "pending", } return await post_agent_message( db=db, tenant_id=tenant_id, agent_id=agent_id, agent_run_id=agent_run_id, content=f"Approval required: {action}", block_type="approval", block_data=block_data, )