leocrm/app/ai/agent_workstream.py

"""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,
    )