a2a-source-edit: write agent.py

agent.py

@@ -1,189 +1,734 @@
-"""receives-support-email-checks-1-919313 agent.
+"""Support email refund workflow agent.
 
-Starter stack:
+Receives support email, checks Stripe refund eligibility with caller-provided
-  - DeepAgents for tool-calling orchestration
+credentials, drafts a customer reply, requires explicit approval before any
-  - Caller-provided LLM credentials via ctx.llm
+refund mutation, and posts redacted audit summaries to Slack.
-  - A tiny model-call middleware hook you can replace with tracing,
-    routing, rate limits, or policy checks
 """
 from __future__ import annotations
 
+import asyncio
+import hashlib
 import json
-from pathlib import Path
+import re
-from typing import Any
+import time
+from email.utils import parseaddr
+from typing import Any, Literal
 
-from pydantic import BaseModel
+import httpx
+from pydantic import BaseModel, Field
 
 import a2a_pack as a2a
 from a2a_pack import (
     A2AAgent,
-    LLMProvisioning,
+    AgentPlatformResources,
-    {{ auth_type }},
+    ConsumerSetup,
+    ConsumerSetupField,
+    EgressPolicy,
+    NoAuth,
     Pricing,
+    Resources,
     RunContext,
     WorkspaceAccess,
     WorkspaceMode,
 )
-from a2a_pack.context import LLMCreds
+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):
-    pass
+    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.",
+    )
 
 
-SYSTEM_PROMPT = """\
+class ReceivesSupportEmailChecks1919313(A2AAgent[ReceivesSupportEmailChecks1919313Config, NoAuth]):
-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 = "receives-support-email-checks-1-919313/.deepagents/skills/"
-DEEPAGENTS_RECURSION_LIMIT = 500
-
-
-class ReceivesSupportEmailChecks1919313(A2AAgent[ReceivesSupportEmailChecks1919313Config, {{ auth_type }}]):
     name = "receives-support-email-checks-1-919313"
-    description = "Receives support email, checks Stripe refund eligibility, drafts replies, requires approval before refunds, and posts audit summaries to Slack."
+    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 = {{ auth_type }}
+    auth_model = NoAuth
 
-    # Hosted generated agents read the caller's saved LLM credential through
+    consumer_setup = ConsumerSetup.from_fields(
-    # ctx.llm. The platform may proxy that credential through LiteLLM, but agent
+        ConsumerSetupField.secret(
-    # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY
+            "STRIPE_SECRET_KEY",
-    # directly.
+            label="Stripe secret key",
-    llm_provisioning = LLMProvisioning.PLATFORM
+            description="Caller-provided Stripe API key used only server-side to inspect and create refunds.",
-    pricing = Pricing(
+            input_type="password",
-        price_per_call_usd=0.0,
+        ),
-        caller_pays_llm=True,
+        ConsumerSetupField.secret(
-        notes="Starter agent uses the caller's saved LLM credential via ctx.llm.",
+            "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=64,
+        max_files=8,
         allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
         require_reason=False,
     )
-    tools_used = ("deepagents", "langchain")
+    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="Ask the starter DeepAgent to answer with tool calls when useful")
+    @a2a.tool(
-    async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str:
+        description="Handle one inbound support email: dedupe, check Stripe refund eligibility, draft reply, and post a Slack audit summary without executing a refund.",
-        creds = ctx.llm
+        timeout_seconds=120,
-        await ctx.emit_progress(f"llm: {creds.model} via {creds.source}")
+        idempotent=True,
-        if not creds.api_key:
+        max_retries=0,
-            return (
+        cost_class="external-api",
-                "LLM key required. Add an LLM credential in Settings > LLM "
+    )
-                "credentials before running this agent; for local --invoke "
+    async def process_support_email(
-                "runs set AGENT_LLM_KEY."
-            )
-        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)
-
-    def _build_deep_agent(
         self,
-        *,
+        ctx: RunContext[NoAuth],
-        ctx: RunContext[{{ auth_type }}],
+        sender_email: str,
-        creds: LLMCreds,
+        subject: str,
-    ) -> Any:
+        body: str,
-        # Lazy imports keep `a2a card` usable before local dependencies are
+        message_id: str,
-        # installed. `a2a deploy` installs requirements.txt during the build.
+        stripe_payment_intent_id: str,
-        from a2a_pack.deepagents import create_a2a_deep_agent
+        requested_refund_amount_cents: int,
-        from langchain.agents.middleware import wrap_model_call
+        currency: str = DEFAULT_CURRENCY,
-        from langchain_core.tools import tool
+        customer_name: str = "customer",
-
+    ) -> dict[str, Any]:
-        @tool
+        """Preview a refund workflow from bounded, typed email fields."""
-        def text_stats(text: str) -> str:
+        email = {
-            """Return exact word, character, and line counts for text."""
+            "sender": sender_email,
-            words = [part for part in text.split() if part.strip()]
+            "subject": subject,
-            return json.dumps(
+            "body": body[:MAX_EMAIL_BODY_CHARS],
-                {
+            "message_id": message_id,
-                    "characters": len(text),
+            "references": [],
-                    "words": len(words),
+            "attachments": [],
-                    "lines": len(text.splitlines()) or 1,
+            "date": None,
-                }
+        }
-            )
+        return await self._process_email_payload(
-
-        @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,
+            email=email,
-            backend=backend,
+            stripe_payment_intent_id=stripe_payment_intent_id,
-            skills=skill_sources or None,
+            requested_refund_amount_cents=requested_refund_amount_cents,
-            tools=[text_stats],
+            currency=currency,
-            middleware=[log_model_call],
+            customer_name=customer_name,
-            system_prompt=SYSTEM_PROMPT,
         )
 
+    @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
 
-def _runtime_skills_root(ctx: RunContext[Any]) -> str:
+        normalized = _normalize_refund_action(
-    workspace = getattr(ctx, "_workspace", None)
+            payment_intent_id=stripe_payment_intent_id,
-    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
+            amount_cents=amount_cents,
-    if not prefixes:
+            currency=currency,
-        outputs_prefix = getattr(workspace, "outputs_prefix", None)
+            customer_email=customer_email,
-        prefixes = (outputs_prefix or "outputs/",)
+            reason=reason,
-    prefix = str(prefixes[0]).strip("/")
+        )
-    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
+        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 _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]:
+def _read_setup(ctx: RunContext[Any]) -> dict[str, Any]:
-    """Copy packaged DeepAgents skills into the invocation workspace.
+    missing: list[str] = []
-
+    values: dict[str, str] = {}
-    DeepAgents loads skills from its backend, while source-controlled
+    for name in ("STRIPE_SECRET_KEY", "SLACK_BOT_TOKEN"):
-    ``skills/`` folders live in the image. This bridge lets generated agents
+        try:
-    ship reusable SKILL.md bundles without giving up durable A2A workspace
+            values[name] = ctx.consumer_secret(name)
-    files.
+        except Exception:
-    """
+            missing.append(name)
-    root = Path(__file__).parent / "skills"
+    slack_channel = str(ctx.consumer_config("SLACK_CHANNEL_ID", "") or "").strip()
-    if not root.exists():
+    if not slack_channel:
-        return []
+        missing.append("SLACK_CHANNEL_ID")
-    runtime_skills_root = _runtime_skills_root(ctx)
+    if missing:
-    uploads: list[tuple[str, bytes]] = []
+        return {
-    for path in root.rglob("*"):
+            "status": "setup_required",
-        if path.is_file():
+            "missing": sorted(set(missing)),
-            rel = path.relative_to(root).as_posix()
+            "message": "Configure caller-provided Stripe and Slack setup before running this workflow.",
-            uploads.append((runtime_skills_root + rel, path.read_bytes()))
+        }
-    if uploads:
+    return {
-        backend.upload_files(uploads)
+        "status": "ready",
-        return [runtime_skills_root]
+        "stripe_key": values["STRIPE_SECRET_KEY"],
-    return []
+        "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 _last_message_text(state: dict[str, Any]) -> str:
+def _clean_base_url(value: Any) -> str:
-    messages = state.get("messages") or []
+    text = str(value or "").strip().rstrip("/")
-    if not messages:
+    return text or "https://api.stripe.com"
-        return json.dumps(state, default=str)
 
-    content = getattr(messages[-1], "content", None)
+
-    if isinstance(content, str):
+def _validate_refund_request(
-        return content
+    *,
-    if isinstance(content, list):
+    stripe_payment_intent_id: str,
-        parts: list[str] = []
+    amount_cents: int,
-        for item in content:
+    currency: str,
-            if isinstance(item, dict):
+    allowed_currencies: list[str],
-                text = item.get("text") or item.get("content")
+    max_refund_cents: int,
-                if text:
+) -> dict[str, Any]:
-                    parts.append(str(text))
+    if not re.fullmatch(r"pi_[A-Za-z0-9_]+", stripe_payment_intent_id or ""):
-            elif item:
+        return {"status": "declined", "decline_code": "invalid_payment_intent", "executed": False}
-                parts.append(str(item))
+    currency_clean = (currency or "").lower().strip()
-        return "\n".join(parts) if parts else json.dumps(content, default=str)
+    if currency_clean not in {c.lower() for c in allowed_currencies}:
-    return str(content or messages[-1])
+        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())