a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,533 @@
-"""production-proof-v116-high-utili-7255-2 agent.
-
-Starter stack:
-  - DeepAgents for tool-calling orchestration
-  - Caller-provided LLM credentials via ctx.llm
-  - A tiny model-call middleware hook you can replace with tracing,
-    routing, rate limits, or policy checks
-"""
+"""People-team shared inbox triage and policy-grounded draft assistant."""
 from __future__ import annotations
 
+import hashlib
 import json
-from pathlib import Path
-from typing import Any
+import os
+import re
+from datetime import UTC, datetime
+from typing import Any, Literal
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
+    AccountAccess,
+    AgentDatabase,
+    AgentDatabaseEnv,
+    AgentDatabaseMigrations,
+    AgentPlatformResources,
+    InboundEmailPayload,
     LLMProvisioning,
-    {{ auth_type }},
+    PlatformUserAuth,
     Pricing,
+    Resources,
     RunContext,
     WorkspaceAccess,
     WorkspaceMode,
 )
-from a2a_pack.context import LLMCreds
+
+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):
-    pass
+    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,
+    )
 
 
-SYSTEM_PROMPT = """\
-You are a compact tool-calling agent.
-
-Use the text_stats tool when the user asks about text, counts, summaries,
-or anything where exact length/word numbers would help. Mention tool results
-briefly instead of dumping raw JSON.
-"""
-
-RUNTIME_SKILLS_DIR = "production-proof-v116-high-utili-7255-2/.deepagents/skills/"
-DEEPAGENTS_RECURSION_LIMIT = 500
+class DraftResult(BaseModel):
+    to: str
+    subject: str
+    body: str
+    approval_required: bool
+    approval_reason: str
+    policy_basis: list[str]
 
 
-class ProductionProofV116HighUtili72552(A2AAgent[ProductionProofV116HighUtili72552Config, {{ auth_type }}]):
+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."
+    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 = {{ auth_type }}
+    auth_model = PlatformUserAuth
 
-    # Hosted generated agents read the caller's saved LLM credential through
-    # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
-    # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
-    # directly.
     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="Starter agent uses the caller's saved LLM credential via ctx.llm.",
+        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=64,
+        max_files=32,
         allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
         require_reason=False,
     )
-    tools_used = ("deepagents", "langchain")
-
-    @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful")
-    async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
-        creds = ctx.llm
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
-        if not creds.api_key:
-            return (
-                "LLM key required. Add an LLM credential in Settings > LLM "
-                "credentials before running this agent; for local --invoke "
-                "runs set AGENT_LLM_KEY."
+    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,
     )
-        graph = self._build_deep_agent(ctx=ctx, creds=creds)
-        state = await graph.ainvoke(
-            {"messages": [{"role": "user", "content": prompt}]},
-            config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT},
-        )
-        await ctx.emit_progress("deepagent finished")
-        return _last_message_text(state)
+    tools_used = ("postgres", "platform-mailbox", "mcp")
 
-    def _build_deep_agent(
+    @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[{{ auth_type }}],
-        creds: LLMCreds,
-    ) -> Any:
-        # Lazy imports keep `a2a card` usable before local dependencies are
-        # installed. `a2a deploy` installs requirements.txt during the build.
-        from a2a_pack.deepagents import create_a2a_deep_agent
-        from langchain.agents.middleware import wrap_model_call
-        from langchain_core.tools import tool
+        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."],
+            )
 
-        @tool
-        def text_stats(text: str) -> str:
-            """Return exact word, character, and line counts for text."""
-            words = [part for part in text.split() if part.strip()]
-            return json.dumps(
+        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(
             {
-                    "characters": len(text),
-                    "words": len(words),
-                    "lines": len(text.splitlines()) or 1,
+                "message_id": hashlib.sha1(f"{sender}|{subject}|{body}".encode("utf-8")).hexdigest()[:12],
+                "sender": sender[:200],
+                "subject": subject[:200],
+                "body": body[:4000],
             }
         )
+    return out
 
-        @wrap_model_call
-        async def log_model_call(request: Any, handler: Any) -> Any:
-            messages = request.state.get("messages", [])
-            print(
-                "[middleware] model_call "
-                f"model={creds.model} source={creds.source} messages={len(messages)}"
-            )
-            return await handler(request)
 
-        backend = ctx.workspace_backend()
-        skill_sources = _seed_runtime_skills(backend, ctx)
-        # create_a2a_deep_agent resolves provider:model strings with
-        # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing,
-        # provider-specific extra body, and runtime model overrides.
-        return create_a2a_deep_agent(
-            ctx,
-            creds=creds,
-            backend=backend,
-            skills=skill_sources or None,
-            tools=[text_stats],
-            middleware=[log_model_call],
-            system_prompt=SYSTEM_PROMPT,
+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 _runtime_skills_root(ctx: RunContext[Any]) -> str:
-    workspace = getattr(ctx, "_workspace", None)
-    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
-    if not prefixes:
-        outputs_prefix = getattr(workspace, "outputs_prefix", None)
-        prefixes = (outputs_prefix or "outputs/",)
-    prefix = str(prefixes[0]).strip("/")
-    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
+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 _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
-    """Copy packaged DeepAgents skills into the invocation workspace.
-
-    DeepAgents loads skills from its backend, while source-controlled
-    ``skills/`` folders live in the image. This bridge lets generated agents
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
-    files.
-    """
-    root = Path(__file__).parent / "skills"
-    if not root.exists():
-        return []
-    runtime_skills_root = _runtime_skills_root(ctx)
-    uploads: list[tuple[str, bytes]] = []
-    for path in root.rglob("*"):
-        if path.is_file():
-            rel = path.relative_to(root).as_posix()
-            uploads.append((runtime_skills_root + rel, path.read_bytes()))
-    if uploads:
-        backend.upload_files(uploads)
-        return [runtime_skills_root]
-    return []
+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 _last_message_text(state: dict[str, Any]) -> str:
-    messages = state.get("messages") or []
-    if not messages:
-        return json.dumps(state, default=str)
+def _priority_rank(priority: str) -> int:
+    return {"urgent": 0, "high": 1, "normal": 2, "low": 3}.get(priority, 9)
 
-    content = getattr(messages[-1], "content", None)
-    if isinstance(content, str):
-        return content
-    if isinstance(content, list):
-        parts: list[str] = []
-        for item in content:
-            if isinstance(item, dict):
-                text = item.get("text") or item.get("content")
-                if text:
-                    parts.append(str(text))
-            elif item:
-                parts.append(str(item))
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
-    return str(content or messages[-1])
+
+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."
+    )