Temporal (Python)

CommitLayer for the Temporal Python SDK. Temporal makes the workflow durable. CommitLayer independently verifies the result.

Two pieces:

• an activity interceptor that records a CommitLayer action for each mapped activity before the activity executes, from the activity's own arguments;
• a completion verifier: an activity the workflow calls after its work, which waits for the verification to settle and returns the verdict, so the workflow can decide what "done" means.

The boundary, in one sentence: The agent may declare its intent. It may read the verdict. It must never control the independent facts, the governing contract, or the final evidence.

pip install 'commitlayer-integrations[temporal]'
export COMMITLAYER_API_KEY=spk_...   # read by commitlayer.Client; never log or print it

1. Record each activity as an action: the interceptor

from concurrent.futures import ThreadPoolExecutor

from commitlayer import Client
from commitlayer_temporal import CommitLayerInterceptor, CommitLayerVerifier
from temporalio.client import Client as TemporalClient
from temporalio.worker import Worker

cl = Client()  # COMMITLAYER_API_KEY / COMMITLAYER_API_URL from the environment

MAPPING = {
    # activity name -> what to record
    "refund_payment": {
        "contract_id": "refund_workflow",
        "subject": "kwargs.order_id",  # a path into the activity's arguments
        "params": {"amount": "kwargs.amount", "reason": "kwargs.reason"},
        "subject_kind": "order",
        "action_type": "refund_order",
        # "action_id": "refund:{subject_id}",               # optional template; see below
        # "contract_version": 2,                            # optional: pin the version
    },
    "reverse_refund": {
        "contract_id": "refund_reversal",
        "subject": "kwargs.order_id",
        "compensates": "kwargs.failed_action_id",  # path to the action this one corrects
    },
}


async def main() -> None:
    temporal = await TemporalClient.connect("localhost:7233")
    verifier = CommitLayerVerifier(cl)
    worker = Worker(
        temporal,
        task_queue="orders",
        workflows=[RefundWorkflow],
        activities=[refund_payment, reverse_refund, notify_customer, verifier.commitlayer_verify],
        activity_executor=ThreadPoolExecutor(8),  # the verifier is a sync activity
        interceptors=[CommitLayerInterceptor(cl, MAPPING, agent_id="refund-agent")],
    )
    await worker.run()

What gets recorded for refund_payment("order_10231", 180.0, "customer_cancelled") in workflow wf-42, activity id 7, attempt 2:

{
  "action_id": "temporal:wf-42:7",
  "contract_id": "refund_workflow",
  "workflow_id": "wf-42",
  "agent_id": "refund-agent",
  "action_type": "refund_order",
  "subject_id": "order_10231",
  "subject_kind": "order",
  "requested_at": "2026-10-15T10:42:00Z",
  "params": {"amount": 180.0, "reason": "customer_cancelled", "temporal_activity_id": "7", "temporal_attempt": 2},
  "channel": "temporal"
}

That is the body of POST /v1/actions, nothing more. The activity's return value is never sent: verification reads the systems of record (payment provider, order system, CRM), not the agent's report of its own work.

Mapping reference

Key	Meaning
contract_id	The action contract (required).
subject	Path to the subject id (required).
params	{name: path}: the parameters the contract limits (amount, tier).
action_id	Template; default temporal:{workflow_id}:{activity_id}.
action_type, subject_kind	Sent as given.
compensates	Path to the action_id of an earlier, failed action this one corrects.
contract_version	Pin the contract version the action is recorded under.

Paths: args.0.order_id (first positional argument, then key or attribute), kwargs.order_id (the argument bound to the activity function's parameter order_id; Temporal passes arguments positionally, the integration binds them to the signature), or just order_id (same as kwargs.order_id). Dicts, dataclasses and plain objects all work. A path that does not resolve is a recording failure (see fail-open), never a silent zero.

Action id: the default temporal:{workflow_id}:{activity_id} is the same for every retry attempt of one activity execution, and the API is idempotent on action_id, so retries update one action instead of recording several. Templates may use {workflow_id}, {workflow_run_id}, {activity_id}, {attempt}, {activity}, {subject_id} and any param by name ({amount}). The activity id and attempt are always in params as temporal_activity_id / temporal_attempt.

Defaults: agent_id (falls back to the client's) and workflow_id (default: the Temporal workflow id) are set on the interceptor. channel is always temporal. requested_at is stamped when the interceptor runs, before the activity, so the request time precedes every effect.

Fail-open

With fail_open=True (default), a recording failure (API unreachable, 5xx, a mapping path that does not resolve) is logged as a warning on the commitlayer.temporal logger and the activity runs anyway. The gap shows up on the CommitLayer side as an action that was never recorded, not as a business action that never happened. With fail_open=False the interceptor raises ApplicationError(type="CommitLayerRecordFailed", non_retryable=True) before the activity runs, so the activity fails without retrying and the workflow sees the failure. Nothing in the log line carries the API key.

2. Gate completion on the verdict: the verifier

Recording does not verify, and a verification is provisional until its settlement window closes and every source has caught up (status: "settled"). The verifier is a sync activity, commitlayer_verify(action_ref, timeout_seconds=900, poll_seconds=5), that waits on the API's completion gate (GET /v1/actions/{ref}/wait, held for up to poll_seconds per call and answered the moment the verdict settles), heartbeating after each call, until the verification is settled or the timeout passes. The loop lives in commitlayer_integrations.verdict.poll_until_settled; the SDK's own wait_for_settlement does the same outside a workflow engine.

It returns a small JSON-serialisable dict, safe for Temporal history:

{
    "action_id": "temporal:wf-42:7",
    "contract_id": "refund_workflow",
    "contract_version": 2,
    "status": "settled",  # "pending" | "settled" | "unverified"
    "settled": True,
    "outcome": "PASS",  # did every expected effect happen?
    "policy": "PASS",  # ...and were preconditions, limits and prohibitions respected?
    "passed": True,  # settled AND both lenses PASS: the completion signal
    "reason_codes": {"outcome": ["effect_observed:stripe.refund_created", ...], "policy": [...]},
    "reason_labels": {...},
    "headline": "PASS — all 4 expected effects observed within 15m; ...",
    "settles_at": "2026-10-15T10:57:00Z",
    "verdict_hash": "...",
    "timed_out": False,  # True when the timeout passed while still pending
}

The pattern

from datetime import timedelta

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from commitlayer_temporal import CommitLayerVerifier


@workflow.defn
class RefundWorkflow:
    @workflow.run
    async def run(self, order_id: str, amount: float, reason: str) -> dict:
        handle = workflow.start_activity(
            refund_payment,
            args=[order_id, amount, reason],
            activity_id="refund",  # fixed id: the action is temporal:<wf>:refund
            start_to_close_timeout=timedelta(minutes=2),
        )
        await handle

        action_ref = f"temporal:{workflow.info().workflow_id}:refund"
        verdict = await workflow.execute_activity(
            CommitLayerVerifier.commitlayer_verify,
            args=[action_ref, 900],  # wait up to 15 minutes for settlement
            start_to_close_timeout=timedelta(minutes=20),
            heartbeat_timeout=timedelta(minutes=1),
        )

        if verdict["passed"]:
            # Only now: the systems of record show the refund, the cancellation, the note and the message.
            await workflow.execute_activity(
                notify_customer, args=[order_id, "done"], start_to_close_timeout=timedelta(minutes=1)
            )
            return {"status": "completed", "verdict": verdict}

        if verdict["settled"]:
            # FAIL (or UNKNOWN) under a lens: route to compensation. The compensating activity's action
            # carries compensates=<action_ref>, so the failed action is marked COMPENSATED once the
            # correction itself settles PASS. The original verdict never changes.
            await workflow.execute_activity(
                reverse_refund, args=[order_id, action_ref], start_to_close_timeout=timedelta(minutes=2)
            )
            return {"status": "compensated", "verdict": verdict}

        # timed_out while pending (UNKNOWN past the deadline): the workflow's decision. Wait again, escalate
        # to a person, or hold the confirmation. Do not report success.
        return {"status": "unresolved", "verdict": verdict}

Three rules this pattern enforces:

1. No "done" before PASS. The confirmation to the customer runs only on verdict["passed"], which is a settled PASS under both lenses. A provisional PASS is not a completion signal.
2. FAIL routes to compensation. The compensation is itself a recorded action (compensates in the mapping), so it is verified on the same terms; nothing is marked resolved on anyone's say-so.
3. UNKNOWN past the timeout is the workflow's decision. The verifier reports timed_out: True with the last status it saw; it does not guess. A workflow may wait longer (call the activity again: polls are idempotent reads), escalate, or hold.

Why a fixed activity_id: the default action id is built from the Temporal activity id, so a workflow that sets activity_id="refund" knows the action reference without passing anything around. Without it, use an action_id template over the arguments ("refund:{subject_id}") and build the same string in the workflow.

The verifier activity is synchronous because the SDK's transport is urllib; register an activity_executor on the worker. It heartbeats after every poll with {status, outcome, policy}. Cancellation reaches it as Temporal's CancelledError in its thread; between polls it also checks activity.is_cancelled().

Testing without a server

The interceptor's body-building is a pure function, commitlayer_temporal.build_action_body(mapping, ref=..., args=..., kwargs=...), and temporalio.testing.ActivityEnvironment runs both the interceptor and the verifier with activity.info() and heartbeats in place. See integrations/python/tests/test_temporal_*.py: a fake transport on commitlayer.Client(transport=...) captures the exact request bodies.