Advanced Topics

Long-running execution

A long-running Provider should remain observable, respond to cancellation at safe points, bound its memory and Progress rate, and return only after it has a factual terminal outcome.

async def on_execute(self, request, context):
    context.cancellation.raise_if_cancelled()
    operation = await self._backend.start(request.inputs)

    while True:
        context.cancellation.raise_if_cancelled()
        state = await self._backend.poll(operation)

        if state.complete:
            return CapabilityExecutionResult(output=state.output)

        await context.progress.report(
            {"phase": state.phase, "percentComplete": state.percent}
        )

The SDK does not choose a polling interval or rate-limit Progress. The Provider must select values appropriate for its backend and declared Action contract.

Cancellation responsibilities

Cancellation involves three separate facts:

• Runtime or a test/control integration requests cancellation.
• Provider code observes context.cancellation.
• The Capability adapter attempts to stop the underlying operation.

context.cancellation.raise_if_cancelled()

For work blocked on a Provider-controlled awaitable, wait for backend completion and cancellation concurrently:

import asyncio

backend_task = asyncio.create_task(self._backend.run(command))
cancel_task = asyncio.create_task(context.cancellation.wait_for_cancel())

done, _ = await asyncio.wait(
    {backend_task, cancel_task},
    return_when=asyncio.FIRST_COMPLETED,
)

if cancel_task in done:
    try:
        await self._backend.stop()
    finally:
        backend_task.cancel()
        await asyncio.gather(backend_task, return_exceptions=True)
    context.cancellation.raise_if_cancelled()

cancel_task.cancel()
await asyncio.gather(cancel_task, return_exceptions=True)
result = await backend_task

Own and clean up every task created by such a race. CancellableProvider.cancel(execution_id) is an optional structural protocol for integrations that can request cancellation by execution ID. Implementing it does not automatically connect it to CancellationToken. Do not call CancellationToken.request_cancel() from ordinary Action logic to simulate a Runtime request.

Physical termination

Python task completion is not evidence that a robot, motor, media stream, or remote service operation stopped. Use PhysicalTerminationStatus according to backend evidence:

When cancellation raises CancellationCapabilityError, preserve stop evidence in the integration's supported diagnostics or recovery observation; do not claim success solely to carry a termination status.

Deadlines and timeouts

request.deadline and context.deadline are absolute, timezone-aware facts. Neither object starts a timer or interrupts an awaitable.

from vega_capability_sdk import TimeoutCapabilityError

if request.deadline is not None and context.clock.now() >= request.deadline:
    raise TimeoutCapabilityError("Execution deadline has passed")

Runtime may also express timeout through cancellation. Handle both paths consistently and avoid issuing the same physical command again after uncertainty.

Managed background tasks

Use BaseCapabilityProvider.create_background_task() for Provider-owned work that should be cancelled during shutdown:

async def initialize(self, context):
    await super().initialize(context)
    self.create_background_task(
        self._monitor_backend(),
        name="backend-health-monitor",
    )

async def _monitor_backend(self):
    try:
        while True:
            await self._backend.refresh()
            await asyncio.sleep(5)
    finally:
        await self._backend.release_monitor()

Managed tasks can be created only while the Provider is ready. Because on_initialize runs while the base state is initializing, create persistent background work only after await super().initialize(context) returns, or lazily during the first execution. Do not call create_background_task() inside on_initialize.

Concurrency and shared state

The SDK does not serialize execute() calls. Design as if multiple executions can overlap unless the deployed Runtime contract explicitly guarantees otherwise.

• Keep request-specific state in local variables or an execution-keyed record.
• Protect a shared adapter only when the backend requires serialization.
• Do not store the “current execution” in one mutable Provider attribute.
• Ensure shutdown and cancellation can coexist with in-flight backend calls.
• Keep lock scope small and never hold a synchronous lock across await.

Recovery after Runtime restart

Implement RecoveryCapableProvider only when the backend exposes a durable operation reference that can be queried after process or Runtime restart.

from vega_capability_sdk import (
    ExecutionSafetyClass,
    ProviderProbeResult,
    ProviderProbeStatus,
)


class RecoverableProvider(BaseCapabilityProvider):
    @property
    def supports_status_query(self) -> bool:
        return True

    @property
    def supports_reattach(self) -> bool:
        return False

    @property
    def supports_resume(self) -> bool:
        return False

    @property
    def supports_cancel_after_restart(self) -> bool:
        return True

    @property
    def execution_safety_class(self) -> ExecutionSafetyClass:
        return ExecutionSafetyClass.STATUS_QUERY_REQUIRED

    async def probe_execution(self, execution_id, external_ref):
        if external_ref is None:
            return ProviderProbeResult(
                status=ProviderProbeStatus.UNKNOWN,
                execution_id=execution_id,
            )

        observation = await self._backend.query(external_ref)
        return ProviderProbeResult(
            status=ProviderProbeStatus.ACTIVE,
            execution_id=execution_id,
            external_ref=external_ref,
            supports_cancel_after_restart=True,
            raw={"backend_state": observation.state},
        )

When evidence is incomplete, return unknown and choose the more conservative class.

Error and performance boundaries

• Convert known backend conditions to the most specific CapabilityError.
• Allow unexpected ordinary exceptions to reach the base Provider wrapper.
• Do not mark a failure retryable merely because retry is convenient.
• Move blocking SDK or device calls off the event loop using an adapter strategy appropriate to the dependency.
• Bound request sizes, buffers, Artifact writes, Progress frequency, logs, and diagnostics.
• Never expose raw vendor responses without reviewing secrets and personal data.

Test matrix

• Cancellation before backend start.
• Cancellation during backend work.
• Backend stop acknowledgement, uncertainty, and failure.
• Deadline already expired.
• Simultaneous executions against shared state.
• Shutdown with managed and execution-local tasks.
• Each recovery probe status and contradictory probe rejection.
• Service unavailable during recovery.
• Output and Progress schema compliance.

Common pitfalls

• Treating task cancellation as physical termination.
• Leaking raw execution-local tasks after a race.
• Calling create_background_task() during on_initialize.
• Replaying work inside probe_execution.
• Reporting not_found when the backend itself is unavailable.
• Assuming resource leases serialize Provider code.
• Returning optimistic recovery or termination facts without evidence.