552 lines
21 KiB
Python
552 lines
21 KiB
Python
"""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 _queue_sort_key(queue_entry: QueueItem) -> int:
|
|
return _priority_rank(queue_entry.priority)
|
|
|
|
|
|
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: list[str] = []
|
|
for priority_name in ("urgent", "high", "normal", "low"):
|
|
priority_count = counts.get(priority_name, 0)
|
|
if priority_count:
|
|
parts.append(f"{priority_count} {priority_name}")
|
|
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."
|
|
)
|