Velvet Handler-Author Guide

Last verified against: Velvet v2 (2026-05-04 UTC) Parent epic: #907 Design doc: docs/architecture/velvet/v2-rotation-flows.md Operator runbook: docs/ops/runbooks/velvet-operator.md BusAdapter source: velvet/adapters/base.py Manifest: docs/architecture/velvet/subscription-manifest.yml

Reading time: ~20 min

1. What this guide covers

You are here because you need to add support for a new credential vendor or a new delivery destination. This guide walks you through:

1. Understanding the v2 service-bus architecture and where an adapter fits.
2. Implementing BusAdapter.push() correctly.
3. Populating RotationContext and returning AdapterResult.
4. Declaring your adapter in the subscription manifest.
5. Testing your adapter against the contract.

After reading this guide you should be able to ship a new adapter without reading source code beyond the files linked here.

2. Architecture overview

In Velvet v2, a rotation is orchestrated in three stages:

Stage 1 — Verify      Auth probe with the current token (no mutations)
Stage 2 — Mint + Distribute   New token minted at vendor; pushed to all consumers
Stage 3 — Validate + Revoke   Consumers healthchecked with new token; old token revoked

Where adapters live: Stage 2 (distribute) and Stage 3 (validate). An adapter is responsible for one specific delivery destination — one row in the subscription manifest. The flow runner calls adapter.push() for every active subscriber when distributing.

What an adapter does NOT own:

• The mint step: token creation at the vendor is handled by a separate vendor module (thin, no distribution logic).
• The revoke step: revocation is handled by the flow runner calling the vendor's revoke API directly. Adapters do not revoke.
• Multiple subscribers: each adapter instance handles exactly one consumer_id.

This separation means a single credential rotation (e.g., HK_PLATFORM_FULL) can fan out to five consumers — four Heroku config-var PATCH adapters and one GitHub Actions secret adapter — without any adapter knowing about the others.

3. BusAdapter protocol

File: velvet/adapters/base.py ADR: docs/architecture/adr/0037-velvet-service-bus-subscription-model.md

3a. Abstract interface

class BusAdapter(abc.ABC):

    @abc.abstractmethod
    def push(
        self,
        credential_name: str,
        new_value: str,
        context: RotationContext,
    ) -> AdapterResult:
        ...

push() is the only abstract method. Every adapter subclass must implement it.

3b. RotationContext fields

RotationContext is a dataclass. The flow runner constructs it and passes it to push().

RotationContext.__repr__ and __str__ redact new_value as <REDACTED>. This is invariant I1: no credential value may appear in logs, audit records, or error messages.

To get a safe audit-correlation handle for the new value:

safe_hash = context.new_value_sha256()   # SHA-256 hex digest — safe to log

3c. AdapterResult fields

Constructing AdapterResult(ok=True, error_message="something") raises ValueError — the invariant is enforced at construction time.

3d. Contract rules for push()

These are non-negotiable. The flow runner treats violations as bugs.

1. Never raise. push() must catch all exceptions internally and return AdapterResult(ok=False). If an uncaught exception propagates, the flow runner marks the consumer as failed with an "adapter bug" error.

2. Never log new_value. Use context.new_value_sha256() for log correlation. Use credential_name for identification.

3. Be idempotent. Calling push() twice with the same job_id and new_value must produce the same observable outcome. If the first call succeeded, the second call must return ok=True without creating a duplicate resource at the vendor. The SHA-256 hash pattern (store _last_rotated_sha256 on the instance) is the recommended implementation.

4. No shell invocations. No subprocess.run(), os.system(), or CLI invocations of any kind. All vendor interactions must use the vendor's HTTP API or official Python SDK.

5. Feature flag gate. Call self._assert_flag_on() (or your adapter-specific flag check) at the top of push() before making any network calls.

4. The four lifecycle methods and where adapters fit

The v2 design separates concerns across four named operations. Each operation is handled by a different component:

Your adapter only implements Distribute. The flow runner calls push() once per consumer per job, in parallel across all active subscribers.

5. Companion secrets

Some vendors require storing metadata alongside the token value to enable future rotation. This uses the companion secret pattern:

• The companion secret key is: {credential_name}__{DESCRIPTOR} (double underscore)
• Examples:
• HEROKU_API_KEY__AUTH_ID — stores the Heroku OAuth authorization UUID needed to revoke the old token
• CF_DNS_EDIT_RAXX_APP__TOKEN_ID — would store the CF token ID if CF supported programmatic rotation

Your adapter should read the companion secret from an environment variable at call time (never at module load). When your adapter mints a new token, write the new companion value to vault via InfisicalWriteAdapter or directly via the Infisical API before returning AdapterResult(ok=True).

6. SSM routing for password-class credentials

Some credentials are classified as password class (e.g., database passwords, service account passwords). These live in AWS SSM Parameter Store, not Infisical, per ADR-0040 and the D2 path shape decision.

SSM path convention: /raxx/{env}/{vendor}/{name}

If your credential is password class, use SSMWriteAdapter instead of InfisicalWriteAdapter for the vault write step. The two adapters share the BusAdapter interface; you choose based on the credential class.

To classify a credential as password class: set credential_class: password in the manifest entry. The manifest validator (velvet/manifests/validator.py) enforces that password-class credentials use update_method: SSM_PUT.

7. Declaring a new adapter in the manifest

File: docs/architecture/velvet/subscription-manifest.yml ADR: docs/architecture/adr/0040-velvet-consumer-registration-static-manifest.md

Add one entry per consumer. The manifest is loaded at Velvet startup and reloaded on SIGHUP. Runtime registrations are not accepted.

Full schema

- token_name: YOUR_CREDENTIAL_NAME        # Required. Token taxonomy name.
  consumer_id: your-unique-consumer-id    # Required. Stable; never reuse after decommission.
  env: prod                               # Required. prod or staging.
  active: true                            # Required. Set false to suppress during rollout.
  update_endpoint: "https://..."          # Required. Where Velvet delivers the new value.
  update_method: PATCH                    # Required. See update_method values below.
  update_auth_token_name: SOME_TOKEN      # Required (nullable). Token Velvet uses to auth the update.
  healthcheck_endpoint: "https://..."     # Optional (null = no healthcheck).
  healthcheck_method: GET                 # Optional. HTTP verb.
  healthcheck_auth_header: "Authorization: Bearer {token}"  # Optional. {token} is replaced at runtime.
  healthcheck_success_status: 200         # Optional. Expected HTTP status.
  healthcheck_timeout_s: 15              # Optional. Default 15.
  required: true                          # Optional. Default true. If false, failure is a warning, not a blocker.
  capabilities:                           # Required.
    - update
    - healthcheck                         # Include only if healthcheck_endpoint is non-null.
  description: "Human label for console"  # Required.

update_method values

Adding to the manifest

1. Add the YAML entry.
2. Run the manifest validator locally:

python3 -m velvet.manifests.validator docs/architecture/velvet/subscription-manifest.yml

3. Open a PR. CI runs the validator as part of the manifest-lint job.
4. After merge: redeploy raxx-velvet-prod and raxx-velvet-staging. Velvet reloads the manifest on startup.
5. Verify the new consumer appears in GET /tokens/{token_name}/subscribers.

Decommissioning a consumer

Set active: false first (to suppress distribution attempts while the PR is in flight). Do not delete the entry immediately — the consumer_id must remain in the manifest as long as there are rotation_job_consumers rows referencing it in the database (retention: 2 years).

8. Testing your adapter

Every adapter must have tests covering at minimum these five scenarios:

1. Feature flag off — push() returns AdapterResult(ok=False) without making any network call.
2. Happy path — The vendor API call returns the success status code; push() returns ok=True with the correct consumer_id.
3. Terminal vendor failure — The vendor API returns a non-retryable error code (e.g., 401, 403, 422); push() returns ok=False with a non-empty error_message that contains no credential values.
4. Transient failure — The vendor API returns 5xx or a ConnectionError; push() returns ok=False with is_retry_eligible returning True.
5. Idempotency — Calling push() twice with the same new_value after a successful first call returns ok=True on the second call without making a second network call. Verify by asserting the mock was called exactly once.

Additional recommended scenarios:

• new_value does not appear in any log output at DEBUG, INFO, WARNING, or ERROR level (invariant I1).
• push() catches an unexpected exception and returns ok=False rather than propagating it.
• If your adapter uses a companion secret, verify it is written correctly after a successful push.

Test file location

velvet/tests/test_adapter_{vendor_name}.py

Test pattern (minimal)

import os
import unittest
from unittest.mock import MagicMock, patch

from velvet.adapters.base import AdapterResult, RotationContext
from velvet.adapters.your_vendor import YourVendorAdapter

_FLAG_ENV = {"FLAG_VELVET_YOUR_VENDOR_ADAPTER": "1"}

def _make_ctx(**kwargs) -> RotationContext:
    defaults = dict(
        job_id="job-test",
        credential_name="YOUR_CREDENTIAL_NAME",
        new_value="new_secret_do_not_log",
        rotate_timestamp="2026-05-04T06:00:00Z",
        env="staging",
    )
    defaults.update(kwargs)
    return RotationContext(**defaults)


class TestYourVendorAdapter(unittest.TestCase):

    def test_flag_off_returns_error_without_network_call(self):
        adapter = YourVendorAdapter(consumer_id="your-consumer-id")
        ctx = _make_ctx()
        with patch.dict(os.environ, {"FLAG_VELVET_YOUR_VENDOR_ADAPTER": "0"}):
            with patch("velvet.adapters.your_vendor.requests") as mock_req:
                result = adapter.push("YOUR_CREDENTIAL_NAME", "new_value", ctx)
        self.assertFalse(result.ok)
        mock_req.get.assert_not_called()
        mock_req.post.assert_not_called()
        mock_req.patch.assert_not_called()

    def test_happy_path_returns_ok_true(self):
        adapter = YourVendorAdapter(consumer_id="your-consumer-id")
        ctx = _make_ctx()
        mock_resp = MagicMock()
        mock_resp.status_code = 200
        mock_resp.ok = True
        with patch.dict(os.environ, _FLAG_ENV):
            with patch("velvet.adapters.your_vendor.requests.post", return_value=mock_resp):
                result = adapter.push("YOUR_CREDENTIAL_NAME", ctx.new_value, ctx)
        self.assertTrue(result.ok)
        self.assertEqual(result.consumer_id, "your-consumer-id")

    def test_terminal_failure_401(self):
        adapter = YourVendorAdapter(consumer_id="your-consumer-id")
        ctx = _make_ctx()
        mock_resp = MagicMock()
        mock_resp.status_code = 401
        mock_resp.ok = False
        with patch.dict(os.environ, _FLAG_ENV):
            with patch("velvet.adapters.your_vendor.requests.post", return_value=mock_resp):
                result = adapter.push("YOUR_CREDENTIAL_NAME", ctx.new_value, ctx)
        self.assertFalse(result.ok)
        self.assertIsNotNone(result.error_message)
        self.assertNotIn(ctx.new_value, result.error_message)  # Invariant I1

    def test_transient_failure_5xx_is_retry_eligible(self):
        adapter = YourVendorAdapter(consumer_id="your-consumer-id")
        ctx = _make_ctx()
        mock_resp = MagicMock()
        mock_resp.status_code = 503
        mock_resp.ok = False
        with patch.dict(os.environ, _FLAG_ENV):
            with patch("velvet.adapters.your_vendor.requests.post", return_value=mock_resp):
                result = adapter.push("YOUR_CREDENTIAL_NAME", ctx.new_value, ctx)
        self.assertFalse(result.ok)
        self.assertTrue(adapter.is_retry_eligible(result))

    def test_idempotency_no_second_network_call(self):
        adapter = YourVendorAdapter(consumer_id="your-consumer-id")
        ctx = _make_ctx()
        mock_resp = MagicMock()
        mock_resp.status_code = 200
        mock_resp.ok = True
        with patch.dict(os.environ, _FLAG_ENV):
            with patch("velvet.adapters.your_vendor.requests.post", return_value=mock_resp) as mock_post:
                adapter.push("YOUR_CREDENTIAL_NAME", ctx.new_value, ctx)
                adapter.push("YOUR_CREDENTIAL_NAME", ctx.new_value, ctx)
        # Second push must not make a second network call
        self.assertEqual(mock_post.call_count, 1)

9. Adapter examples

9a. Postmark (simplest — validate-only)

File: velvet/adapters/postmark.py Issue: #950

Postmark has no token-creation API. The Postmark adapter implements Stage 2 distribution as a validation step: it calls GET /server with the new token (which was written to vault by the Infisical adapter in the same fan-out) and confirms HTTP 200 means the token is live.

Key characteristics: - Single HTTP GET call. No POST or PATCH. - Idempotency: stores _last_validated_sha256 on the instance. Second push with the same value returns ok=True without a network call. - Retry logic: 5xx and network errors are transient; 4xx is terminal. - Revoke: operator-manual only (no Postmark API for revocation).

This is the right starting point for any adapter that validates rather than writes.

9b. Heroku (multi-step — full cycle)

File: velvet/adapters/heroku.py Issue: #1100

The Heroku adapter orchestrates a complete token lifecycle inside a single push() call:

1. Read old token from HEROKU_PLATFORM_API_TOKEN env var.
2. Pre-validate the old token (GET /account). Fail fast if it is already invalid.
3. Idempotency check.
4. Mint new token (POST /oauth/authorizations).
5. Validate new token (GET /account).
6. Distribute to all target apps (PATCH /apps/{app}/config-vars) — authenticates with the old token.
7. Revoke old authorization (DELETE /oauth/authorizations/{auth_id}) — authenticates with the new token.

Key characteristics: - The old token is read from an env var, not from new_value. new_value is the expected outcome and is used for idempotency detection. - Distribute uses the OLD token for auth (the target apps have not yet received the new token). - If distribute is incomplete, push returns ok=False without revoking the old token. - The companion secret HEROKU_API_KEY__AUTH_ID stores the OAuth authorization UUID needed to revoke. - All intermediate failures include whether the old token is still live (to guide recovery).

Use this pattern when the adapter must orchestrate multiple sub-steps and the order of operations matters.

9c. Infisical (simple write — two flavors)

File: velvet/adapters/infisical.py Issue: #1094

The Infisical write adapter writes the new credential value to vault. It handles the create-vs-update distinction and the folder-creation pre-condition automatically:

1. Try PATCH (update existing secret).
2. If 404: try POST (create new secret).
3. If 404 again (folder missing): create the folder, then retry POST.

Key characteristics: - Handles the Infisical 404 folder-not-found quirk automatically (feedback: vault_folder_must_exist). - Uses the D2 path convention: /raxx/v{N}/{env}/{vendor}/{name}. - vendor is derived from the credential name prefix if not supplied explicitly. - Used as the delivery mechanism for any INFISICAL_WRITE manifest entry.

Two flavors based on manifest update_method: - INFISICAL_WRITE: Velvet writes to vault using its own machine identity. Use for internal vault entries. - External HTTPS endpoints: use a dedicated adapter that handles the specific vendor's API shape.

10. Step-by-step: adding support for a new vendor

Step 1 — Understand what the vendor offers

Answer these questions before writing any code:

• Does the vendor's API support programmatic token creation (mint)? If not, use update_method: OPERATOR_MANUAL.
• Does the vendor's API support programmatic token revocation? If not, revoke is operator-manual.
• What HTTP endpoint and method does the vendor require to accept a new token value? (This drives your update_method.)
• Does the vendor have a healthcheck endpoint that accepts the token as auth and returns a distinct success status? (If yes, add healthcheck capability.)
• Does the vendor require a companion secret (e.g., an authorization ID needed for future revocation)?

Step 2 — Write the adapter

Create velvet/adapters/{vendor_name}.py. Subclass BusAdapter. Implement push().

Skeleton:

"""velvet.adapters.{vendor_name} — {VendorName}TokenAdapter"""

from __future__ import annotations
import hashlib
import logging
import os
from typing import Optional

import requests
from velvet.adapters.base import AdapterResult, BusAdapter, RotationContext

logger = logging.getLogger(__name__)
_TIMEOUT_S = 15
_FLAG_ENV_VAR = "FLAG_VELVET_{VENDOR_NAME}_ADAPTER"


def _flag_is_on() -> bool:
    val = os.environ.get(_FLAG_ENV_VAR, "").strip().lower()
    return val in ("1", "true", "yes")


class {VendorName}TokenAdapter(BusAdapter):
    """Delivers a new {vendor} credential to {consumer description}."""

    def __init__(self, consumer_id: str) -> None:
        self._consumer_id = consumer_id
        self._last_pushed_sha256: Optional[str] = None

    def push(
        self,
        credential_name: str,
        new_value: str,
        context: RotationContext,
    ) -> AdapterResult:
        # 1. Feature flag gate
        if not _flag_is_on():
            return AdapterResult(
                consumer_id=self._consumer_id,
                ok=False,
                error_message=f"{_FLAG_ENV_VAR} is off",
            )

        # 2. Idempotency check
        incoming_sha256 = hashlib.sha256(new_value.encode()).hexdigest()
        if self._last_pushed_sha256 == incoming_sha256:
            return AdapterResult(consumer_id=self._consumer_id, ok=True, http_status=200)

        # 3. Make vendor API call (wrap in try/except — push() must never raise)
        try:
            resp = requests.post(
                "https://api.vendor.example.com/update",
                headers={"Authorization": f"Bearer {new_value}"},  # never logged
                json={"key": "value"},
                timeout=_TIMEOUT_S,
            )
        except Exception as exc:
            return AdapterResult(
                consumer_id=self._consumer_id,
                ok=False,
                error_message=f"Network error: {type(exc).__name__}",
            )

        if resp.status_code == 200:
            self._last_pushed_sha256 = incoming_sha256
            return AdapterResult(consumer_id=self._consumer_id, ok=True, http_status=200)

        return AdapterResult(
            consumer_id=self._consumer_id,
            ok=False,
            error_message=f"Vendor returned HTTP {resp.status_code}",
            http_status=resp.status_code,
        )

    def is_retry_eligible(self, result: AdapterResult) -> bool:
        if result.ok:
            return False
        if result.http_status and result.http_status >= 500:
            return True
        return False

Step 3 — Add a feature flag

Add the flag to backend_v2/api/feature_flags.yaml:

FLAG_VELVET_{VENDOR_NAME}_ADAPTER:
  default: false
  description: "Enable {VendorName} BusAdapter push() calls (Velvet v2)"

Step 4 — Add manifest entries

Add one entry per consumer to docs/architecture/velvet/subscription-manifest.yml. See section 7 for the full schema.

Run the manifest validator:

python3 -m velvet.manifests.validator docs/architecture/velvet/subscription-manifest.yml

Step 5 — Write tests

velvet/tests/test_adapter_{vendor_name}.py covering the five required scenarios from section 8.

Step 6 — Register the adapter in the factory

Add your adapter class to velvet/adapters/__init__.py:

from velvet.adapters.your_vendor import YourVendorAdapter  # noqa: F401

And register it in velvet/registry/adapter_factory.py:

ADAPTER_REGISTRY["{update_method_name}"] = YourVendorAdapter

Step 7 — PR and deployment

1. Open a PR. Link it to parent epic #907 and to the relevant B-slot card.
2. After merge: redeploy raxx-velvet-staging first. Run a testing-flow rotation against a throwaway token to confirm the adapter works end-to-end.
3. After staging validation: redeploy raxx-velvet-prod and set the feature flag to true on the prod Heroku config vars.

11. Security invariants (required for every adapter)

12. Checklist before opening a PR

• [ ] push() never raises (bare except Exception at the outermost level)
• [ ] new_value does not appear in any log statement at any level
• [ ] new_value does not appear in any AdapterResult.error_message
• [ ] Feature flag gates all network calls
• [ ] Idempotency: second push with same new_value returns ok=True without a network call
• [ ] is_retry_eligible() correctly returns True for transient failures, False for terminal failures
• [ ] No subprocess.run() or shell invocation
• [ ] Manifest entry added and validator passes locally
• [ ] Test file covers all five required scenarios
• [ ] Flag added to backend_v2/api/feature_flags.yaml with default: false
• [ ] Companion secret documented if applicable