Custody Permits

Create and manage custody permits (authorizations) for vehicle custody transfers.

Overview

Custody permits authorize specific parties to take custody of vehicles. Each authorization specifies:

• VIN: The vehicle being authorized
• Route: Origin and destination locations
• Person Identity: Who can take custody (from VECU IDV SDK)
• Role: Owner, admin, or driver
• Validity Period: When the authorization is valid

Methods

create()

Create a new custody authorization for a vehicle.

Returns: Authorization object with authorization_id, status, version (multi-driver only — monotonic version for If-Match ETag concurrency on later update_drivers() / assign_to_driver() calls), driver_authorization_mode, etc.

Raises:

• DuplicateAuthorizationError — 409, an active authorization already exists for (VIN, origin) regardless of client_id or organization_id. ADR-042 enforces strict uniqueness at the (VIN, geohash) marker; cross-tenant collisions deny the existing authorization details (no existing_authorization_id returned).
• PoolFanOutError — 503 POOL_FANOUT_FAILURE, multi-driver POST fan-out into the authorization pool failed partway and was rolled back. Retry the entire POST — idempotent at (VIN, origin). Body carries succeeded_driver_count, rolled_back_entries, and rollback_failures (any composite IDs left orphaned at auth-pool that need ops attention).
• MultiDriverModeRequiredError — 400, the client is configured for single_driver mode but the request supplied authorized_drivers. Subclass of InvalidConfigurationError. Contact your VECU administrator to opt the client into multi-driver mode.
• DriverLimitExceededError — 429, the requested authorized_drivers list exceeds the per-client cap (default 25). NO state changes applied. Body carries attempted_count and max_allowed.
• ValidationError — 400, e.g., both person_identity_key AND authorized_drivers supplied (CONFLICTING_DRIVER_FIELDS), or neither supplied (MISSING_DRIVER).

Example (single-driver — unchanged from ADR-040 era):

from vecu_custody import CustodyClient

client = CustodyClient.sandbox(token="your-jwt-token")

# Create an authorization
auth = client.authorizations.create(
    vin="9HGBH41JXMN999999",
    origin="LOC-AUCTION-MANHEIM-ATLANTA",
    destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
    person_identity_key="vecu_gfTRAjYnn_y-8zj-aBc4dEf5",
    authorized_by="system-auction-integration",
    make_model="Honda Accord 2023"
)

print(f"Authorization ID: {auth.authorization_id}")
print(f"Status: {auth.status}")
print(f"Expires: {auth.expires_at}")

Example (multi-driver — ADR-042):

from vecu_custody import CustodyClient
from vecu_custody.models import AuthorizedDriver
from vecu_custody.exceptions import (
    DriverLimitExceededError,
    MultiDriverModeRequiredError,
    PoolFanOutError,
)

client = CustodyClient.sandbox(token="your-jwt-token")

# Create a multi-driver authorization with three eligible drivers
try:
    auth = client.authorizations.create(
        vin="9HGBH41JXMN999999",
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        authorized_drivers=[
            AuthorizedDriver(
                person_identity_key="vecu_gfTRAjYnn_y-8zj-aBc4dEf5",
                name="Alice Driver",
            ),
            AuthorizedDriver(
                person_identity_key="vecu_kMxLWqPrBn4-aT2bCdEf5gHi",
                name="Bob Driver",
            ),
            AuthorizedDriver(
                person_identity_key="vecu_pQrStUvWx-9876yZa",
                name="Carol Driver",
            ),
        ],
        authorized_by="system-auction-integration",
        make_model="Honda Accord 2023",
        actor_identifier="ops-user-42@client.example.com",
    )

    # Multi-driver responses include a monotonic version (use as ETag on
    # later modifications) and the driver authorization mode stamped at
    # creation time.
    print(f"Authorization ID: {auth.authorization_id}")
    print(f"Mode: {auth.driver_authorization_mode}")  # "multi_driver"
    print(f"Version: {auth.version}")  # 1
    print(f"Drivers on the list: {len(auth.authorized_drivers)}")
except MultiDriverModeRequiredError:
    # Client is configured for single_driver mode — coordinate with VECU
    # to opt in before retrying.
    print("Client not configured for multi-driver mode")
except DriverLimitExceededError as e:
    print(f"Driver list ({e.attempted_count}) exceeds cap ({e.max_allowed})")
except PoolFanOutError as e:
    # Custody-service rolled back any partially-added pool entries. Retry
    # the entire POST — the (VIN, origin) marker has been cleared.
    print(
        f"Pool fan-out failed after {e.succeeded_driver_count} drivers; "
        f"rollback {'incomplete' if e.rollback_failures else 'complete'}"
    )

get()

Retrieve a specific authorization by ID.

Returns: Authorization object

Raises: AuthorizationNotFoundError if authorization doesn't exist

Example:

from vecu_custody import CustodyClient
from vecu_custody.exceptions import AuthorizationNotFoundError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.get("AUTH-12345678")
    print(f"VIN: {auth.vin}")
    print(f"Status: {auth.status}")
    print(f"Origin: {auth.origin}")
    print(f"Destination: {auth.destination}")
except AuthorizationNotFoundError:
    print("Authorization not found")

list()

List authorizations with optional filtering and pagination.

Returns: Iterator of Authorization objects (if auto_paginate=True) or AuthorizationListResponse (if auto_paginate=False)

Example:

from vecu_custody import CustodyClient

client = CustodyClient.sandbox(token="your-jwt-token")

# Auto-paginate through all authorizations (returns iterator)
for auth in client.authorizations.list(vin="9HGBH41JXMN999999"):
    print(f"{auth.authorization_id}: {auth.status}")

# Manual pagination (returns single page)
page = client.authorizations.list(
    vin="9HGBH41JXMN999999",
    limit=20,
    auto_paginate=False
)
print(f"Total count: {page.total_count}")
print(f"Page size: {len(page.authorizations)}")

if page.next_token:
    next_page = client.authorizations.list(
        vin="9HGBH41JXMN999999",
        next_token=page.next_token,
        auto_paginate=False
    )

cancel()

Cancel an authorization.

Returns: Authorization object with status="CANCELLED"

Raises: AuthorizationNotFoundError if authorization doesn't exist, ValidationError if authorization cannot be cancelled

Example:

from vecu_custody import CustodyClient

client = CustodyClient.sandbox(token="your-jwt-token")

# Cancel an authorization
cancelled = client.authorizations.cancel("AUTH-12345678")
print(f"Status: {cancelled.status}")  # "CANCELLED"

update_drivers()

Add or remove drivers on an existing multi-driver authorization (ADR-042 §Decision §6). Available only when the authorization was created under multi_driver mode and the calling client owns the authorization. ETag-guarded for optimistic concurrency, idempotent on add/remove operations, and atomic across pool fan-out side effects.

Returns: updated Authorization with the new version and the post-PATCH authorized_drivers list. Save the new version for the next call.

Raises:

• MultiDriverModeRequiredError — 400, the authorization was created under single_driver mode. PATCH /drivers is unavailable on single-driver authorizations even from the legitimate owner.
• NotAuthorizationOwnerError — 403, the caller's client_id does not match the authorization's client_id.
• EmptyDriverListError — 400, the net effect of the PATCH would leave zero drivers on the authorization. NO state changes applied.
• AuthorizationAlreadyAssignedError — 409, the authorization has already been assigned via assign_to_driver(). Modifying drivers on an ASSIGNED authorization is not permitted.
• StaleETagError — 412, the supplied version does not match the current version. Re-fetch via get() and retry with the fresh version.
• DriverLimitExceededError — 429, the proposed list (after applying add/remove) exceeds the per-client cap (default 25). NO state changes applied.

Example:

from vecu_custody import CustodyClient
from vecu_custody.models import AuthorizedDriver
from vecu_custody.exceptions import (
    AuthorizationAlreadyAssignedError,
    EmptyDriverListError,
    MultiDriverModeRequiredError,
    StaleETagError,
)

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    # Add Dave, remove Alice — atomic single round-trip
    updated = client.authorizations.update_drivers(
        authorization_id="123e4567-e89b-12d3-a456-426614174000",
        version=3,  # ETag from prior create() / get() / update_drivers()
        add=[
            AuthorizedDriver(
                person_identity_key="vecu_dAvEpKxYz1234567",
                name="Dave Driver",
            ),
        ],
        remove=["vecu_gfTRAjYnn_y-8zj-aBc4dEf5"],  # Alice's person_identity_key
        actor_identifier="ops-user-42@client.example.com",
    )
    print(f"New version: {updated.version}")  # 4
    print(f"Drivers now: {len(updated.authorized_drivers)}")
except StaleETagError as e:
    # Concurrent PATCH bumped the version; refetch and retry
    fresh = client.authorizations.get(
        "123e4567-e89b-12d3-a456-426614174000"
    )
    print(f"Stale version; current is {fresh.version}")
except MultiDriverModeRequiredError:
    print("Authorization was not created under multi_driver mode")
except EmptyDriverListError:
    print("PATCH would leave no drivers on the authorization")
except AuthorizationAlreadyAssignedError:
    print("Cannot modify drivers after PUT /assign")

assign_to_driver()

Lock the authorization to a single chosen driver (ADR-042 §Decision §7). Transitions the authorization to ASSIGNED, cancels non-assigned pool entries, and revokes credentials issued for non-assigned drivers. Sync-blocking: by the time the call returns, all non-assigned pool entries are cancelled and all non-assigned credentials are revoked at the credential service.

Returns: AssignAuthorizationResponse with assigned_driver_pik, new version, revoked_credential_ids (credentials revoked as part of the assign — empty pre-acceptance), and driver_authorization_mode.

Raises:

• DriverNotOnListError — 400, the requested person_identity_key is not on the authorization's authorized_drivers list. Add the driver via update_drivers() first.
• NotAuthorizationOwnerError — 403, caller does not own the authorization.
• AuthorizationAlreadyAssignedToDifferentDriverError — 409, the authorization is already ASSIGNED to a different driver. Reassignment is not permitted; cancel and recreate to assign a different driver.
• AuthorizationNotAssignableError — 409, the authorization is in a terminal status (RELEASED, EXPIRED, REVOKED, or CANCELLED).
• StaleETagError — 412, version mismatch.
• AssignPartialFailureError — 503 ASSIGN_PARTIAL_FAILURE, the authorization IS assigned but credential revocation completed only partially. The call is safe to retry with the bumped version (revocation is idempotent on the credential-service side). Body carries revoked_credential_ids (succeeded) and failed_revocations (each with credential_id + reason).

Idempotency: assigning the already-assigned driver to an already- ASSIGNED authorization returns 200 with the existing state and a fresh ETag — safe to retry.

Example:

from vecu_custody import CustodyClient
from vecu_custody.exceptions import (
    AssignPartialFailureError,
    AuthorizationAlreadyAssignedToDifferentDriverError,
    DriverNotOnListError,
    StaleETagError,
)

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    response = client.authorizations.assign_to_driver(
        authorization_id="123e4567-e89b-12d3-a456-426614174000",
        version=4,
        person_identity_key="vecu_gfTRAjYnn_y-8zj-aBc4dEf5",  # Alice
        actor_identifier="ops-user-42@client.example.com",
    )
    print(f"Status: {response.status}")  # ASSIGNED
    print(f"Assigned driver: {response.assigned_driver_pik}")
    print(f"Revoked credentials: {response.revoked_credential_ids}")
    print(f"New version: {response.version}")
except DriverNotOnListError as e:
    print(
        f"Driver {e.requested_person_identity_key} is not on the list. "
        "Use update_drivers() to add first."
    )
except AuthorizationAlreadyAssignedToDifferentDriverError:
    print("Authorization already ASSIGNED to a different driver — cancel-and-recreate to change.")
except AssignPartialFailureError as e:
    # Authorization IS assigned; credential-revoke partially failed.
    # Safe to retry with the bumped version.
    print(
        f"Partial revocation: {len(e.revoked_credential_ids)} ok, "
        f"{len(e.failed_revocations)} failed"
    )
    # Retry with the bumped version
    response = client.authorizations.assign_to_driver(
        authorization_id="123e4567-e89b-12d3-a456-426614174000",
        version=e.current_version,
        person_identity_key="vecu_gfTRAjYnn_y-8zj-aBc4dEf5",
    )

add_or_create_driver()

SDK-side convenience wrapper (ADR-042 §Decision §10). Tries create() first; on a 409 DUPLICATE_AUTHORIZATION for the same (VIN, origin), falls back to fetching the existing authorization via list() (auto-scoped to the caller's client_id) and update_drivers() to add the driver. Returns a unified result indicating which path was taken.

This method exists so a per-driver call shape (one driver per request) can map onto the strict-uniqueness server design without the client having to write the POST→409→GET→PATCH dance manually.

Returns: AddOrCreateDriverResult with:

• authorization — the resulting Authorization.
• created — True if the POST path was taken (new authorization); False if the PATCH fallback was taken.
• added — True if update_drivers() actually added the driver; False if the driver was already on the existing list (idempotent no-op).

Raises: any of the typed exceptions raised by create() or update_drivers(). Only the specific 409 DUPLICATE_AUTHORIZATION case triggers the PATCH fallback — other 409 reasons (e.g., AUTHORIZATION_ALREADY_ASSIGNED) bubble up unchanged. ETag-mismatch retries are bounded; exhaustion raises RetryExhaustedError.

Example:

from vecu_custody import CustodyClient
from vecu_custody.models import AuthorizedDriver
from vecu_custody.exceptions import RetryExhaustedError

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    result = client.authorizations.add_or_create_driver(
        vin="9HGBH41JXMN999999",
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        driver=AuthorizedDriver(
            person_identity_key="vecu_gfTRAjYnn_y-8zj-aBc4dEf5",
            name="Alice Driver",
        ),
        authorized_by="dispatch-system",
        make_model="Honda Accord 2023",
        actor_identifier="ops-user-42@client.example.com",
    )

    if result.created:
        print(f"Created new authorization {result.authorization.authorization_id}")
    elif result.added:
        print(f"Added driver to existing authorization {result.authorization.authorization_id}")
    else:
        print(f"Driver was already on the list (no-op)")
except RetryExhaustedError:
    # ETag conflicts persisted past the retry budget — another caller
    # is mutating the same authorization concurrently
    print("Concurrent modifications exhausted retry budget; try again later")

Response Object

Authorization

AssignAuthorizationResponse

Returned by assign_to_driver(). Extends the standard authorization shape with revocation telemetry from the sync-blocking AC-12 contract.

Common Use Cases

Auction to Dealership Transfer

from vecu_custody import CustodyClient

client = CustodyClient.sandbox(token="your-jwt-token")

# Create authorization for auction to dealership transfer
auth = client.authorizations.create(
    vin="9HGBH41JXMN999999",
    origin="LOC-AUCTION-MANHEIM-ATLANTA",
    destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
    person_identity_key="vecu_gfTRAjYnn_y-8zj-aBc4dEf5",
    authorized_by="system-auction-integration",
    make_model="Honda Accord 2023",
    transport_order_id="TO-2024-001"
)

print(f"Authorization ID: {auth.authorization_id}")
print(f"Valid until: {auth.expires_at}")

Check Authorization Status

# Get authorization details
auth = client.authorizations.get("AUTH-12345678")

if auth.status == "RELEASED":
    print("Vehicle has been released")
elif auth.status == "PENDING":
    print("Authorization pending — waiting for releasability")
elif auth.status == "REVOKED":
    print("Authorization has been revoked")
elif auth.status == "EXPIRED":
    print("Authorization has expired")

List Active Authorizations for VIN

# Get all authorizations for a specific vehicle
for auth in client.authorizations.list(vin="9HGBH41JXMN999999"):
    print(f"{auth.authorization_id}: {auth.status}")
    print(f"  Route: {auth.origin} → {auth.destination}")
    print(f"  Expires: {auth.expires_at}")

Error Handling

from vecu_custody import CustodyClient
from vecu_custody.exceptions import (
    AuthorizationNotFoundError,
    DuplicateAuthorizationError,
    ValidationError,
)

client = CustodyClient.sandbox(token="your-jwt-token")

try:
    auth = client.authorizations.create(
        vin="9HGBH41JXMN999999",
        origin="LOC-AUCTION-MANHEIM-ATLANTA",
        destination="LOC-DEALERSHIP-CARMAX-ORLANDO",
        person_identity_key="vecu_gfTRAjYnn_y-8zj-aBc4dEf5",
        authorized_by="system-integration",
        make_model="Honda Accord 2023",
    )
except DuplicateAuthorizationError as e:
    # ADR-042: existing_authorization_id is suppressed on cross-tenant
    # 409s (disclosure guard reduces to same_client). Guard for None.
    existing_id = getattr(e, "existing_authorization_id", None)
    if existing_id is not None:
        print(f"Authorization already exists: {existing_id}")
    else:
        print("(VIN, origin) already in use by a different tenant")
except ValidationError as e:
    print(f"Invalid parameters: {e.message}")

Multi-Driver Exception Hierarchy (ADR-042)

Multi-driver operations introduce additional typed exceptions. All inherit from CustodyError and the appropriate intermediate base class (e.g., MultiDriverModeRequiredError is a subclass of InvalidConfigurationError, mirroring the existing InvalidOriginAddressError precedent).

from vecu_custody.exceptions import (
    AssignPartialFailureError,
    AuthorizationAlreadyAssignedError,
    AuthorizationAlreadyAssignedToDifferentDriverError,
    AuthorizationNotAssignableError,
    DriverLimitExceededError,
    DriverNotOnListError,
    EmptyDriverListError,
    InvalidConfigurationError,  # Parent of MultiDriverModeRequiredError
    MultiDriverModeRequiredError,
    NotAuthorizationOwnerError,
    PoolFanOutError,
    RetryExhaustedError,
    StaleETagError,
)

# Catch the parent class to handle the whole config-error family at once
try:
    client.authorizations.update_drivers(...)
except InvalidConfigurationError as e:
    # Handles MultiDriverModeRequiredError + any future config errors
    print(f"Configuration issue: {e}")

Next Steps

• Multi-Driver Integration Guide - End-to-end flows for multi_driver mode
• Multi-Driver Migration Guide - Moving from exception-mode to strict-only uniqueness
• Transfers - Create and track custody transfers
• Releasability - Check if vehicles can be released