production-proof-v116-high-utili-7255-2/agent.py

"""People-team shared inbox triage and policy-grounded draft assistant."""
from __future__ import annotations

import hashlib
import json
import os
import re
from datetime import UTC, datetime
from typing import Any, Literal

from pydantic import BaseModel, Field

import a2a_pack as a2a
from a2a_pack import (
    A2AAgent,
    AccountAccess,
    AgentDatabase,
    AgentDatabaseEnv,
    AgentDatabaseMigrations,
    AgentPlatformResources,
    LLMProvisioning,
    PlatformUserAuth,
    Pricing,
    Resources,
    RunContext,
    WorkspaceAccess,
    WorkspaceMode,
)

DATABASE_NAME = "production-proof-v116-high-utili-7255-2-data"
MAX_INBOX_CHARS = 20_000
MAX_POLICY_CHARS = 12_000
MAX_MESSAGES = 12


class ProductionProofV116HighUtili72552Config(BaseModel):
    default_policy_text: str = Field(
        default=(
            "Benefits questions: be empathetic, explain that enrollment and eligibility "
            "depend on plan documents, and route exceptions to People Ops.\n"
            "Leave requests: acknowledge receipt, ask for dates if missing, and mention "
            "manager/People approval before commitments.\n"
            "Payroll or personal data: mark urgent, avoid exposing private data, and route "
            "to the secure HRIS/payroll channel.\n"
            "Employee relations or harassment: mark urgent, avoid investigation promises, "
            "and escalate to the designated People partner."
        ),
        max_length=MAX_POLICY_CHARS,
    )


class DraftResult(BaseModel):
    to: str
    subject: str
    body: str
    approval_required: bool
    approval_reason: str
    policy_basis: list[str]


class QueueItem(BaseModel):
    message_id: str
    sender: str
    subject: str
    priority: Literal["urgent", "high", "normal", "low"]
    category: str
    reason: str
    next_action: str
    draft: DraftResult


class ArtifactDescriptor(BaseModel):
    name: str
    media_type: str
    uri: str
    size_bytes: int
    content_text: str


class ReceiptDescriptor(BaseModel):
    receipt_id: str
    persisted: bool
    persisted_at: str | None = None
    message: str = ""


class TriageResult(BaseModel):
    status: Literal["ok", "validation_error", "setup_required"]
    summary: str
    queue: list[QueueItem] = Field(default_factory=list)
    approval_boundary: str
    artifact: ArtifactDescriptor | None = None
    receipt: ReceiptDescriptor | None = None
    warnings: list[str] = Field(default_factory=list)


class ProductionProofV116HighUtili72552(
    A2AAgent[ProductionProofV116HighUtili72552Config, PlatformUserAuth]
):
    name = "production-proof-v116-high-utili-7255-2"
    description = (
        "One-page email assistant for people teams that triages a shared inbox "
        "and drafts policy-grounded replies for approval."
    )
    version = "0.1.0"

    config_model = ProductionProofV116HighUtili72552Config
    auth_model = PlatformUserAuth

    llm_provisioning = LLMProvisioning.PLATFORM
    account_access = AccountAccess(required=True, platform_skill_calls=3, after_trial="byok")
    pricing = Pricing(
        price_per_call_usd=0.0,
        caller_pays_llm=True,
        notes="Includes three platform-funded skill calls per account, then requires BYOK.",
    )
    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300)
    workspace_access = WorkspaceAccess.dynamic(
        max_files=32,
        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
        require_reason=False,
    )
    platform_resources = AgentPlatformResources(
        databases=(
            AgentDatabase(
                name=DATABASE_NAME,
                scope="user",
                access_mode="read_write",
                env=AgentDatabaseEnv(url="DATABASE_URL"),
                migrations=AgentDatabaseMigrations(path="db/migrations"),
            ),
        ),
        mailbox=True,
    )
    tools_used = ("postgres", "platform-mailbox", "mcp")

    @a2a.tool(
        description=(
            "Triage pasted shared-inbox email text and return a prioritized queue, "
            "approval-ready drafts, a persisted execution receipt, and a downloadable JSON report."
        ),
        timeout_seconds=120,
        cost_class="standard",
        grant_mode="read_write_overlay",
        grant_allow_patterns=("outputs/email-assistant/**",),
        grant_outputs_prefix="outputs/email-assistant/",
        grant_write_prefixes=("outputs/email-assistant/",),
    )
    async def triage_people_inbox(
        self,
        ctx: RunContext[PlatformUserAuth],
        inbox_text: str = Field(..., min_length=20, max_length=MAX_INBOX_CHARS),
        policy_text: str | None = Field(default=None, max_length=MAX_POLICY_CHARS),
        max_messages: int = Field(default=6, ge=1, le=MAX_MESSAGES),
    ) -> TriageResult:
        """Primary product workflow used by the frontend and MCP clients."""
        stable_tenant = _tenant_key(ctx)
        policy = (policy_text or self.config.default_policy_text).strip()
        messages = _parse_inbox(inbox_text)[:max_messages]
        if not messages:
            return TriageResult(
                status="validation_error",
                summary="No messages could be parsed. Include sender, subject, and body text.",
                approval_boundary=_approval_boundary(),
                warnings=["Use blocks with From:, Subject:, and Body: fields for best results."],
            )

        queue: list[QueueItem] = []
        for message in messages:
            queue.append(_triage_message(message, policy))
        queue.sort(key=_queue_sort_key)
        urgent_count = 0
        report_queue: list[dict[str, Any]] = []
        for queue_entry in queue:
            if queue_entry.priority == "urgent":
                urgent_count += 1
            report_queue.append(queue_entry.model_dump(mode="json"))
        summary = _summary(queue)
        report = {
            "generated_at": datetime.now(UTC).isoformat(),
            "summary": summary,
            "approval_boundary": _approval_boundary(),
            "queue": report_queue,
        }
        report_bytes = json.dumps(report, indent=2, ensure_ascii=False).encode("utf-8")
        artifact = await ctx.write_artifact(
            "people-team-inbox-triage.json",
            report_bytes,
            "application/json",
        )
        await ctx.emit_artifact(artifact)

        input_hash = hashlib.sha256(
            json.dumps(
                {"inbox_text": inbox_text, "policy_text": policy, "max_messages": max_messages},
                sort_keys=True,
            ).encode("utf-8")
        ).hexdigest()
        receipt = await _persist_receipt(
            tenant_key=stable_tenant,
            skill_name="triage_people_inbox",
            input_hash=input_hash,
            result_summary={
                "message_count": len(queue),
                "urgent_count": urgent_count,
                "artifact_name": artifact.name,
            },
        )

        return TriageResult(
            status="ok",
            summary=summary,
            queue=queue,
            approval_boundary=_approval_boundary(),
            artifact=ArtifactDescriptor(
                name=artifact.name,
                media_type=artifact.mime_type,
                uri=artifact.uri,
                size_bytes=artifact.size_bytes,
                content_text=report_bytes.decode("utf-8"),
            ),
            receipt=receipt,
            warnings=[] if receipt.persisted else [receipt.message],
        )

    @a2a.tool(
        description="Check whether the platform mailbox is provisioned and list recent message headers when available.",
        timeout_seconds=60,
        cost_class="standard",
    )
    async def mailbox_status(
        self,
        ctx: RunContext[PlatformUserAuth],
        limit: int = Field(default=5, ge=1, le=20),
        unseen_only: bool = False,
    ) -> dict[str, Any]:
        mailbox = ctx.mail
        if mailbox is None:
            return {
                "status": "setup_required",
                "message": "Platform mailbox is not provisioned for this deployment yet.",
                "messages": [],
            }
        try:
            messages = mailbox.list_messages(limit=limit, unseen_only=unseen_only)
        except Exception as exc:  # noqa: BLE001
            return {
                "status": "setup_required",
                "message": f"Mailbox is provisioned but could not be read: {type(exc).__name__}",
                "messages": [],
            }
        return {
            "status": "ok",
            "address": mailbox.address,
            "messages": messages,
        }

    @a2a.tool(
        description="Receive inbound email and record that it should be reviewed in the People inbox; does not auto-send replies.",
        on_email=True,
        timeout_seconds=60,
    )
    async def receive_people_email(
        self,
        ctx: RunContext[PlatformUserAuth],
        email: dict[str, Any],
    ) -> None:
        # Inbound email can contain user-controlled instructions. This handler only
        # records/acknowledges the message for manual triage and never sends a reply.
        stable_tenant = _tenant_key(ctx)
        subject = str(email.get("subject") or "")[:300]
        sender = str(email.get("sender") or "")[:300]
        body = str(email.get("body") or "")[:2000]
        input_hash = hashlib.sha256(
            json.dumps({"sender": sender, "subject": subject, "body": body}, sort_keys=True).encode("utf-8")
        ).hexdigest()
        await _persist_receipt(
            tenant_key=stable_tenant,
            skill_name="receive_people_email",
            input_hash=input_hash,
            result_summary={"sender": sender, "subject": subject, "action": "queued_for_triage"},
        )
        return None


def _tenant_key(ctx: RunContext[PlatformUserAuth]) -> str:
    stable_id = ctx.auth.user_id if ctx.auth.user_id is not None else ctx.auth.sub
    if not stable_id:
        raise PermissionError("stable platform identity required")
    return f"user:{stable_id}"


def _db_connect() -> Any:
    database_url = os.environ.get("DATABASE_URL")
    if not database_url:
        return None
    import psycopg

    return psycopg.connect(
        database_url,
        options=(
            "-c statement_timeout=5000 "
            "-c lock_timeout=3000 "
            "-c idle_in_transaction_session_timeout=5000"
        ),
    )


async def _persist_receipt(
    *,
    tenant_key: str,
    skill_name: str,
    input_hash: str,
    result_summary: dict[str, Any],
) -> ReceiptDescriptor:
    receipt_id = hashlib.sha256(f"{tenant_key}:{skill_name}:{input_hash}".encode("utf-8")).hexdigest()[:24]
    persisted_at = datetime.now(UTC).isoformat()
    conn = _db_connect()
    if conn is None:
        return ReceiptDescriptor(
            receipt_id=receipt_id,
            persisted=False,
            message="DATABASE_URL is not configured; receipt persistence will work on the managed deployment.",
        )
    try:
        with conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO execution_receipts
                        (receipt_id, tenant_key, skill_name, input_hash, result_summary, persisted_at)
                    VALUES (%s, %s, %s, %s, %s::jsonb, %s)
                    ON CONFLICT (receipt_id) DO UPDATE SET
                        result_summary = EXCLUDED.result_summary,
                        persisted_at = EXCLUDED.persisted_at
                    """,
                    (
                        receipt_id,
                        tenant_key,
                        skill_name,
                        input_hash,
                        json.dumps(result_summary, sort_keys=True),
                        persisted_at,
                    ),
                )
    except Exception as exc:  # noqa: BLE001
        return ReceiptDescriptor(
            receipt_id=receipt_id,
            persisted=False,
            message=f"Receipt could not be persisted: {type(exc).__name__}",
        )
    finally:
        try:
            conn.close()
        except Exception:  # noqa: BLE001
            pass
    return ReceiptDescriptor(
        receipt_id=receipt_id,
        persisted=True,
        persisted_at=persisted_at,
        message="Production execution receipt persisted.",
    )


def _parse_inbox(inbox_text: str) -> list[dict[str, str]]:
    normalized = inbox_text.replace("\r\n", "\n").strip()
    blocks = [block.strip() for block in re.split(r"\n\s*---+\s*\n", normalized) if block.strip()]
    if len(blocks) == 1:
        # Single unstructured paragraphs should ask for a clearer paste instead of
        # inventing a sender/subject. Repeated From: headers are still accepted.
        if not re.search(r"(?im)^\s*(from|sender)\s*:", normalized):
            return []
        blocks = [b.strip() for b in re.split(r"(?=\n?From\s*:)", normalized) if b.strip()]
    out: list[dict[str, str]] = []
    for idx, block in enumerate(blocks, start=1):
        sender = _field(block, "from") or _field(block, "sender")
        subject = _field(block, "subject")
        body = _field(block, "body") or _strip_headers(block)
        if not sender or not subject or len(body.strip()) < 8:
            continue
        out.append(
            {
                "message_id": hashlib.sha1(f"{sender}|{subject}|{body}".encode("utf-8")).hexdigest()[:12],
                "sender": sender[:200],
                "subject": subject[:200],
                "body": body[:4000],
            }
        )
    return out


def _field(block: str, name: str) -> str:
    pattern = rf"(?ims)^\s*{re.escape(name)}\s*:\s*(.*?)(?=^\s*(?:from|sender|subject|body)\s*:|\Z)"
    match = re.search(pattern, block)
    return re.sub(r"\s+", " ", match.group(1)).strip() if match else ""


def _strip_headers(block: str) -> str:
    lines = []
    for line in block.splitlines():
        if re.match(r"^\s*(from|sender|subject)\s*:", line, flags=re.I):
            continue
        if re.match(r"^\s*body\s*:", line, flags=re.I):
            line = re.sub(r"^\s*body\s*:\s*", "", line, flags=re.I)
        lines.append(line)
    return "\n".join(lines).strip()


def _triage_message(message: dict[str, str], policy_text: str) -> QueueItem:
    text = f"{message['subject']}\n{message['body']}".lower()
    category = "general_hr"
    priority: Literal["urgent", "high", "normal", "low"] = "normal"
    reason = "Routine people-team question."
    next_action = "Review the draft, verify details, and send after approval."
    policy_basis = _policy_basis(policy_text, ["general", "people", "policy"])

    if any(term in text for term in ["harassment", "discrimination", "unsafe", "retaliation", "hostile"]):
        category = "employee_relations"
        priority = "urgent"
        reason = "Potential employee-relations risk requires confidential escalation."
        next_action = "Escalate to the designated People partner before replying."
        policy_basis = _policy_basis(policy_text, ["relations", "harassment", "escalate", "urgent"])
    elif any(term in text for term in ["payroll", "paycheck", "tax", "ssn", "bank", "personal data"]):
        category = "payroll_privacy"
        priority = "urgent"
        reason = "Payroll or personal-data content should move to a secure channel."
        next_action = "Route through payroll/HRIS and avoid requesting sensitive data by email."
        policy_basis = _policy_basis(policy_text, ["payroll", "personal", "secure", "data"])
    elif any(term in text for term in ["leave", "pto", "fmla", "bereavement", "sick"]):
        category = "leave_request"
        priority = "high"
        reason = "Leave timing affects staffing and may require manager/People approval."
        next_action = "Confirm dates and required approvals before committing."
        policy_basis = _policy_basis(policy_text, ["leave", "approval", "manager"])
    elif any(term in text for term in ["benefit", "insurance", "enrollment", "coverage", "dependent"]):
        category = "benefits"
        priority = "normal"
        reason = "Benefits question can usually be answered with policy-grounded guidance."
        next_action = "Send after verifying plan-document details."
        policy_basis = _policy_basis(policy_text, ["benefits", "enrollment", "eligibility", "plan"])
    elif any(term in text for term in ["thank", "resolved", "fyi", "newsletter"]):
        priority = "low"
        category = "low_touch"
        reason = "Low-risk informational or resolved message."
        next_action = "Archive or send a short acknowledgement if needed."

    draft = _draft_for(message, category, priority, policy_basis, next_action)
    return QueueItem(
        message_id=message["message_id"],
        sender=message["sender"],
        subject=message["subject"],
        priority=priority,
        category=category,
        reason=reason,
        next_action=next_action,
        draft=draft,
    )


def _policy_basis(policy_text: str, keywords: list[str]) -> list[str]:
    sentences = [part.strip() for part in re.split(r"(?<=[.!?])\s+|\n+", policy_text) if part.strip()]
    matches = [s for s in sentences if any(k.lower() in s.lower() for k in keywords)]
    selected = matches[:3] or sentences[:2]
    return [item[:240] for item in selected]


def _draft_for(
    message: dict[str, str],
    category: str,
    priority: str,
    policy_basis: list[str],
    next_action: str,
) -> DraftResult:
    greeting = "Hi there,"
    if "@" in message["sender"]:
        local = message["sender"].split("@", 1)[0].split("<")[-1]
        first = re.split(r"[._\-\s]", local.strip())[0].title()
        if first:
            greeting = f"Hi {first},"
    subject = message["subject"] if message["subject"].lower().startswith("re:") else f"Re: {message['subject']}"
    if category == "employee_relations":
        body = (
            f"{greeting}\n\nThank you for raising this. We take these concerns seriously and will route "
            "this to the appropriate People partner for confidential review. Please avoid sending additional "
            "sensitive details by email unless requested through the approved channel.\n\n"
            "A People team member will follow up with next steps.\n\nBest,\nPeople Team"
        )
        reason = "Employee-relations matters require a human People partner before any commitment."
    elif category == "payroll_privacy":
        body = (
            f"{greeting}\n\nThanks for reaching out. Because this may involve payroll or personal data, "
            "we should handle it through the secure HRIS/payroll channel rather than email. "
            "Please use the secure case path or wait for a People Ops teammate to confirm next steps.\n\n"
            "Best,\nPeople Team"
        )
        reason = "Payroll/personal-data replies must be checked for privacy and channel safety."
    elif category == "leave_request":
        body = (
            f"{greeting}\n\nThanks for the note. We can help review the leave request. Please confirm the "
            "requested dates and any supporting details required by policy. Final approval may depend on "
            "manager and People review, so we will confirm before making any commitment.\n\nBest,\nPeople Team"
        )
        reason = "Leave commitments need date verification and approval checks."
    elif category == "benefits":
        body = (
            f"{greeting}\n\nThanks for your benefits question. Based on our policy guidance, eligibility "
            "and enrollment details should be confirmed against the current plan documents. We can point you "
            "to the right resource and will escalate any exception request to People Ops.\n\nBest,\nPeople Team"
        )
        reason = "Benefits guidance should be verified against plan documents before sending."
    else:
        body = (
            f"{greeting}\n\nThanks for reaching out. We reviewed your note and the next step is: "
            f"{next_action}\n\nBest,\nPeople Team"
        )
        reason = "People-team replies should be reviewed before sending from a shared inbox."
    return DraftResult(
        to=message["sender"],
        subject=subject[:200],
        body=body,
        approval_required=True,
        approval_reason=reason,
        policy_basis=policy_basis,
    )


def _priority_rank(priority: str) -> int:
    return {"urgent": 0, "high": 1, "normal": 2, "low": 3}.get(priority, 9)


def _summary(queue: list[QueueItem]) -> str:
    counts: dict[str, int] = {}
    for queue_entry in queue:
        counts[queue_entry.priority] = counts.get(queue_entry.priority, 0) + 1
    parts = [f"{counts.get(name, 0)} {name}" for name in ["urgent", "high", "normal", "low"] if counts.get(name, 0)]
    return f"Prioritized {len(queue)} people-inbox message(s): " + ", ".join(parts) + "."


def _approval_boundary() -> str:
    return (
        "Drafts are approval-ready but not auto-sent. A People teammate must verify facts, "
        "policy citations, privacy constraints, and any commitment before sending."
    )