receives-support-email-checks-1-919313/agent.py

"""Support email refund workflow agent.

Receives support email, checks Stripe refund eligibility with caller-provided
credentials, drafts a customer reply, requires explicit approval before any
refund mutation, and posts redacted audit summaries to Slack.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import re
import time
from email.utils import parseaddr
from typing import Any

import httpx
from pydantic import BaseModel, Field

import a2a_pack as a2a
from a2a_pack import (
    A2AAgent,
    ConsumerSetup,
    ConsumerSetupField,
    EgressPolicy,
    NoAuth,
    Pricing,
    Resources,
    RunContext,
    WorkspaceAccess,
    WorkspaceMode,
)


LEDGER_PATH = "outputs/support-refund-audit-ledger.json"
MAX_EMAIL_BODY_CHARS = 12_000
MAX_REFUND_AMOUNT_CENTS = 100_000
DEFAULT_CURRENCY = "usd"
HTTP_TIMEOUT = httpx.Timeout(12.0, connect=4.0, read=8.0, write=8.0)


class ReceivesSupportEmailChecks1919313Config(BaseModel):
    max_auto_eligible_refund_cents: int = Field(
        default=50_000,
        ge=1,
        le=MAX_REFUND_AMOUNT_CENTS,
        description="Maximum refund amount the agent may mark eligible for approval.",
    )
    allowed_refund_currencies: list[str] = Field(
        default_factory=lambda: [DEFAULT_CURRENCY],
        description="Lowercase ISO currencies that may be proposed for refunds.",
    )


class ReceivesSupportEmailChecks1919313(A2AAgent[ReceivesSupportEmailChecks1919313Config, NoAuth]):
    name = "receives-support-email-checks-1-919313"
    description = (
        "Receives support email, checks Stripe refund eligibility with caller-provided "
        "credentials, drafts the customer reply, requires explicit approval before "
        "executing refunds, and posts redacted audit summaries to Slack."
    )
    version = "0.1.0"

    config_model = ReceivesSupportEmailChecks1919313Config
    auth_model = NoAuth

    consumer_setup = ConsumerSetup.from_fields(
        ConsumerSetupField.secret(
            "STRIPE_SECRET_KEY",
            label="Stripe secret key",
            description="Caller-provided Stripe API key used only server-side to inspect and create refunds.",
            input_type="password",
        ),
        ConsumerSetupField.secret(
            "SLACK_BOT_TOKEN",
            label="Slack bot token",
            description="Caller-provided Slack token used only server-side for audit summaries.",
            input_type="password",
        ),
        ConsumerSetupField.config(
            "SLACK_CHANNEL_ID",
            label="Slack audit channel ID",
            description="Slack channel where refund audit summaries should be posted.",
            input_type="text",
        ),
        ConsumerSetupField.config(
            "STRIPE_BASE_URL",
            label="Stripe API base URL",
            description="Optional test override. Defaults to https://api.stripe.com.",
            required=False,
            input_type="url",
        ),
        ConsumerSetupField.config(
            "SLACK_BASE_URL",
            label="Slack API base URL",
            description="Optional test override. Defaults to https://slack.com/api.",
            required=False,
            input_type="url",
        ),
    )
    resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300)
    workspace_access = WorkspaceAccess.dynamic(
        max_files=8,
        allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
        require_reason=False,
    )
    egress = EgressPolicy(allow_hosts=("api.stripe.com", "slack.com"))
    tools_used = ("email", "stripe", "slack", "httpx")
    pricing = Pricing(
        price_per_call_usd=0.0,
        caller_pays_llm=False,
        notes="Deterministic support refund workflow. Provider API usage is paid through caller-provided Stripe and Slack credentials.",
    )

    @a2a.tool(
        description="Handle one inbound support email: dedupe, check Stripe refund eligibility, draft reply, and post a Slack audit summary without executing a refund.",
        timeout_seconds=120,
        idempotent=True,
        max_retries=0,
        cost_class="external-api",
    )
    async def process_support_email(
        self,
        ctx: RunContext[NoAuth],
        sender_email: str,
        subject: str,
        body: str,
        message_id: str,
        stripe_payment_intent_id: str,
        requested_refund_amount_cents: int,
        currency: str = DEFAULT_CURRENCY,
        customer_name: str = "customer",
    ) -> dict[str, Any]:
        """Preview a refund workflow from bounded, typed email fields."""
        email = {
            "sender": sender_email,
            "subject": subject,
            "body": body[:MAX_EMAIL_BODY_CHARS],
            "message_id": message_id,
            "references": [],
            "attachments": [],
            "date": None,
        }
        return await self._process_email_payload(
            ctx,
            email=email,
            stripe_payment_intent_id=stripe_payment_intent_id,
            requested_refund_amount_cents=requested_refund_amount_cents,
            currency=currency,
            customer_name=customer_name,
        )

    @a2a.tool(
        description="Execute a previously proposed Stripe refund only when the caller supplies explicit approval bound to the exact action digest.",
        timeout_seconds=120,
        idempotent=True,
        max_retries=0,
        cost_class="payment-mutation",
    )
    async def execute_approved_refund(
        self,
        ctx: RunContext[NoAuth],
        action_digest: str,
        approval_token: str,
        stripe_payment_intent_id: str,
        amount_cents: int,
        currency: str = DEFAULT_CURRENCY,
        customer_email: str = "",
        reason: str = "requested_by_customer",
    ) -> dict[str, Any]:
        """Execute one approved refund with a Stripe idempotency key."""
        setup = _read_setup(ctx)
        if setup["status"] != "ready":
            return setup

        normalized = _normalize_refund_action(
            payment_intent_id=stripe_payment_intent_id,
            amount_cents=amount_cents,
            currency=currency,
            customer_email=customer_email,
            reason=reason,
        )
        expected_digest = _action_digest(normalized)
        if expected_digest != action_digest:
            return {
                "status": "declined",
                "decline_code": "tampered_action",
                "message": "The approval digest does not match the requested refund action.",
                "expected_action_digest": expected_digest,
                "provided_action_digest": action_digest,
            }
        expected_token = _approval_phrase(action_digest)
        if approval_token != expected_token:
            return {
                "status": "approval_required",
                "approval_required": True,
                "action_digest": action_digest,
                "approval_instructions": f"To execute, submit approval_token exactly as: {expected_token}",
            }

        ledger = await _load_ledger(ctx)
        idempotency_key = _idempotency_key("refund", action_digest)
        if idempotency_key in ledger.get("refunds", {}):
            previous = ledger["refunds"][idempotency_key]
            return {
                "status": "duplicate_execution",
                "executed": previous.get("executed", False),
                "idempotency_key": idempotency_key,
                "provider_refund_id": previous.get("provider_refund_id"),
                "audit": previous.get("audit"),
            }

        await ctx.emit_progress("Creating approved Stripe refund with idempotency key")
        stripe_result = await _create_stripe_refund(setup, normalized, idempotency_key)
        now = _now()
        audit = {
            "event": "refund_execution",
            "status": stripe_result["status"],
            "action_digest": action_digest,
            "idempotency_key": idempotency_key,
            "payment_intent_id": stripe_payment_intent_id,
            "amount_cents": amount_cents,
            "currency": currency.lower(),
            "customer_email_hash": _hash_customer(customer_email),
            "timestamp": now,
        }
        if stripe_result["status"] == "executed":
            audit["provider_refund_id"] = stripe_result.get("provider_refund_id")
        slack_result = await _post_slack_audit(setup, audit)
        audit["slack_status"] = slack_result["status"]

        ledger.setdefault("refunds", {})[idempotency_key] = {
            "executed": stripe_result["status"] == "executed",
            "provider_refund_id": stripe_result.get("provider_refund_id"),
            "audit": audit,
            "created_at": now,
        }
        await _save_ledger(ctx, ledger)

        if stripe_result["status"] == "timeout":
            return {
                "status": "needs_reconciliation",
                "executed": False,
                "idempotency_key": idempotency_key,
                "action_digest": action_digest,
                "message": "Stripe timed out. Reconcile by checking Stripe for the idempotency key before retrying.",
                "audit": audit,
                "slack": slack_result,
            }
        if stripe_result["status"] != "executed":
            return {
                "status": "declined",
                "executed": False,
                "decline_code": stripe_result.get("error_type", "provider_error"),
                "idempotency_key": idempotency_key,
                "action_digest": action_digest,
                "audit": audit,
                "slack": slack_result,
            }
        return {
            "status": "executed",
            "executed": True,
            "provider_refund_id": stripe_result.get("provider_refund_id"),
            "idempotency_key": idempotency_key,
            "action_digest": action_digest,
            "audit": audit,
            "slack": slack_result,
        }

    @a2a.tool(
        description="Inbound email handler for the agent mailbox. It deduplicates and returns a draft reply; refund execution still requires a separate explicit approval call.",
        on_email=True,
        timeout_seconds=120,
        idempotent=True,
        max_retries=0,
        cost_class="external-api",
    )
    async def receive_support_email(
        self,
        ctx: RunContext[NoAuth],
        email: dict[str, Any],
    ) -> dict[str, str] | None:
        parsed = _extract_refund_request(email)
        if not parsed.get("stripe_payment_intent_id"):
            return {
                "subject": _reply_subject(email.get("subject", "Support request")),
                "body": (
                    "Thanks for contacting support. We could not identify a Stripe payment intent "
                    "in your message, so a teammate will review this manually."
                ),
            }
        result = await self._process_email_payload(
            ctx,
            email=email,
            stripe_payment_intent_id=parsed["stripe_payment_intent_id"],
            requested_refund_amount_cents=int(parsed.get("amount_cents") or 0),
            currency=str(parsed.get("currency") or DEFAULT_CURRENCY),
            customer_name=parsed.get("customer_name") or "customer",
        )
        draft = result.get("draft_reply") or {}
        body = str(draft.get("body") or "Thanks for contacting support. We are reviewing your refund request.")
        return {"subject": str(draft.get("subject") or _reply_subject(email.get("subject", "Support request"))), "body": body}

    async def _process_email_payload(
        self,
        ctx: RunContext[NoAuth],
        *,
        email: dict[str, Any],
        stripe_payment_intent_id: str,
        requested_refund_amount_cents: int,
        currency: str,
        customer_name: str,
    ) -> dict[str, Any]:
        setup = _read_setup(ctx)
        if setup["status"] != "ready":
            return setup
        validation = _validate_refund_request(
            stripe_payment_intent_id=stripe_payment_intent_id,
            amount_cents=requested_refund_amount_cents,
            currency=currency,
            allowed_currencies=self.config.allowed_refund_currencies,
            max_refund_cents=self.config.max_auto_eligible_refund_cents,
        )
        if validation["status"] != "valid":
            return validation

        message_id = str(email.get("message_id") or "").strip()
        sender = _sender_email(str(email.get("sender") or ""))
        email_key = _email_key(message_id, sender, stripe_payment_intent_id)
        ledger = await _load_ledger(ctx)
        if email_key in ledger.get("emails", {}):
            previous = ledger["emails"][email_key]
            return {
                "status": "duplicate_email",
                "duplicate": True,
                "message_id": message_id,
                "previous_action_digest": previous.get("action_digest"),
                "draft_reply": previous.get("draft_reply"),
                "audit": previous.get("audit"),
            }

        await ctx.emit_progress("Checking Stripe payment intent refund eligibility")
        stripe_check = await _check_stripe_payment_intent(setup, stripe_payment_intent_id)
        policy = _evaluate_refund_policy(
            stripe_check=stripe_check,
            requested_amount_cents=requested_refund_amount_cents,
            currency=currency,
            max_refund_cents=self.config.max_auto_eligible_refund_cents,
        )
        normalized_action = _normalize_refund_action(
            payment_intent_id=stripe_payment_intent_id,
            amount_cents=requested_refund_amount_cents,
            currency=currency,
            customer_email=sender,
            reason="requested_by_customer",
        )
        action_digest = _action_digest(normalized_action)
        draft_reply = _draft_customer_reply(
            sender=sender,
            subject=str(email.get("subject") or "Support request"),
            customer_name=customer_name,
            policy=policy,
            amount_cents=requested_refund_amount_cents,
            currency=currency,
            action_digest=action_digest,
        )
        audit = {
            "event": "refund_preview",
            "status": policy["status"],
            "message_id_hash": _stable_hash(message_id),
            "customer_email_hash": _hash_customer(sender),
            "payment_intent_id": stripe_payment_intent_id,
            "amount_cents": requested_refund_amount_cents,
            "currency": currency.lower(),
            "action_digest": action_digest,
            "approval_required": policy.get("approval_required", True),
            "timestamp": _now(),
        }
        slack_result = await _post_slack_audit(setup, audit)
        audit["slack_status"] = slack_result["status"]

        ledger.setdefault("emails", {})[email_key] = {
            "action_digest": action_digest,
            "draft_reply": draft_reply,
            "audit": audit,
            "created_at": _now(),
        }
        await _save_ledger(ctx, ledger)

        return {
            "status": policy["status"],
            "duplicate": False,
            "approval_required": True,
            "executed": False,
            "action_digest": action_digest,
            "idempotency_key_preview": _idempotency_key("refund", action_digest),
            "approval_instructions": (
                "Review the draft and policy result. To execute the refund, call "
                "execute_approved_refund with this action_digest and approval_token "
                f"exactly: {_approval_phrase(action_digest)}"
            ),
            "policy": policy,
            "draft_reply": draft_reply,
            "audit": audit,
            "slack": slack_result,
        }


def _read_setup(ctx: RunContext[Any]) -> dict[str, Any]:
    missing: list[str] = []
    values: dict[str, str] = {}
    for name in ("STRIPE_SECRET_KEY", "SLACK_BOT_TOKEN"):
        try:
            values[name] = ctx.consumer_secret(name)
        except Exception:
            missing.append(name)
    slack_channel = str(ctx.consumer_config("SLACK_CHANNEL_ID", "") or "").strip()
    if not slack_channel:
        missing.append("SLACK_CHANNEL_ID")
    if missing:
        return {
            "status": "setup_required",
            "missing": sorted(set(missing)),
            "message": "Configure caller-provided Stripe and Slack setup before running this workflow.",
        }
    return {
        "status": "ready",
        "stripe_key": values["STRIPE_SECRET_KEY"],
        "slack_token": values["SLACK_BOT_TOKEN"],
        "slack_channel": slack_channel,
        "stripe_base_url": _clean_base_url(ctx.consumer_config("STRIPE_BASE_URL", "https://api.stripe.com")),
        "slack_base_url": _clean_base_url(ctx.consumer_config("SLACK_BASE_URL", "https://slack.com/api")),
    }


def _clean_base_url(value: Any) -> str:
    text = str(value or "").strip().rstrip("/")
    return text or "https://api.stripe.com"


def _validate_refund_request(
    *,
    stripe_payment_intent_id: str,
    amount_cents: int,
    currency: str,
    allowed_currencies: list[str],
    max_refund_cents: int,
) -> dict[str, Any]:
    if not re.fullmatch(r"pi_[A-Za-z0-9_]+", stripe_payment_intent_id or ""):
        return {"status": "declined", "decline_code": "invalid_payment_intent", "executed": False}
    currency_clean = (currency or "").lower().strip()
    if currency_clean not in {c.lower() for c in allowed_currencies}:
        return {"status": "declined", "decline_code": "unsupported_currency", "currency": currency_clean, "executed": False}
    if amount_cents <= 0 or amount_cents > min(max_refund_cents, MAX_REFUND_AMOUNT_CENTS):
        return {"status": "declined", "decline_code": "amount_out_of_policy", "max_refund_cents": max_refund_cents, "executed": False}
    return {"status": "valid"}


async def _check_stripe_payment_intent(setup: dict[str, Any], payment_intent_id: str) -> dict[str, Any]:
    url = f"{setup['stripe_base_url']}/v1/payment_intents/{payment_intent_id}"
    try:
        async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
            resp = await client.get(
                url,
                headers={"Authorization": f"Bearer {setup['stripe_key']}"},
                params={"expand[]": "latest_charge"},
            )
    except (httpx.TimeoutException, asyncio.TimeoutError):
        return {"status": "timeout", "error_type": "provider_timeout"}
    except httpx.HTTPError as exc:
        return {"status": "provider_error", "error_type": type(exc).__name__}
    if resp.status_code in {401, 403}:
        return {"status": "auth_error", "error_type": "stripe_auth"}
    if resp.status_code == 404:
        return {"status": "not_found", "error_type": "payment_intent_not_found"}
    if resp.status_code == 429:
        return {"status": "rate_limited", "error_type": "stripe_rate_limit"}
    if resp.status_code >= 400:
        return {"status": "provider_error", "error_type": "stripe_error", "status_code": resp.status_code}
    try:
        data = resp.json()
    except ValueError:
        return {"status": "provider_error", "error_type": "malformed_json"}
    return {
        "status": "ok",
        "payment_intent_status": data.get("status"),
        "amount_received": int(data.get("amount_received") or data.get("amount") or 0),
        "amount_refunded": _amount_refunded(data),
        "currency": str(data.get("currency") or "").lower(),
        "livemode": bool(data.get("livemode")),
    }


def _amount_refunded(payment_intent: dict[str, Any]) -> int:
    charge = payment_intent.get("latest_charge")
    if isinstance(charge, dict):
        return int(charge.get("amount_refunded") or 0)
    charges = payment_intent.get("charges")
    if isinstance(charges, dict):
        data = charges.get("data") or []
        if data and isinstance(data[0], dict):
            return int(data[0].get("amount_refunded") or 0)
    return 0


def _evaluate_refund_policy(
    *,
    stripe_check: dict[str, Any],
    requested_amount_cents: int,
    currency: str,
    max_refund_cents: int,
) -> dict[str, Any]:
    if stripe_check["status"] == "timeout":
        return {"status": "needs_reconciliation", "approval_required": True, "reason": "Stripe timed out; reconcile before retrying."}
    if stripe_check["status"] != "ok":
        return {"status": "declined", "approval_required": True, "reason": stripe_check.get("error_type", "stripe_lookup_failed")}
    if stripe_check.get("currency") != currency.lower():
        return {"status": "declined", "approval_required": True, "reason": "Currency mismatch."}
    if stripe_check.get("payment_intent_status") not in {"succeeded", "requires_capture"}:
        return {"status": "declined", "approval_required": True, "reason": "Payment intent is not in a refundable state."}
    refundable = max(0, int(stripe_check.get("amount_received") or 0) - int(stripe_check.get("amount_refunded") or 0))
    if requested_amount_cents > refundable:
        return {"status": "declined", "approval_required": True, "reason": "Requested refund exceeds remaining refundable amount.", "refundable_amount_cents": refundable}
    if requested_amount_cents > max_refund_cents:
        return {"status": "declined", "approval_required": True, "reason": "Requested refund exceeds configured policy limit."}
    return {"status": "proposed", "approval_required": True, "refundable_amount_cents": refundable, "reason": "Eligible for an approved refund."}


async def _create_stripe_refund(setup: dict[str, Any], action: dict[str, Any], idempotency_key: str) -> dict[str, Any]:
    try:
        async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
            resp = await client.post(
                f"{setup['stripe_base_url']}/v1/refunds",
                headers={
                    "Authorization": f"Bearer {setup['stripe_key']}",
                    "Idempotency-Key": idempotency_key,
                },
                data={
                    "payment_intent": action["payment_intent_id"],
                    "amount": str(action["amount_cents"]),
                    "reason": action["reason"],
                    "metadata[action_digest]": _action_digest(action),
                },
            )
    except (httpx.TimeoutException, asyncio.TimeoutError):
        return {"status": "timeout", "error_type": "provider_timeout"}
    except httpx.HTTPError as exc:
        return {"status": "provider_error", "error_type": type(exc).__name__}
    if resp.status_code in {401, 403}:
        return {"status": "provider_error", "error_type": "stripe_auth"}
    if resp.status_code == 429:
        return {"status": "provider_error", "error_type": "stripe_rate_limit"}
    if resp.status_code >= 400:
        return {"status": "provider_error", "error_type": "stripe_error", "status_code": resp.status_code}
    try:
        data = resp.json()
    except ValueError:
        return {"status": "provider_error", "error_type": "malformed_json"}
    return {"status": "executed", "provider_refund_id": data.get("id"), "provider_status": data.get("status")}


async def _post_slack_audit(setup: dict[str, Any], audit: dict[str, Any]) -> dict[str, Any]:
    payload = {
        "channel": setup["slack_channel"],
        "text": _slack_audit_text(audit),
        "unfurl_links": False,
        "unfurl_media": False,
    }
    try:
        async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
            resp = await client.post(
                f"{setup['slack_base_url']}/chat.postMessage",
                headers={"Authorization": f"Bearer {setup['slack_token']}", "Content-Type": "application/json"},
                json=payload,
            )
    except (httpx.TimeoutException, asyncio.TimeoutError):
        return {"status": "timeout", "message": "Slack audit post timed out."}
    except httpx.HTTPError as exc:
        return {"status": "provider_error", "error_type": type(exc).__name__}
    if resp.status_code in {401, 403}:
        return {"status": "auth_error"}
    if resp.status_code == 429:
        return {"status": "rate_limited"}
    if resp.status_code >= 400:
        return {"status": "provider_error", "status_code": resp.status_code}
    try:
        data = resp.json()
    except ValueError:
        return {"status": "provider_error", "error_type": "malformed_json"}
    if data.get("ok") is not True:
        return {"status": "provider_error", "error_type": str(data.get("error") or "slack_error")}
    return {"status": "posted", "message_ts": data.get("ts"), "channel": data.get("channel")}


def _slack_audit_text(audit: dict[str, Any]) -> str:
    return (
        f"Support refund audit: {audit.get('event')} status={audit.get('status')} "
        f"pi={audit.get('payment_intent_id')} amount={audit.get('amount_cents')} {audit.get('currency')} "
        f"digest={audit.get('action_digest')} customer_hash={audit.get('customer_email_hash')}"
    )


def _draft_customer_reply(*, sender: str, subject: str, customer_name: str, policy: dict[str, Any], amount_cents: int, currency: str, action_digest: str) -> dict[str, str]:
    dollars = f"{amount_cents / 100:.2f} {currency.upper()}"
    greeting = f"Hi {customer_name.strip() or 'there'},"
    if policy["status"] == "proposed":
        body = (
            f"{greeting}\n\nThanks for contacting support. We found your payment and your requested refund of {dollars} "
            "appears eligible under our refund policy. For your protection, no refund has been issued yet; "
            "a support teammate must approve the exact refund action first.\n\n"
            f"Reference: {action_digest}\n\nWe will follow up once the review is complete."
        )
    elif policy["status"] == "needs_reconciliation":
        body = (
            f"{greeting}\n\nThanks for contacting support. We are reviewing your refund request, but our payment provider "
            "did not respond in time. A teammate will reconcile the payment record before taking any action."
        )
    else:
        body = (
            f"{greeting}\n\nThanks for contacting support. We reviewed the refund request and cannot automatically mark it "
            f"eligible because: {policy.get('reason', 'it requires manual review')}. A teammate will review it manually."
        )
    return {"to": sender, "subject": _reply_subject(subject), "body": body}


async def _load_ledger(ctx: RunContext[Any]) -> dict[str, Any]:
    try:
        view = await ctx.workspace.open_view(
            purpose="read support refund audit ledger",
            hints=[LEDGER_PATH],
            max_files=1,
            mode=WorkspaceMode.READ_WRITE_OVERLAY,
            reason="Deduplicate emails and refund executions.",
        )
        for f in view.files:
            if f.path == LEDGER_PATH:
                raw = await view.read(LEDGER_PATH)
                return json.loads(raw.decode("utf-8"))
    except Exception:
        pass
    return {"emails": {}, "refunds": {}}


async def _save_ledger(ctx: RunContext[Any], ledger: dict[str, Any]) -> None:
    data = json.dumps(ledger, sort_keys=True, indent=2).encode("utf-8")
    try:
        grant = await ctx.workspace.request_access(
            files=[LEDGER_PATH],
            mode=WorkspaceMode.READ_WRITE_OVERLAY,
            reason="Persist support email and refund idempotency ledger.",
            purpose="write support refund audit ledger",
        )
        view = await ctx.workspace.open_view(
            purpose="write support refund audit ledger",
            hints=[LEDGER_PATH],
            max_files=1,
            mode=WorkspaceMode.READ_WRITE_OVERLAY,
            reason=f"Use grant {grant.grant_id} to write audit ledger.",
        )
        await view.write(LEDGER_PATH, data)
    except Exception:
        ref = await ctx.write_artifact("support-refund-audit-ledger.json", data, "application/json")
        await ctx.emit_artifact(ref)


def _normalize_refund_action(*, payment_intent_id: str, amount_cents: int, currency: str, customer_email: str, reason: str) -> dict[str, Any]:
    return {
        "payment_intent_id": payment_intent_id.strip(),
        "amount_cents": int(amount_cents),
        "currency": currency.lower().strip(),
        "customer_email_hash": _hash_customer(customer_email),
        "reason": reason.strip() or "requested_by_customer",
    }


def _action_digest(action: dict[str, Any]) -> str:
    return hashlib.sha256(json.dumps(action, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()


def _approval_phrase(action_digest: str) -> str:
    return f"APPROVE-REFUND-{action_digest[:16]}"


def _idempotency_key(prefix: str, digest: str) -> str:
    return f"a2a-{prefix}-{digest[:32]}"


def _email_key(message_id: str, sender: str, payment_intent_id: str) -> str:
    return _stable_hash("|".join([message_id.strip().lower(), sender.strip().lower(), payment_intent_id.strip()]))


def _hash_customer(value: str) -> str:
    return _stable_hash((value or "").strip().lower())[:16]


def _stable_hash(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def _sender_email(value: str) -> str:
    parsed = parseaddr(value)[1]
    return parsed or value.strip()


def _reply_subject(subject: str) -> str:
    text = (subject or "Support request").strip()
    return text if text.lower().startswith("re:") else f"Re: {text}"


def _extract_refund_request(email: dict[str, Any]) -> dict[str, Any]:
    text = f"{email.get('subject', '')}\n{email.get('body', '')}"[:MAX_EMAIL_BODY_CHARS]
    pi_match = re.search(r"\b(pi_[A-Za-z0-9_]+)\b", text)
    amount_match = re.search(r"(?:\$|USD\s*)?(\d+(?:\.\d{1,2})?)\s*(usd|USD)?", text)
    cents = int(float(amount_match.group(1)) * 100) if amount_match else 0
    sender = _sender_email(str(email.get("sender") or ""))
    name = parseaddr(str(email.get("sender") or ""))[0] or "customer"
    return {
        "stripe_payment_intent_id": pi_match.group(1) if pi_match else "",
        "amount_cents": cents,
        "currency": "usd",
        "customer_email": sender,
        "customer_name": name,
    }


def _now() -> int:
    return int(time.time())