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, Literal

import httpx
from pydantic import BaseModel, Field

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


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",
        ),
    )
    platform_resources = AgentPlatformResources(mailbox=True)
    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: InboundEmailPayload,
    ) -> 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("livemode") is False:
        # Test mode is allowed for dry-run eligibility, but still requires approval for execution.
        pass
    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:
        # Artifact fallback keeps an audit copy even if durable workspace access is unavailable.
        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())