Conductor Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Orkes Conductor (formerly Netflix Conductor) is a workflow orchestration engine built around a poll-and-acknowledge task model: workers long-poll a queue, execute a unit of work, and post the result back as COMPLETED or FAILED. That model gives you durable, auditable workflows across distributed workers — but when the work being executed is a Stripe charge, the acknowledgment gap between "charge succeeded" and "task marked COMPLETED" introduces a double-billing window that Stripe restricted keys alone cannot close.

This post covers three Conductor-specific failure modes that each cause a second stripe.charges.create() call on an already-billed customer, and the two-layer governance pattern — content-hash idempotency keys plus per-task vault keys via a spend-cap proxy — that closes all three without restructuring your workflow topology.

Failure mode 1: Task retry policy re-invokes the worker from line 1 after stripe.charges.create() has already succeeded

When you define a Conductor task with retryCount > 0, Conductor will re-queue the task whenever the worker fails to post a COMPLETED result within responseTimeoutSeconds, or when the worker explicitly posts a FAILED status with a retriable error type. Either way, a new worker poll picks up the re-queued task and executes the worker function from the first line.

The failure window is the gap between the Stripe API call succeeding and the Conductor COMPLETED acknowledgment being posted. Consider this sequence: your billing worker calls stripe.charges.create() and receives ch_A in the response. Then, before the worker can call task_client.update_task_by_ref_name(status="COMPLETED", ...), one of these events occurs: the worker process is OOM-killed, the network between the worker and Conductor server drops, or the downstream database write your worker was performing raises an exception. In all three cases, Conductor never receives the COMPLETED acknowledgment. When responseTimeoutSeconds elapses, Conductor re-queues the task. Worker 2 polls, executes from line 1, calls stripe.charges.create() without an idempotency key — and ch_B is created. Customer billed twice.

# billing_worker.py — UNSAFE: no idempotency key, task exception causes re-queue
import stripe
import os
from conductor.client.worker.worker_task import worker_task

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # shared, unrestricted

@worker_task(task_definition_name="bill_customer")
def bill_customer(customer_id: str, amount_cents: int, billing_period: str) -> dict:
    # If this call succeeds but task update below fails → re-queue → second charge
    charge = stripe.charges.create(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=f"Subscription {billing_period}",
    )
    # OOM / network drop / DB exception here → Conductor re-queues → ch_B
    save_charge_to_db(charge.id, customer_id, billing_period)
    return {"charge_id": charge.id, "status": "charged"}

The fix is a content-hash idempotency key computed from inputs that are stable across all retries. Conductor passes the same workflow input to every retry of a task, so a key derived from customer_id + amount_cents + billing_period will be identical on Worker 2's call — and Stripe will return the original ch_A object rather than creating ch_B. Additionally, Stripe permanent errors (card declined, invalid customer) should be returned as structured output rather than raised as Python exceptions; raised exceptions cause Conductor to FAILED-retry the task, which would re-hit Stripe on a charge that was correctly rejected and should not be retried.

# billing_worker.py — SAFE: stable idempotency key + permanent-error handling
import hashlib
import stripe
import os
from conductor.client.worker.worker_task import worker_task

@worker_task(task_definition_name="bill_customer")
def bill_customer(
    customer_id: str,
    amount_cents: int,
    billing_period: str,
    vault_key: str,  # per-task scoped key from upstream issue_vault_keys task
) -> dict:
    idempotency_key = hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:conductor-billing".encode()
    ).hexdigest()[:32]

    stripe_client = stripe.Stripe(
        api_key=vault_key,
        base_url="https://proxy.keybrake.com/stripe/",
    )

    try:
        charge = stripe_client.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
            idempotency_key=idempotency_key,
        )
        save_charge_to_db(charge.id, customer_id, billing_period)
        return {"charge_id": charge.id, "status": "charged"}
    except stripe.error.CardError as e:
        # Permanent failure — return as structured output, NOT raise
        # Raising would cause Conductor to FAILED-retry → re-hits Stripe on declined card
        return {"status": "card_declined", "error_code": e.code, "charge_id": None}
    except stripe.error.InvalidRequestError as e:
        return {"status": "invalid_request", "error": str(e), "charge_id": None}

Failure mode 2: FORK_JOIN parallel tasks share one STRIPE_SECRET_KEY across all concurrent workers with no per-task spend cap

When your workflow fans out billing across N customers using a FORK_JOIN task, Conductor dispatches N parallel task instances simultaneously. All N workers that poll and execute those tasks share the same STRIPE_SECRET_KEY injected via environment variable — the same unrestricted production key, with no per-worker dollar ceiling, no per-customer scope, and no circuit breaker.

The exposure is: if a data error appears in the amount_cents field of your workflow input (a unit conversion bug, a decimal-to-integer overflow, or an upstream API returning a price in the wrong currency), all N parallel workers will fire charges simultaneously before any result returns. With 500 customers in a FORK_JOIN fan-out, 500 charges hit Stripe in the same second — all for the wrong amount, all using the same unrestricted key. No Stripe feature stops this; restricted API keys scope by endpoint and resource type, not by dollar total per key invocation.

# Conductor workflow definition (JSON) — UNSAFE FORK_JOIN
{
  "name": "monthly_billing_workflow",
  "tasks": [
    {
      "name": "prepare_billing_batch",
      "taskReferenceName": "prepare_billing_batch_ref",
      "type": "SIMPLE"
    },
    {
      "name": "fork_billing",
      "taskReferenceName": "fork_billing_ref",
      "type": "FORK_JOIN",
      "forkTasks": [
        [{ "name": "bill_customer", "taskReferenceName": "bill_ref_0", "type": "SIMPLE" }],
        [{ "name": "bill_customer", "taskReferenceName": "bill_ref_1", "type": "SIMPLE" }]
        // ... N branches, all workers share STRIPE_SECRET_KEY from environment
      ]
    },
    { "name": "join_billing", "taskReferenceName": "join_billing_ref", "type": "JOIN",
      "joinOn": ["bill_ref_0", "bill_ref_1"] }
  ]
}

The fix is a dedicated issue_vault_keys task that runs before the FORK_JOIN block. This task calls the Keybrake API once, issuing N vault keys — one per customer — each scoped to POST /v1/charges with a per-key USD cap of customer_amount_cents / 100 * 1.10. The vault keys are written into the workflow's output data and threaded into each forked branch as input. Each parallel bill_customer task uses its own capped vault key rather than the shared STRIPE_SECRET_KEY. A data error in amount_cents will now be blocked at the proxy spend cap after the first customer is overcharged — not after all 500 have been.

# issue_vault_keys_worker.py — pre-fan-out vault key issuance
import requests
import os
from conductor.client.worker.worker_task import worker_task

KEYBRAKE_API_KEY = os.environ["KEYBRAKE_API_KEY"]

@worker_task(task_definition_name="issue_vault_keys")
def issue_vault_keys(customers: list[dict]) -> dict:
    vault_keys = {}
    for customer in customers:
        resp = requests.post(
            "https://proxy.keybrake.com/keys",
            headers={"Authorization": f"Bearer {KEYBRAKE_API_KEY}"},
            json={
                "vendor": "stripe",
                "allowed_endpoints": ["POST /v1/charges"],
                "daily_usd_cap": round(customer["amount_cents"] / 100 * 1.10, 2),
                "expires_in_seconds": 3600,
                "label": f"billing-{customer['customer_id']}-{customer['billing_period']}",
            },
            timeout=10,
        )
        resp.raise_for_status()
        vault_keys[customer["customer_id"]] = resp.json()["vault_key"]
    # Conductor stores this as task output; each forked branch reads its own key
    return {"vault_keys": vault_keys}

Failure mode 3: Workflow restart re-executes COMPLETED billing tasks that already charged customers

Conductor exposes two administrative APIs for recovering failed workflows. POST /workflow/{workflowId}/retry re-runs from the first FAILED task — tasks that are already COMPLETED are skipped, their outputs replayed from the workflow's execution state. This is safe for billing tasks: the COMPLETED billing task is not re-executed.

POST /workflow/{workflowId}/restart is the dangerous one. It re-runs the entire workflow from task 1, discarding all previous task states. Every task — including the billing task that already reached COMPLETED and already created ch_A — runs again. Without idempotency keys, Worker 3 calls stripe.charges.create() and creates ch_B. The Conductor UI makes restart trivially easy, and it is the natural first instinct when a workflow ends up in a terminal FAILED state due to a downstream task failure after billing completed.

The idempotency key fix from Failure Mode 1 handles restart transparently: because the key is derived from stable inputs (customer_id + amount_cents + billing_period), Worker 3's call returns Stripe's cached ch_A response. No second charge fires. The vault key also provides a secondary signal: the proxy audit log will show a second request for the same customer arriving on the restarted run, making the restart visible in the Keybrake audit trail even though no money moved.

There is one remaining gap: the issue_vault_keys pre-fan-out task also re-runs on a workflow restart, and it will issue a fresh set of vault keys. These new keys are separate from the keys issued in the original run. The idempotency key on the Stripe call still protects against double-charging, but the proxy's spend-cap accounting resets. Add a pre-flight check in issue_vault_keys that looks up whether the billing period is already marked complete in your database before issuing new keys, and returns the existing keys (stored from the original run) if it is:

# issue_vault_keys_worker.py — restart-safe pre-flight check
@worker_task(task_definition_name="issue_vault_keys")
def issue_vault_keys(customers: list[dict], billing_period: str) -> dict:
    # Pre-flight: if period already billed, return existing vault keys from DB
    existing = db.query(
        "SELECT customer_id, vault_key FROM billing_vault_keys WHERE billing_period = ?",
        [billing_period]
    )
    if existing:
        return {"vault_keys": {row["customer_id"]: row["vault_key"] for row in existing}}

    vault_keys = {}
    for customer in customers:
        resp = requests.post(
            "https://proxy.keybrake.com/keys",
            headers={"Authorization": f"Bearer {KEYBRAKE_API_KEY}"},
            json={
                "vendor": "stripe",
                "allowed_endpoints": ["POST /v1/charges"],
                "daily_usd_cap": round(customer["amount_cents"] / 100 * 1.10, 2),
                "expires_in_seconds": 7200,
                "label": f"billing-{customer['customer_id']}-{billing_period}",
            },
            timeout=10,
        )
        resp.raise_for_status()
        vault_keys[customer["customer_id"]] = resp.json()["vault_key"]

    # Persist keys so workflow restart can recover them
    for customer_id, key in vault_keys.items():
        db.execute(
            "INSERT OR IGNORE INTO billing_vault_keys (billing_period, customer_id, vault_key) VALUES (?, ?, ?)",
            [billing_period, customer_id, key],
        )
    return {"vault_keys": vault_keys}

Approach comparison

Gap analysis

1. failureWorkflow compensating charges re-fire on retry

Conductor supports a failureWorkflow field on your workflow definition — a separate workflow that triggers automatically when the primary workflow enters a FAILED terminal state. Teams use this for compensation logic: if a multi-step process partially succeeded before failing, the failure workflow issues refunds or rollbacks. If the failure workflow itself fails and is retried (or restarted), any stripe.refunds.create() calls within it re-execute. Apply the same idempotency key pattern — SHA-256(charge_id + "refund" + billing_period)[:32] — to all refund calls in failure workflows, not just to charges in the primary workflow.

2. Conductor's built-in HTTP task type bypasses all worker governance

Conductor supports an HTTP built-in task type that makes HTTP calls directly from the Conductor server, without a polling worker. Teams sometimes use it to call the Stripe API directly in a workflow definition (e.g., {"uri": "https://api.stripe.com/v1/charges", "method": "POST"}). This bypasses your worker-level idempotency key logic entirely — the HTTP task has no idempotency key field in its task definition schema. If the HTTP task fails and Conductor retries it, a second POST fires. Switch any billing HTTP tasks to proper worker tasks where you control the idempotency key, or configure the HTTP task to call proxy.keybrake.com/stripe/v1/charges and set a stable Idempotency-Key header using Conductor's variable substitution (${workflow.workflowId}-${customer_id}-billing).

3. DYNAMIC_FORK vault key sizing when fan-out count is determined at runtime

DYNAMIC_FORK tasks determine the number of branches from upstream task output at runtime — you don't know N when you define the workflow. The issue_vault_keys pre-fan-out task needs the full customer list to issue one key per customer, which means it must receive the complete customers array as its input. If the upstream prepare_billing_batch task produces a paginated cursor rather than a full array, issue_vault_keys must exhaust pagination before issuing keys. Don't issue keys lazily inside the billing worker — that reintroduces the shared-key problem because each worker would need a spend cap covering the full batch, not just one customer.

4. Timed-out tasks re-queued by Conductor may overlap with the original worker

When a billing worker exceeds responseTimeoutSeconds, Conductor re-queues the task without guaranteed termination of the original worker. In some deployment environments (Kubernetes with slow pod termination, EC2 with graceful shutdown delays), the original worker may still be executing the Stripe call while the re-queued task has already been picked up by a second worker. Both workers call stripe.charges.create() concurrently. With a stable idempotency key, Stripe serializes the two concurrent requests and returns the same charge object to both — but verify that the idempotency key is computed before the Stripe call (not passed from a dynamic generator inside the call) so both workers compute the same key from the same stable inputs.

Enforcement tests

# test_conductor_billing.py
import hashlib
import pytest
from unittest.mock import patch, MagicMock
from billing_worker import bill_customer
from issue_vault_keys_worker import issue_vault_keys

def make_idempotency_key(customer_id, amount_cents, billing_period):
    return hashlib.sha256(
        f"{customer_id}:{amount_cents}:{billing_period}:conductor-billing".encode()
    ).hexdigest()[:32]

def test_idempotency_key_is_stable_across_retries():
    key1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
    key2 = make_idempotency_key("cus_ABC", 2999, "2026-07")
    assert key1 == key2, "Idempotency key must be identical across task retries"

def test_idempotency_key_differs_by_customer():
    key1 = make_idempotency_key("cus_ABC", 2999, "2026-07")
    key2 = make_idempotency_key("cus_DEF", 2999, "2026-07")
    assert key1 != key2, "Different customers must not share idempotency keys"

def test_card_declined_returns_struct_not_raises():
    with patch("billing_worker.stripe.Stripe") as mock_stripe:
        mock_charge = mock_stripe.return_value.charges.create
        mock_charge.side_effect = __import__("stripe").error.CardError(
            "Your card was declined.", None, "card_declined", http_status=402
        )
        result = bill_customer(
            customer_id="cus_ABC",
            amount_cents=2999,
            billing_period="2026-07",
            vault_key="vk_test_abc123",
        )
    assert result["status"] == "card_declined"
    assert result["charge_id"] is None
    # Should not raise — raising would cause Conductor to FAILED-retry a declined card

def test_vault_key_cap_set_per_customer_not_per_batch():
    customers = [
        {"customer_id": "cus_A", "amount_cents": 1000, "billing_period": "2026-07"},
        {"customer_id": "cus_B", "amount_cents": 5000, "billing_period": "2026-07"},
    ]
    with patch("issue_vault_keys_worker.requests.post") as mock_post:
        mock_post.return_value.json.return_value = {"vault_key": "vk_test_xxx"}
        mock_post.return_value.raise_for_status = lambda: None
        issue_vault_keys(customers=customers, billing_period="2026-07")

    calls = mock_post.call_args_list
    assert len(calls) == 2, "One vault key must be issued per customer"
    caps = [c.kwargs["json"]["daily_usd_cap"] for c in calls]
    assert caps[0] == pytest.approx(11.00), "cus_A cap should be $10 × 1.10"
    assert caps[1] == pytest.approx(55.00), "cus_B cap should be $50 × 1.10"

def test_restart_safe_pre_flight_returns_existing_keys():
    customers = [{"customer_id": "cus_A", "amount_cents": 1000, "billing_period": "2026-07"}]
    with patch("issue_vault_keys_worker.db") as mock_db:
        mock_db.query.return_value = [{"customer_id": "cus_A", "vault_key": "vk_existing"}]
        result = issue_vault_keys(customers=customers, billing_period="2026-07")
    assert result["vault_keys"]["cus_A"] == "vk_existing", "Restart must reuse original vault keys"

Frequently asked questions

Can I use Conductor's built-in HTTP task to call Stripe instead of a worker?

You can, but it's not recommended for billing tasks. Conductor's HTTP task type does not expose an idempotency key field in the task definition, so every retry of the HTTP task sends a new POST to Stripe without deduplication. If you must use the HTTP task for Stripe calls, point it at proxy.keybrake.com/stripe/v1/charges and set a stable Idempotency-Key header using workflow variable interpolation — ${workflow.workflowId}-${task.referenceTaskName} gives you a key that is unique per task instance and stable across retries of that instance. Worker tasks with explicit idempotency key computation are more transparent and easier to test.

What is the difference between retry and restart for billing workflows?

POST /workflow/{id}/retry re-runs from the first FAILED task, replaying COMPLETED task outputs. Billing tasks that reached COMPLETED are not re-executed — this is safe even without idempotency keys. POST /workflow/{id}/restart re-runs from task 1, re-executing everything including COMPLETED tasks — this re-calls stripe.charges.create() on every customer. The idempotency key pattern makes restart safe. If your team uses the Conductor UI, consider adding a workflow-level policy that blocks restart on billing workflows via Conductor's RBAC task permissions, leaving only retry available to operators.

How long should vault key TTLs be for large FORK_JOIN fan-outs?

Issue vault keys with a TTL equal to your expected FORK_JOIN completion time plus a generous buffer — at minimum expected_completion_seconds × 2. If 500 customers each take up to 5 seconds of worker processing time and you have 50 parallel workers, the expected completion time is 50 seconds, so a 300-second (5-minute) TTL is reasonable. Add buffer for Conductor queue latency and task scheduling jitter. If the FORK_JOIN runs longer than the TTL, workers holding expired vault keys will receive 401 responses from the proxy — their tasks will fail, and Conductor will retry them, but issue_vault_keys will return the existing keys from the database rather than issuing new ones (per the restart-safe pre-flight pattern above). The new workers will use the same idempotency keys, and Stripe will return the original charge objects.

Is the Stripe idempotency key the same as the Conductor task idempotency key?

No. Conductor has its own task idempotency key concept (taskId uniqueness within a workflow) that prevents duplicate task scheduling within the Conductor control plane. The Stripe idempotency key operates at the Stripe API layer and prevents duplicate charges when Stripe receives the same key on multiple requests. You need both: Conductor's task deduplication prevents duplicate task instances from being scheduled, while Stripe's idempotency key prevents duplicate charges if a task does get retried or restarted. They are complementary, not interchangeable.

What happens if the issue_vault_keys task itself fails mid-issuance?

If issue_vault_keys fails after issuing keys for some customers but not all, Conductor retries it according to its retry policy. On retry, the pre-flight check queries the database for existing keys — customers who already had keys issued will have their existing keys returned; customers who did not will have new keys issued. The database INSERT uses INSERT OR IGNORE so the pre-flight check is idempotent. This does mean the retry of issue_vault_keys may issue some new vault keys with fresh TTLs while returning older keys for others — that is fine, as long as all keys are valid when the FORK_JOIN executes.

How do I handle Stripe card declines without causing Conductor to retry the billing task?

Return the decline as a structured output dict rather than raising an exception. When a worker raises an uncaught exception (or explicitly calls task_client.update_task with status=FAILED and a retriable error reason), Conductor re-queues the task up to retryCount times. A declined card is not retriable — the customer's card will still be declined on the next attempt. Catch stripe.error.CardError and stripe.error.InvalidRequestError and return them as fields in your task's output dict. Conductor will mark the task COMPLETED (because the worker returned a result), and your downstream tasks can inspect the status field to route the workflow into a dunning or notification branch rather than a failure terminal.