Zapier AI Actions & Zapier Agents Cost Control: Task Billing, Retry Storms, Quota Exhaustion, and Inter-Zap Loops

Zapier built its business on the premise that automations are simple: one trigger, a handful of steps, a predictable number of tasks executed per month. Its pricing reflects that model — plans are tiered by task count, from 750 tasks per month on Starter to millions on Enterprise. The math was easy when a Zap ran a fixed number of steps per trigger. The math breaks when AI enters the workflow.

Zapier AI Actions let you put an LLM reasoning step inside any Zap — the AI action receives context from previous steps and can call downstream Zapier actions based on its output. Zapier Agents (launched 2024, now deeply integrated into the platform) go further: you configure an AI agent that autonomously decides which Zapier-connected apps to call, in what sequence, and how many times. Each of those autonomous tool calls registers as a Zap task. A Zapier Agent handling a customer support request might run 12 actions per ticket. At 800 tickets per month on a 1,000-task plan, you're already four months over quota before the month ends.

Four failure modes make this worse than a simple count underestimate:

• Per-task billing accumulation — every Zapier Agent action and every AI Actions step in a multi-step Zap counts as a separate billed task; complex AI workflows consume 5–20 tasks per trigger, not the 1–3 tasks that simple automations use.
• Retry storm amplification — Zapier automatically retries failed Zaps; if an AI step hits a rate limit or timeout, Zapier re-runs the entire Zap, re-billing all completed steps as new tasks while the failing AI step hits the rate limit again.
• Connection quota exhaustion — Zapier Agents track quota consumption differently from classic Zaps; agents that handle bursts of inbound requests (form submissions, webhook events, support ticket floods) can exhaust monthly task quotas in a single afternoon.
• Inter-Zap loop patterns — AI-driven Zaps that write data to apps can trigger other Zaps watching the same apps; circular trigger chains are invisible from the Zapier editor because each Zap only shows its own trigger, not the chain it's part of.

Failure Mode 1 — Per-Task Billing Accumulation

Zapier's billing unit is the task — one action step completed successfully in one Zap run. A trigger is not a task. A filter step is not a task. But every app action (send email, create record, update sheet, call an API, run an AI model) is a task. In a classic three-step Zap (trigger → filter → create Notion page), only the Notion action counts. One trigger = one task.

Zapier AI Actions breaks this model in two ways. First, AI Actions itself counts as one task per invocation — you're paying for the LLM reasoning step as a Zapier task regardless of what it does. Second, any downstream Zapier actions that the AI decides to call based on its output are additional tasks. A Zap that uses AI to classify an inbound email and then routes it to one of five downstream actions based on the classification always bills at least two tasks per run: one for the AI classification step, one for the routing action. If the AI decides to run multiple downstream actions, each one is a separate task.

Zapier Agents compound this further because the agent loop runs entirely inside Zapier's infrastructure. Each action the agent decides to execute — search a spreadsheet, look up a CRM record, send a Slack message, run a web search — is a separate Zapier task. An agent configured to handle a customer support request typically needs to: look up the customer record (1 task), search the knowledge base (1 task), check order history (1 task), optionally check a shipping API (1 task), draft and send a reply (1 task), and log the resolution (1 task). That's 6 tasks per ticket at minimum. Agents handling ambiguous queries often run 10–15 actions per session as they explore multiple data sources.

The guard pattern for billing accumulation is a per-session action counter that the agent checks before taking each action. Zapier Agents support custom instructions that include context about current session state, and you can use Zapier's Code step or a webhook to an external counter service. The simpler approach for teams with a backend is to count tasks via Zapier's webhook step and reject sessions that exceed the per-session ceiling before they exhaust the monthly quota:

from flask import Flask, request, jsonify
import sqlite3
import time
import threading

app = Flask(__name__)
db_lock = threading.Lock()

DB_PATH = "zapier_agent_budget.db"

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS agent_sessions (
                session_id TEXT PRIMARY KEY,
                action_count INTEGER DEFAULT 0,
                started_at REAL,
                last_action_at REAL,
                status TEXT DEFAULT 'active'
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS monthly_usage (
                month TEXT PRIMARY KEY,
                total_tasks INTEGER DEFAULT 0,
                task_ceiling INTEGER DEFAULT 0
            )
        """)

class ZapierAgentBudget:
    """
    Tracks per-session action counts and monthly task totals for Zapier Agents.
    Exposed as Flask webhook endpoints for Zapier Code/Webhook steps to call
    at the start of each agent action.
    """

    MAX_ACTIONS_PER_SESSION = 12
    MONTHLY_TASK_CEILING = 1800  # reserve 200 tasks buffer on a 2,000-task plan

    @staticmethod
    def check_and_increment(session_id: str) -> dict:
        month = time.strftime("%Y-%m")
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                # Upsert session
                conn.execute("""
                    INSERT INTO agent_sessions (session_id, action_count, started_at, last_action_at)
                    VALUES (?, 1, ?, ?)
                    ON CONFLICT(session_id) DO UPDATE SET
                        action_count = action_count + 1,
                        last_action_at = excluded.last_action_at
                """, (session_id, time.time(), time.time()))

                row = conn.execute(
                    "SELECT action_count, status FROM agent_sessions WHERE session_id = ?",
                    (session_id,)
                ).fetchone()
                action_count, status = row

                # Upsert monthly total
                conn.execute("""
                    INSERT INTO monthly_usage (month, total_tasks, task_ceiling)
                    VALUES (?, 1, ?)
                    ON CONFLICT(month) DO UPDATE SET total_tasks = total_tasks + 1
                """, (month, ZapierAgentBudget.MONTHLY_TASK_CEILING))

                monthly_row = conn.execute(
                    "SELECT total_tasks, task_ceiling FROM monthly_usage WHERE month = ?",
                    (month,)
                ).fetchone()
                monthly_tasks, monthly_ceiling = monthly_row

                if status == 'blocked':
                    return {
                        "allow": False,
                        "reason": "session_blocked",
                        "message": f"Session {session_id} was previously blocked.",
                        "action_count": action_count,
                    }

                if action_count > ZapierAgentBudget.MAX_ACTIONS_PER_SESSION:
                    conn.execute(
                        "UPDATE agent_sessions SET status = 'blocked' WHERE session_id = ?",
                        (session_id,)
                    )
                    return {
                        "allow": False,
                        "reason": "session_action_ceiling",
                        "message": (
                            f"Session {session_id} reached {action_count} actions "
                            f"(ceiling: {ZapierAgentBudget.MAX_ACTIONS_PER_SESSION}). "
                            "The agent is likely looping on an ambiguous query. "
                            "Ending session to prevent task quota burn."
                        ),
                        "action_count": action_count,
                        "monthly_tasks_used": monthly_tasks,
                    }

                if monthly_tasks > monthly_ceiling:
                    return {
                        "allow": False,
                        "reason": "monthly_quota_ceiling",
                        "message": (
                            f"Monthly task usage {monthly_tasks} exceeds soft ceiling "
                            f"{monthly_ceiling}. Blocking new agent actions to preserve "
                            "task reserve for critical Zaps."
                        ),
                        "action_count": action_count,
                        "monthly_tasks_used": monthly_tasks,
                    }

                return {
                    "allow": True,
                    "action_count": action_count,
                    "monthly_tasks_used": monthly_tasks,
                    "monthly_tasks_remaining": monthly_ceiling - monthly_tasks,
                }


@app.route("/agent/check", methods=["POST"])
def check_action():
    data = request.get_json(force=True)
    session_id = data.get("session_id", "")
    if not session_id:
        return jsonify({"allow": False, "reason": "missing_session_id"}), 400
    result = ZapierAgentBudget.check_and_increment(session_id)
    status_code = 200 if result["allow"] else 429
    return jsonify(result), status_code


if __name__ == "__main__":
    init_db()
    app.run(port=8080)

Wire this into your Zapier Agent as the first step in every action path. In the Zapier Agent's "Before action" instructions, add: "Before taking any action, call the webhook at [your endpoint]/agent/check with the current session_id. If the response contains "allow": false, stop the session and summarize what you've found so far." Zapier Agents follow custom instructions faithfully when they're framed as pre-action checks rather than post-action filters.

Failure Mode 2 — Retry Storm Amplification

Zapier automatically retries failed Zap runs. When any step in a Zap fails with a recoverable error (API timeout, 429 rate limit, 503 service unavailable), Zapier queues a retry. Classic Zaps retry the failed step, not the entire Zap. But multi-step Zaps with AI Actions have a subtler problem: if an early step completes successfully but a later AI step fails, Zapier may re-run steps that already completed — sending duplicate emails, creating duplicate records, triggering duplicate downstream Zaps — while billing additional tasks for the re-run steps.

The rate-limit cascade is the most common form. A Zapier Zap calls an OpenAI model (via the AI Actions step or a direct HTTP action). The OpenAI API returns a 429 during a period of high demand. Zapier retries the Zap after a backoff delay. The OpenAI API is still under load and returns another 429. Zapier retries again. After three retries, Zapier marks the Zap as errored and sends you a notification email. In the meantime, you've consumed 4× the expected tasks (original run + 3 retries), generated 4× the OpenAI API calls (contributing to the rate limit that caused the problem), and potentially executed any steps before the AI action up to 4 times — including actions that are not idempotent.

The amplification is multiplicative when multiple Zaps share the same rate-limited API key. If ten Zaps all hit OpenAI at the same time and all receive 429 responses, Zapier queues ten retry sets. If each Zap has 3 retries, you generate 40 Zap runs, 40 OpenAI calls (all still rate-limited), and 40 sets of billed tasks. The retry mechanism designed to recover from transient failures becomes the primary driver of the spike that caused the failure.

The guard pattern is two-pronged. First, add idempotency checks to any step that runs before your AI action — if the step already ran for this trigger event, skip it instead of re-executing. Second, use Zapier's built-in filter to check a circuit breaker state before the AI step runs, preventing the AI action from being attempted during a known rate-limit event:

import time
import sqlite3
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)
db_lock = threading.Lock()
DB_PATH = "zapier_retry_guard.db"

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS rate_limit_events (
                provider TEXT PRIMARY KEY,
                consecutive_failures INTEGER DEFAULT 0,
                first_failure_at REAL,
                last_failure_at REAL,
                breaker_open INTEGER DEFAULT 0,
                recovery_after REAL DEFAULT 0
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS zap_idempotency (
                event_key TEXT PRIMARY KEY,
                status TEXT,
                completed_at REAL
            )
        """)

BREAKER_THRESHOLD = 3      # open after 3 consecutive 429s
RECOVERY_WINDOW_SEC = 120  # stay open for 2 minutes, then allow probe

class ZapierRetryGuard:

    @staticmethod
    def check_idempotency(event_key: str) -> dict:
        """Check if this trigger event has already been processed."""
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                row = conn.execute(
                    "SELECT status, completed_at FROM zap_idempotency WHERE event_key = ?",
                    (event_key,)
                ).fetchone()
                if row and row[0] == "completed":
                    return {
                        "already_processed": True,
                        "completed_at": row[1],
                        "message": (
                            f"Event {event_key!r} already completed at {row[1]}. "
                            "This is a Zapier retry — skip downstream steps to prevent duplicates."
                        ),
                    }
                # Mark as in-progress
                conn.execute("""
                    INSERT INTO zap_idempotency (event_key, status, completed_at)
                    VALUES (?, 'in_progress', ?)
                    ON CONFLICT(event_key) DO NOTHING
                """, (event_key, time.time()))
                return {"already_processed": False}

    @staticmethod
    def mark_completed(event_key: str):
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                conn.execute(
                    "UPDATE zap_idempotency SET status = 'completed', completed_at = ? WHERE event_key = ?",
                    (time.time(), event_key)
                )

    @staticmethod
    def record_rate_limit(provider: str) -> dict:
        """Record a 429 from the given provider and report breaker state."""
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                now = time.time()
                conn.execute("""
                    INSERT INTO rate_limit_events
                        (provider, consecutive_failures, first_failure_at, last_failure_at)
                    VALUES (?, 1, ?, ?)
                    ON CONFLICT(provider) DO UPDATE SET
                        consecutive_failures = consecutive_failures + 1,
                        last_failure_at = excluded.last_failure_at
                """, (provider, now, now))

                row = conn.execute(
                    "SELECT consecutive_failures FROM rate_limit_events WHERE provider = ?",
                    (provider,)
                ).fetchone()
                consecutive = row[0]

                if consecutive >= BREAKER_THRESHOLD:
                    recovery_after = now + RECOVERY_WINDOW_SEC
                    conn.execute("""
                        UPDATE rate_limit_events
                        SET breaker_open = 1, recovery_after = ?
                        WHERE provider = ?
                    """, (recovery_after, provider))
                    return {
                        "breaker_open": True,
                        "consecutive_failures": consecutive,
                        "recovery_after": recovery_after,
                        "message": (
                            f"Circuit breaker OPEN for {provider!r} after "
                            f"{consecutive} consecutive 429s. "
                            f"Blocking all AI actions using {provider!r} for "
                            f"{RECOVERY_WINDOW_SEC}s to prevent retry storm amplification."
                        ),
                    }
                return {"breaker_open": False, "consecutive_failures": consecutive}

    @staticmethod
    def check_breaker(provider: str) -> dict:
        """Check if the circuit breaker is open before attempting an AI action."""
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                row = conn.execute(
                    "SELECT breaker_open, recovery_after, consecutive_failures "
                    "FROM rate_limit_events WHERE provider = ?",
                    (provider,)
                ).fetchone()
                if not row:
                    return {"allow": True}
                breaker_open, recovery_after, consecutive = row
                if breaker_open and time.time() < recovery_after:
                    return {
                        "allow": False,
                        "reason": "breaker_open",
                        "provider": provider,
                        "recovery_in_seconds": int(recovery_after - time.time()),
                        "consecutive_failures": consecutive,
                        "message": (
                            f"Circuit breaker OPEN for {provider!r}. "
                            f"Do not retry — wait {int(recovery_after - time.time())}s "
                            "before next attempt. Zapier retry will be blocked by this filter."
                        ),
                    }
                # Probe: reset if recovery window has passed
                if breaker_open and time.time() >= recovery_after:
                    conn.execute(
                        "UPDATE rate_limit_events SET breaker_open = 0, "
                        "consecutive_failures = 0 WHERE provider = ?",
                        (provider,)
                    )
                return {"allow": True}


@app.route("/retry/check-idempotency", methods=["POST"])
def check_idempotency():
    data = request.get_json(force=True)
    return jsonify(ZapierRetryGuard.check_idempotency(data.get("event_key", "")))

@app.route("/retry/check-breaker", methods=["POST"])
def check_breaker():
    data = request.get_json(force=True)
    return jsonify(ZapierRetryGuard.check_breaker(data.get("provider", "openai")))

@app.route("/retry/record-rate-limit", methods=["POST"])
def record_rate_limit():
    data = request.get_json(force=True)
    return jsonify(ZapierRetryGuard.record_rate_limit(data.get("provider", "openai")))

@app.route("/retry/mark-completed", methods=["POST"])
def mark_completed():
    data = request.get_json(force=True)
    ZapierRetryGuard.mark_completed(data.get("event_key", ""))
    return jsonify({"ok": True})

if __name__ == "__main__":
    init_db()
    app.run(port=8081)

Wire /retry/check-idempotency as the first step in any Zap that has side effects before an AI action. Use Zapier's Filter step on the response: if already_processed is true, stop the Zap. Wire /retry/check-breaker immediately before the AI action step. If the breaker is open, stop the Zap — Zapier will not retry a Zap that was explicitly stopped by a filter, only one that errored. When the AI action succeeds, call /retry/mark-completed with the event key. When it returns a 429, call /retry/record-rate-limit before allowing Zapier to attempt a retry.

Failure Mode 3 — Connection Quota Exhaustion During Bursts

Zapier Agents are asynchronous. When a user submits a form, sends a message to your agent's webhook endpoint, or triggers an agent via an inbound email, Zapier queues the request and processes it when capacity is available. The quota accounting happens at execution time, not at queue time. If 200 support requests arrive in a two-hour window and each triggers a Zapier Agent that runs 10 actions, you consume 2,000 tasks in two hours — potentially exhausting a month's task allocation before 10% of the month has passed.

The problem compounds because Zapier's quota enforcement is not real-time per-task. It updates on a billing-cycle basis with some lag. Teams that check their task count at the end of a burst discover they've been 300% over quota for the past six hours, with Zapier continuing to process (and bill at overage rates) because the system didn't enforce a hard stop mid-run. Zapier's overage pricing is typically 1.25–2× the per-task rate on the base plan. A burst that costs 2,000 tasks over a 1,800-task monthly ceiling at $0.01/task overage = $20 in overages from a single afternoon event.

The burst pattern is different from the steady-state accumulation in Failure Mode 1. In steady state, you gradually approach the ceiling. In a burst, you blow through the ceiling in a short window before any monitoring alert fires. The guard pattern for bursts is a sliding window rate limiter that caps how many agent sessions can start in any rolling hour:

import time
import sqlite3
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)
db_lock = threading.Lock()
DB_PATH = "zapier_burst_guard.db"

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS session_starts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                started_at REAL,
                session_id TEXT,
                source TEXT
            )
        """)

class ZapierBurstGuard:
    """
    Sliding window rate limiter for Zapier Agent session starts.
    Prevents burst quota exhaustion by capping sessions per rolling hour.
    """

    # Conservative defaults: 20 sessions/hour × 10 actions each = 200 tasks/hour
    # At 1,800-task/month ceiling: allows ~9 burst-hours before hitting ceiling
    MAX_SESSIONS_PER_HOUR = 20
    MAX_SESSIONS_PER_DAY = 150  # prevents sustained overconsumption
    WINDOW_HOUR = 3600
    WINDOW_DAY = 86400

    @staticmethod
    def check_and_record(session_id: str, source: str = "") -> dict:
        now = time.time()
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                # Count sessions in rolling hour
                hour_count = conn.execute(
                    "SELECT COUNT(*) FROM session_starts WHERE started_at > ?",
                    (now - ZapierBurstGuard.WINDOW_HOUR,)
                ).fetchone()[0]

                # Count sessions in rolling day
                day_count = conn.execute(
                    "SELECT COUNT(*) FROM session_starts WHERE started_at > ?",
                    (now - ZapierBurstGuard.WINDOW_DAY,)
                ).fetchone()[0]

                if hour_count >= ZapierBurstGuard.MAX_SESSIONS_PER_HOUR:
                    # Find when the oldest session in the window will age out
                    oldest_in_window = conn.execute(
                        "SELECT MIN(started_at) FROM session_starts WHERE started_at > ?",
                        (now - ZapierBurstGuard.WINDOW_HOUR,)
                    ).fetchone()[0]
                    retry_after = int(oldest_in_window + ZapierBurstGuard.WINDOW_HOUR - now)
                    return {
                        "allow": False,
                        "reason": "hourly_burst_ceiling",
                        "sessions_in_last_hour": hour_count,
                        "ceiling": ZapierBurstGuard.MAX_SESSIONS_PER_HOUR,
                        "retry_after_seconds": retry_after,
                        "message": (
                            f"Hourly agent session ceiling reached: {hour_count} sessions "
                            f"in the last hour (ceiling: {ZapierBurstGuard.MAX_SESSIONS_PER_HOUR}). "
                            f"This burst would exhaust monthly task quota. "
                            f"New sessions blocked for {retry_after}s."
                        ),
                    }

                if day_count >= ZapierBurstGuard.MAX_SESSIONS_PER_DAY:
                    return {
                        "allow": False,
                        "reason": "daily_ceiling",
                        "sessions_today": day_count,
                        "ceiling": ZapierBurstGuard.MAX_SESSIONS_PER_DAY,
                        "message": (
                            f"Daily agent session ceiling reached: {day_count} sessions "
                            f"(ceiling: {ZapierBurstGuard.MAX_SESSIONS_PER_DAY}). "
                            "Blocking further sessions to preserve monthly task quota."
                        ),
                    }

                # Allow and record
                conn.execute(
                    "INSERT INTO session_starts (started_at, session_id, source) VALUES (?, ?, ?)",
                    (now, session_id, source)
                )
                # Prune entries older than 2 days to keep table compact
                conn.execute(
                    "DELETE FROM session_starts WHERE started_at < ?",
                    (now - ZapierBurstGuard.WINDOW_DAY * 2,)
                )
                return {
                    "allow": True,
                    "sessions_in_last_hour": hour_count + 1,
                    "sessions_today": day_count + 1,
                }


@app.route("/burst/check", methods=["POST"])
def burst_check():
    data = request.get_json(force=True)
    result = ZapierBurstGuard.check_and_record(
        session_id=data.get("session_id", ""),
        source=data.get("source", ""),
    )
    return jsonify(result), 200 if result["allow"] else 429


if __name__ == "__main__":
    init_db()
    app.run(port=8082)

Add a webhook step at the very beginning of your Zapier Agent's trigger Zap — before the agent session starts. Call /burst/check. Use a Filter step to stop the Zap if allow is false. This queues demand gracefully at the rate-limiting layer rather than letting Zapier process every inbound trigger and exhaust quota. Adjust MAX_SESSIONS_PER_HOUR based on your plan's task count divided by your observed average actions per session — leave a 20% buffer for retries and unexpected multi-step sessions.

Failure Mode 4 — Inter-Zap Loop Patterns

Zapier's architecture allows Zaps to trigger each other indirectly through shared app data. A Zap watches a Google Sheet for new rows → runs an AI analysis → creates a Notion database item. A second Zap watches the Notion database for new items → runs a different AI classification → writes a summary row back to Google Sheets. When the summary row matches the "new row" trigger in the first Zap, the loop closes. Both Zaps fire indefinitely, each run billing tasks on both ends and consuming LLM tokens for the AI steps in the middle.

This is invisible from the Zapier editor. The first Zap only shows its own trigger (Google Sheets: new row) and has no visibility into what other Zaps watch Notion. The second Zap only shows its own trigger (Notion: new item) and has no visibility into what the Google Sheet row will trigger. Neither Zapier's Zap editor nor its history view shows inter-Zap dependency graphs. Teams discover the loop when they notice the history view showing the same two Zaps firing alternately hundreds of times and the task count spinning upward.

AI Zaps are more prone to creating these loops than classic Zaps because AI-driven steps are more likely to write rich, structured output to a data store (a Notion database, an Airtable base, a Google Sheet) — the exact surfaces that other monitoring Zaps watch. A classic Zap that writes a row to a sheet typically writes a predictable, stable row. An AI Zap that writes a synthesized summary row may write data that looks like new user input to a downstream Zap, especially if the trigger condition is a field-based filter rather than a presence-of-new-row check.

The guard pattern for inter-Zap loops is a provenance tag written to the data record on first creation. Any downstream Zap can check for the tag and stop if it was written by another Zap rather than a human user. The tag should be written to a field that is stable (not overwritten by the AI step) and machine-readable:

import time
import sqlite3
import hashlib
import threading
from flask import Flask, request, jsonify

app = Flask(__name__)
db_lock = threading.Lock()
DB_PATH = "zapier_loop_guard.db"

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS provenance_tags (
                record_key TEXT PRIMARY KEY,
                created_by TEXT,
                created_at REAL,
                source_zap TEXT,
                hop_count INTEGER DEFAULT 0
            )
        """)

MAX_HOP_COUNT = 2  # allow at most 2 Zap-to-Zap hops before blocking

class ZapierLoopGuard:
    """
    Tracks provenance of records written by Zaps to detect inter-Zap loops.

    Usage pattern:
    1. Before writing a record, call /loop/tag with source_zap and a stable record_key.
    2. When a downstream Zap picks up a record, call /loop/check with the record_key.
       If hop_count exceeds MAX_HOP_COUNT, the downstream Zap should stop.
    3. The record_key should be a stable identifier for the data record —
       for Google Sheets, use row ID; for Notion, use page ID; for Airtable, use record ID.
    """

    @staticmethod
    def tag_record(record_key: str, source_zap: str, parent_key: str = "") -> dict:
        """Mark a record as Zap-created. Call before writing to the downstream app."""
        hop_count = 0
        if parent_key:
            with db_lock:
                with sqlite3.connect(DB_PATH) as conn:
                    parent_row = conn.execute(
                        "SELECT hop_count FROM provenance_tags WHERE record_key = ?",
                        (parent_key,)
                    ).fetchone()
                    if parent_row:
                        hop_count = parent_row[0] + 1

        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                conn.execute("""
                    INSERT INTO provenance_tags
                        (record_key, created_by, created_at, source_zap, hop_count)
                    VALUES (?, 'zapier_automation', ?, ?, ?)
                    ON CONFLICT(record_key) DO UPDATE SET
                        created_by = excluded.created_by,
                        created_at = excluded.created_at,
                        source_zap = excluded.source_zap,
                        hop_count = excluded.hop_count
                """, (record_key, time.time(), source_zap, hop_count))

        return {
            "tagged": True,
            "record_key": record_key,
            "hop_count": hop_count,
            "source_zap": source_zap,
        }

    @staticmethod
    def check_record(record_key: str) -> dict:
        """
        Check whether a record was created by a Zap (not a human).
        Returns allow=False if the record is Zap-created and hop_count >= MAX_HOP_COUNT.
        """
        with db_lock:
            with sqlite3.connect(DB_PATH) as conn:
                row = conn.execute(
                    "SELECT created_by, source_zap, hop_count, created_at "
                    "FROM provenance_tags WHERE record_key = ?",
                    (record_key,)
                ).fetchone()

        if not row:
            # No provenance tag — record is from a human or an untagged Zap
            return {"allow": True, "provenance": "unknown"}

        created_by, source_zap, hop_count, created_at = row

        if hop_count >= MAX_HOP_COUNT:
            return {
                "allow": False,
                "reason": "inter_zap_loop_detected",
                "provenance": created_by,
                "source_zap": source_zap,
                "hop_count": hop_count,
                "max_hop_count": MAX_HOP_COUNT,
                "message": (
                    f"Record {record_key!r} was created by Zap {source_zap!r} and has "
                    f"already propagated through {hop_count} Zap hops "
                    f"(ceiling: {MAX_HOP_COUNT}). "
                    "Stopping this Zap to break the inter-Zap loop. "
                    "Check your trigger conditions to prevent circular Zap chains."
                ),
            }

        return {
            "allow": True,
            "provenance": created_by,
            "source_zap": source_zap,
            "hop_count": hop_count,
            "warning": (
                f"Record is Zap-created (hop {hop_count}). "
                f"Allow for now but watch for loop patterns in Zap history."
            ) if hop_count > 0 else None,
        }


@app.route("/loop/tag", methods=["POST"])
def tag_record():
    data = request.get_json(force=True)
    return jsonify(ZapierLoopGuard.tag_record(
        record_key=data.get("record_key", ""),
        source_zap=data.get("source_zap", ""),
        parent_key=data.get("parent_key", ""),
    ))

@app.route("/loop/check", methods=["POST"])
def check_record():
    data = request.get_json(force=True)
    result = ZapierLoopGuard.check_record(data.get("record_key", ""))
    return jsonify(result), 200 if result["allow"] else 429

if __name__ == "__main__":
    init_db()
    app.run(port=8083)

Add a /loop/tag webhook call to every Zap that writes records to a shared data store — immediately before the write step. Pass the source_zap as a stable Zap name or ID so you can identify which Zap created the loop in the history. Add a /loop/check webhook call as the first step in every Zap that might be triggered by another Zap's output, followed by a Filter step that stops the Zap if allow is false. For Zap chains that legitimately need two hops (Zap A writes, Zap B processes and writes, Zap C consumes), set MAX_HOP_COUNT = 3 — the third hop check will block any inadvertent fourth hop from closing a loop.

State Table: Four Failure Modes at a Glance

Failure Mode	Guard Class	Ceiling	What to Watch
Per-task billing accumulation Each agent action = 1 Zap task	ZapierAgentBudget	12 actions/session, 1,800 tasks/month	Zapier task usage dashboard; average actions per agent session
Retry storm amplification 429 → retry → 429 cycle	ZapierRetryGuard	3 consecutive 429s → breaker open for 120s	Zap error history for consecutive failures on same step
Burst quota exhaustion Event flood exhausts monthly allocation	ZapierBurstGuard	20 sessions/hour, 150 sessions/day	Hourly Zap run count; task velocity vs. monthly ceiling
Inter-Zap loop Circular trigger chain between Zaps	ZapierLoopGuard	2 Zap hops max	Zapier history for alternating Zap pairs; task count rate-of-change

Checklist: Before Deploying Zapier AI Workflows

• Count actual actions per agent session. Run 20 test sessions with production-representative inputs. Record the action count for each. Use the 90th percentile as your per-session ceiling, not the average.
• Calculate monthly task headroom before launch. (Expected sessions/month) × (90th-percentile actions/session) should be ≤ 80% of your plan's task limit. Leave 20% for retries and burst events.
• Add idempotency to every step before an AI action. If your trigger can fire multiple times for the same event (webhooks with at-least-once delivery, email triggers, form submissions), the steps before the AI action must be safe to re-run. Store event keys in an external DB or Zapier Storage.
• Map inter-Zap dependencies before writing data. Identify every Zap that writes to a shared data store and every Zap that watches the same store. Draw the dependency graph. Any closed path is a potential loop.
• Add provenance tags to every Zap-created record. Even if you don't think you have a loop today, the first time someone adds a new Zap that watches the same data source, you'll be glad the tags are there.
• Set Zapier's task usage alerts. Zapier supports email alerts at 80% and 100% of monthly task usage. Enable both. Add a third alert via a scheduled Zap that checks the Zapier API for current task usage and sends a Slack message if it exceeds 60% before the 15th of the month.
• Use Zapier's built-in Filter step to stop Zaps, not just Error step. Filters that stop a Zap don't consume tasks after the filter step. Error handling that lets the Zap continue to a final "log error" action still consumes one more task. Stop, don't error-then-log.