"""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, InboundEmailPayload, 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 = [_triage_message(message, policy) for message in messages] queue.sort(key=lambda item: _priority_rank(item.priority)) summary = _summary(queue) report = { "generated_at": datetime.now(UTC).isoformat(), "summary": summary, "approval_boundary": _approval_boundary(), "queue": [item.model_dump(mode="json") for item in 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": sum(1 for item in queue if item.priority == "urgent"), "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: InboundEmailPayload, ) -> 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: # Also split common pasted inbox format with repeated From: headers. 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") or "unknown@example.com" subject = _field(block, "subject") or f"People inbox message {idx}" body = _field(block, "body") or _strip_headers(block) if len(body.strip()) < 8 and subject.startswith("People inbox message"): 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 item in queue: counts[item.priority] = counts.get(item.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." )