diff --git a/agent.py b/agent.py index 1eb2d96..690c35a 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,734 @@ -"""receives-support-email-checks-1-919313 agent. +"""Support email refund workflow agent. -Starter stack: - - DeepAgents for tool-calling orchestration - - Caller-provided LLM credentials via ctx.llm - - A tiny model-call middleware hook you can replace with tracing, - routing, rate limits, or policy checks +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 -from pathlib import Path -from typing import Any +import re +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, - {{ auth_type }}, + AgentPlatformResources, + 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 = """\ -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 }}]): +class ReceivesSupportEmailChecks1919313(A2AAgent[ReceivesSupportEmailChecks1919313Config, NoAuth]): 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 - # ctx.llm. The platform may proxy that credential through LiteLLM, but agent - # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY - # directly. - llm_provisioning = LLMProvisioning.PLATFORM - pricing = Pricing( - price_per_call_usd=0.0, - caller_pays_llm=True, - notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", + 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=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") - async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str: - creds = ctx.llm - await ctx.emit_progress(f"llm: {creds.model} via {creds.source}") - if not creds.api_key: - return ( - "LLM key required. Add an LLM credential in Settings > LLM " - "credentials before running this agent; for local --invoke " - "runs set AGENT_LLM_KEY." - ) - 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( + @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[{{ auth_type }}], - creds: LLMCreds, - ) -> Any: - # Lazy imports keep `a2a card` usable before local dependencies are - # installed. `a2a deploy` installs requirements.txt during the build. - from a2a_pack.deepagents import create_a2a_deep_agent - from langchain.agents.middleware import wrap_model_call - from langchain_core.tools import tool - - @tool - def text_stats(text: str) -> str: - """Return exact word, character, and line counts for text.""" - words = [part for part in text.split() if part.strip()] - return json.dumps( - { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, - } - ) - - @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: 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, - creds=creds, - backend=backend, - skills=skill_sources or None, - tools=[text_stats], - middleware=[log_model_call], - system_prompt=SYSTEM_PROMPT, + 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 -def _runtime_skills_root(ctx: RunContext[Any]) -> str: - workspace = getattr(ctx, "_workspace", None) - prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) - if not prefixes: - outputs_prefix = getattr(workspace, "outputs_prefix", None) - prefixes = (outputs_prefix or "outputs/",) - prefix = str(prefixes[0]).strip("/") - return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}" + 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 _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: - """Copy packaged DeepAgents skills into the invocation workspace. - - DeepAgents loads skills from its backend, while source-controlled - ``skills/`` folders live in the image. This bridge lets generated agents - ship reusable SKILL.md bundles without giving up durable A2A workspace - files. - """ - root = Path(__file__).parent / "skills" - if not root.exists(): - return [] - runtime_skills_root = _runtime_skills_root(ctx) - uploads: list[tuple[str, bytes]] = [] - for path in root.rglob("*"): - if path.is_file(): - rel = path.relative_to(root).as_posix() - uploads.append((runtime_skills_root + rel, path.read_bytes())) - if uploads: - backend.upload_files(uploads) - return [runtime_skills_root] - return [] +def _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 _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +def _clean_base_url(value: Any) -> str: + text = str(value or "").strip().rstrip("/") + return text or "https://api.stripe.com" - content = getattr(messages[-1], "content", None) - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for item in content: - if isinstance(item, dict): - text = item.get("text") or item.get("content") - if text: - parts.append(str(text)) - elif item: - parts.append(str(item)) - return "\n".join(parts) if parts else json.dumps(content, default=str) - return str(content or messages[-1]) + +def _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())