Temporal

A Temporal workflow can carry one bounded grant into every activity it schedules. ChioActivityInterceptor, registered on the activity worker, has the kernel evaluate each activity execution under that grant before the activity body runs.

Temporal keeps scheduling, retries with backoff, history, visibility, and saga compensation. Workflow workers orchestrate and never call Chio, so replay stays deterministic. The activity worker is the enforcement point, because Temporal treats activities as the side-effect boundary, and the Chio sidecar runs beside it: in the same pod on Kubernetes, on the same host on a VM.

A refusal raises a non-retryable error, so Temporal ends the activity instead of retrying an operation the kernel refused. Allowed or refused, each executed activity adds a step to one receipt for its run.

1. Install the adapter on the activity worker

chio-temporal depends on the Chio Python SDK, chio-adapter-base>=0.2.0,<0.3, temporalio>=1.7,<2, and pydantic>=2.5,<3. Install it where your activity worker runs.

You should see Temporal adapter ready.

2. Register the interceptor and the workflow's grant

Register one interceptor on the worker, then register a WorkflowGrant for each workflow before it starts. In this excerpt, CHIO_TEMPORAL_CAPABILITY_FILE names a signed capability token covering quote and publish_price on pricing and read_order on orders; your application supplies the workflow, the activities, the authority client, and the receipt sink.

import asyncio
import os
from pathlib import Path

from application import (
    RepricingWorkflow, authority, quote, my_receipt_sink, read_order, publish_price,
)
from chio import CapabilityToken, ChioScope, Operation, ToolGrant
from chio_temporal import ChioActivityInterceptor, WorkflowGrant
from temporalio.client import Client
from temporalio.worker import Worker


def scope_for(*tools: str, server_id: str = "pricing") -> ChioScope:
    """Local helper. The SDK ships ChioScope; scope_for is yours."""
    return ChioScope(
        grants=[
            ToolGrant(server_id=server_id, tool_name=tool, operations=[Operation.INVOKE])
            for tool in tools
        ]
    )


async def main() -> None:
    client = await Client.connect("localhost:7233")

    interceptor = ChioActivityInterceptor(
        chio_client=authority,
        sidecar_url="http://127.0.0.1:9090",
        # Fallback tool_server id when an activity has no explicit mapping.
        default_tool_server="pricing",
        # Per-activity-type mapping. Activities usually map 1:1 to a tool server.
        activity_tool_server_map={
            "quote":         "pricing",
            "read_order":    "orders",
            "publish_price": "pricing",
        },
        # Optional: drain finalized workflow receipts to a sink.
        receipt_sink=my_receipt_sink,
        # Optional: override argument redaction. Defaults to
        # RedactionPolicy.chio_default() when omitted.
        redaction_policy=None,
    )

    # Before starting the workflow, register the grant for its workflow_id.
    token = CapabilityToken.model_validate_json(
        Path(os.environ["CHIO_TEMPORAL_CAPABILITY_FILE"]).read_text()
    )
    grant = WorkflowGrant(workflow_id="reprice-order-77", token=token, tool_server="pricing")
    interceptor.register_workflow_grant(grant)

    worker = Worker(
        client,
        task_queue="pricing-tasks",
        workflows=[RepricingWorkflow],
        activities=[quote, read_order, publish_price],
        interceptors=[interceptor],
    )
    await worker.run()


asyncio.run(main())

The interceptor looks up the grant when the first activity for a (workflow_id, run_id) pair executes. It evaluates each activity through the sidecar under the grant's capability and records a WorkflowStepReceipt on that run's WorkflowReceipt. State is keyed on the pair, so concurrent workflows on one worker keep separate step lists.

Scope travels in the grant's CapabilityToken.scope; the interceptor takes no scope_map. activity_tool_server_map maps activity types to tool server ids, and default_tool_server covers the rest. Keep every Chio call inside activities, because workflow code must stay deterministic for replay.

3. Narrow one activity below the workflow grant

A grant binds to a workflow_id and, with optional run_id pinning, to one execution; otherwise it covers every run of that id. attenuate_for_activity hands one activity type a narrower grant through an override hook.

narrower = await grant.attenuate_for_activity(
    authority,
    new_scope=scope_for("read_order", server_id="orders"),
    tool_server="orders",
)

# Register a per-activity-type hook that returns the narrower grant.
interceptor.register_activity_grant_override(
    "read_order",
    lambda info: narrower,
)

The hook receives Temporal's activity.Info for each execution, and returning None falls back to the workflow grant. The interceptor checks that the override scope is a subset of the workflow grant's scope and raises ChioTemporalConfigError otherwise.

attenuate_for_activity checks the subset rule, then delegates to the client's attenuate_capability, which signs with the parent subject's key. Keep that key with its owner, never in the sidecar: use a signer-owning client, or register an already issued narrower grant with the same hook, preserving the workflow binding and parent scope.

4. Match a refusal and run your compensation

On a deny, the interceptor records the deny receipt on the run's WorkflowReceipt and raises a non-retryable ApplicationError of type "ChioCapabilityDenied". The workflow receives an ActivityError whose cause is that error.

from datetime import timedelta
from temporalio import workflow
from temporalio.exceptions import ActivityError, ApplicationError
from chio_temporal import DENIED_ERROR_TYPE

@workflow.defn
class TransferWorkflow:
    @workflow.run
    async def run(self, transfer: Transfer) -> TransferResult:
        compensations: list[tuple[str, list]] = []
        debit = await workflow.execute_activity(
            debit_account,
            args=[transfer.source, transfer.amount],
            start_to_close_timeout=timedelta(seconds=30),
        )
        compensations.append(("credit_account", [transfer.source, transfer.amount]))

        try:
            credit = await workflow.execute_activity(
                credit_account,
                args=[transfer.destination, transfer.amount],
                start_to_close_timeout=timedelta(seconds=30),
            )
        except ActivityError as exc:
            cause = exc.cause
            if isinstance(cause, ApplicationError) and cause.type == DENIED_ERROR_TYPE:
                for name, args in reversed(compensations):
                    await workflow.execute_activity(
                        name, args=args,
                        start_to_close_timeout=timedelta(seconds=30),
                    )
            raise

        return TransferResult(debit=debit, credit=credit)

The excerpt assumes your Transfer types and account activities, and each compensating activity needs its own valid authority. Sidecar transport faults raise ApplicationError(type="ChioSidecarError", non_retryable=False), so Temporal applies its usual retry policy to them and to nothing policy refused.

5. Close the run and flush its workflow receipt

Each executed activity appends a WorkflowStepReceipt. When the workflow completes, finalize_workflow closes the envelope and flush_workflow_receipt forwards it to the configured sink.

# At workflow completion (typically from a "finalize" activity).
interceptor.finalize_workflow(
    workflow_id="reprice-order-77",
    run_id=run_id,
    outcome="success",   # or "failure" | "cancelled"
)
envelope = await interceptor.flush_workflow_receipt(
    workflow_id="reprice-order-77",
    run_id=run_id,
)
# envelope["version"] == "chio-temporal/v1"
# envelope has: workflow_id, run_id, parent_workflow_ids, started_at,
# completed_at, outcome, step_count, allow_count, deny_count, steps, metadata

The envelope holds one step per executed activity, with the allow and deny counts beside them. Step receipts issued by the sidecar are signed. The envelope that groups them is unsigned, and a deny the sidecar answers with HTTP 403 is recorded as a synthetic unsigned step.

The interceptor returns the activity's result unchanged, so no receipt id rides back in the value. The id lands on the in-memory WorkflowStepReceipt, and the receipt body stays in the Chio receipt store. On a deny, the raised error carries ChioTemporalError.to_dict() as its details, including the receipt id and the decision, and Temporal persists those details in history like any failure.

Temporal and the kernel divide the work at each activity

Temporal schedules each activity and retries its failures; the kernel admits or refuses the activity before its body runs.

An interceptor around Python code adds admission. Isolation of the activity's own process comes from the host, as the mediation boundary describes, and recovery of its external effects comes from running governed tools in hosted process workers.

Adapter reference

from chio_temporal import build_chio_worker

worker, interceptor, grant = await build_chio_worker(
    client,
    task_queue="pricing-tasks",
    activities=[quote, read_order, publish_price],
    workflows=[RepricingWorkflow],
    chio_client=authority,
    workflow_id="reprice-order-77",
    # Reuse the existing capability; omit scope and subject.
    capability_id=token.id,
    default_tool_server="pricing",
    activity_tool_server_map={"read_order": "orders"},
)

await worker.run()

sdks/python/chio-temporal/
  pyproject.toml                # deps: the Chio Python SDK, chio-adapter-base, temporalio
  src/chio_temporal/
    __init__.py                 # public API
    interceptor.py              # ChioActivityInterceptor, ActivityGrantOverride, DENIED_ERROR_TYPE
    grants.py                   # WorkflowGrant (+ attenuate_for_activity)
    receipt.py                  # WorkflowReceipt, WorkflowStepReceipt, ENVELOPE_VERSION
    worker.py                   # build_chio_worker
    errors.py                   # ChioTemporalError, ChioTemporalConfigError
    py.typed
  tests/
    test_interceptor.py
    test_workflow_receipt.py
    test_redaction.py
    test_temporal_redaction_helpers.py

Next: admit each event a consumer reads