Temporal Integration Reference

Applies to tenuo 0.3.2.

This is the deep reference for Tenuo’s Temporal integration. For the getting-started guide, see Temporal Integration.

Tenuo concepts for Temporal developers

If you’re coming from Temporal’s RBAC or namespace-based access control, here’s the mental model shift:

Two keys, two roles:

Issuer (control_key)                Holder (agent_key)
────────────────────                ─────────────────
Owned by: authorization team        Owned by: worker / CI / agent process
Lives in: Vault, KMS, CI secret,    Lives in: the workflow worker's memory
          or Tenuo Cloud
Used to: mint warrants               Used to: sign PoP on each activity dispatch
If compromised: rotate trusted root  If compromised: restart the worker with a
                                     new key, then re-issue the warrant

The issuer key never touches the worker. The holder key never leaves the worker. Headers carry only the holder key_id and warrant material, not private keys.

Path to production

Checklist for moving past local demos (each item stands alone; links go deeper):

1. Issuer vs holder keys — Issuer (control_key) only mints warrants. The workflow worker signs with a holder key already in memory: signing_key=, or a KeyResolver whose resolve_sync returns that key without starting a thread. EnvKeyResolver is for development, after preload. VaultKeyResolver, AWSSecretsManagerKeyResolver, and GCPSecretManagerKeyResolver are not a signing path inside the workflow sandbox.
2. Preload if you still use env keys in lower envs — Call preload_keys with every holder key_id before Worker(...), because PoP signing runs in the workflow sandbox where os.environ is unavailable for non-determinism reasons.
3. Sandbox passthrough — TenuoTemporalPlugin handles this automatically. If using TenuoWorkerInterceptor manually, you must set SandboxRestrictions.default.with_passthrough_modules("tenuo", "tenuo_core") so PyO3 can load once; without it, workflow tasks fail with ImportError: PyO3 modules may only be initialized once... (details).
4. Named argument constraints — If the warrant constrains fields like path= or bucket=, set activity_fns to the same callables as Worker(activities=[...]), or use tenuo_execute_activity(), so PoP can name arguments correctly.
5. Starting workflows under concurrency — Prefer execute_workflow_authorized(...) so Tenuo headers are bound to workflow_id and are not mixed across parallel starts.
6. Authorized child workflows — Use only tenuo_execute_child_workflow(); the stock workflow.execute_child_workflow() does not propagate warrant headers.
7. Replicas and PoP replay — If more than one worker replica can observe the same first activity attempt, use a shared PopDedupStore; if Temporal retries span longer than your PoP time window, tune retry_pop_max_windows.
8. Issuer rotation without full redeploy — Use a trusted_roots_provider with a short refresh interval so new issuer keys propagate quickly.
9. Cross-namespace Nexus calls — For Temporal Nexus, use the dedicated Temporal Nexus Authorization helpers. Nexus has different retry and idempotency semantics from Activities, so do not assume Activity replay controls are enough at a Nexus Endpoint boundary.

Package layout

All documented symbols can be imported from the top-level package (from tenuo.temporal import X). The package uses lazy loading so only the symbols you reference are imported. For direct imports in library or internal code, the canonical submodule homes are:

TenuoWorkerInterceptor (manual setup)

For cases where you need manual control over interceptors and the sandbox runner (instead of TenuoTemporalPlugin):

from temporalio.client import Client
from temporalio.worker import Worker
from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner, SandboxRestrictions
from tenuo import SigningKey
from tenuo.temporal import (
    TenuoWorkerInterceptor,
    TenuoPluginConfig,
    TenuoClientInterceptor,
    EnvKeyResolver,
    TENUO_TEMPORAL_ACTIVITIES,
)

control = SigningKey.generate()

client_interceptor = TenuoClientInterceptor()
client = await Client.connect("localhost:7233", interceptors=[client_interceptor])

config = TenuoPluginConfig(
    key_resolver=EnvKeyResolver(),
    trusted_roots=[control.public_key],
)

# Pass the same task_queue the Worker uses; the interceptor self-registers
# this config under that queue so workflow_grant / tenuo_execute_child_workflow
# can find the right key resolver when minting attenuated warrants.
worker_interceptor = TenuoWorkerInterceptor(config, task_queue="q")

sandbox_runner = SandboxedWorkflowRunner(
    restrictions=SandboxRestrictions.default.with_passthrough_modules("tenuo", "tenuo_core")
)
worker = Worker(
    client,
    task_queue="q",
    workflows=[...],
    # TENUO_TEMPORAL_ACTIVITIES is required — the plugin path injects it
    # automatically, manual setups must splat it in by hand. Without it,
    # workflow_grant() and tenuo_execute_child_workflow(constraints=...)
    # have no mint activity to dispatch against and fail at runtime.
    activities=[*my_activities, *TENUO_TEMPORAL_ACTIVITIES],
    interceptors=[worker_interceptor],
    workflow_runner=sandbox_runner,
)

task_queue= is required for delegation. If you omit it, basic authorization (activity PoP, constraint matching) still works, but calls to workflow_grant(), tenuo_execute_child_workflow(constraints=...), or delegate_warrant() will fail with a TenuoContextError the first time a workflow tries to mint an attenuated warrant. The error message names the remediation exactly — either pass task_queue= to the interceptor (as above) or call register_worker_config(config, task_queue="q") before Worker(...) starts. The plugin path (TenuoTemporalPlugin) handles this automatically; the kwarg only matters here.

If you need to construct the interceptor before knowing the queue (dynamic worker orchestration, test harnesses), use the helper:

from tenuo.temporal import register_worker_config

worker_interceptor = TenuoWorkerInterceptor(config)
# ... later, when the queue is known ...
register_worker_config(config, task_queue="q")

API Ergonomics

Recommended: execute_workflow_authorized(...)

The safest way to start authorized workflows. Binds headers to a specific workflow ID and executes immediately. When the client was created with TenuoTemporalPlugin, the interceptor is discovered automatically — no need to pass client_interceptor.

result = await execute_workflow_authorized(
    client=client,
    workflow_run_fn=DataProcessingWorkflow.run,
    workflow_id="process-001",
    warrant=warrant,
    key_id="agent-key-1",
    args=["/data/input/report.txt", "/data/output/report.txt"],
    task_queue="data-processing",
)

Long-running workflows: start_workflow_authorized(...)

For workflows where you need a handle to signal, query, or await later (human-in-the-loop gates, multi-day pipelines):

handle = await start_workflow_authorized(
    client=client,
    workflow_run_fn=ApprovalWorkflow.run,
    workflow_id="approval-001",
    warrant=warrant,
    key_id="agent-key-1",
    args=[request_data],
    task_queue="approvals",
)

# Signal later
await handle.signal(ApprovalWorkflow.approve, decision)
result = await handle.result()

Same header binding as execute_workflow_authorized() — but returns a WorkflowHandle immediately instead of blocking on the result.

Advanced: set_headers_for_workflow(...) + client.execute_workflow(...)

Use this when you need manual control over start timing or custom wrappers.

client_interceptor.set_headers_for_workflow(
    "process-001",
    tenuo_headers(warrant, "agent-key-1"),
)
result = await client.execute_workflow(
    DataProcessingWorkflow.run,
    id="process-001",
    args=["/data/input/report.txt", "/data/output/report.txt"],
    task_queue="data-processing",
)

Deprecated: set_headers(...)

set_headers(...) remains for backward compatibility but is deprecated for concurrent usage. Prefer workflow-ID-bound APIs.

Cross-Process Contract

For distributed deployments (separate client and worker processes):

Configuration

Key Management (REQUIRED)

Tenuo NEVER transmits private keys in headers. The workflow worker signs PoP by calling resolve_sync inside the workflow sandbox, so that call has to return a key already in memory.

Pass signing_key= for one holder key, or a KeyResolver whose resolve_sync returns from memory and does not start a thread or do I/O. Load that key before Worker(...). A later change in Vault or a cloud secret store is picked up when the workflow worker restarts with the new key. EnvKeyResolver is for development, and only after preload_all().

VaultKeyResolver, AWSSecretsManagerKeyResolver, and GCPSecretManagerKeyResolver run resolve_sync on a thread pool. The sandbox blocks that thread on every dispatch, including when cache_ttl still holds the key, so warming the cache does not make them usable for workflow signing. The constructors below show how those stores are addressed from ordinary Python. They are not the resolver a sandboxed workflow worker should use to sign.

Signing from memory

from tenuo.temporal import KeyResolver, KeyResolutionError, TenuoPluginConfig

# One holder key for this worker.
config = TenuoPluginConfig(
    signing_key=holder_signing_key,
    trusted_roots=[root_key.public_key],
    strict_mode=True,
)

# Several holder keys. Fill `keys` before Worker(...). Do not fetch inside resolve_sync.
class MemoryKeyResolver(KeyResolver):
    def __init__(self, keys: dict) -> None:
        self._keys = keys

    async def resolve(self, key_id: str):
        return self.resolve_sync(key_id)

    def resolve_sync(self, key_id: str):
        try:
            return self._keys[key_id]
        except KeyError as exc:
            raise KeyResolutionError(key_id=key_id) from exc

config = TenuoPluginConfig(
    key_resolver=MemoryKeyResolver(keys),
    trusted_roots=[root_key.public_key],
    strict_mode=True,
)

If both signing_key and key_resolver are set, the worker calls key_resolver.

Vault

from tenuo.temporal import VaultKeyResolver

resolver = VaultKeyResolver(
    url="https://vault.company.com:8200",
    path_template="production/tenuo/{key_id}",
    token=None,        # Uses VAULT_TOKEN env var
    mount="secret",
    cache_ttl=300,
)

cache_ttl applies only to calls that reach resolve(). Workflow signing never gets that far: resolve_sync starts a thread first. Fetch the key with ordinary Python before Worker(...), then pass the bytes through signing_key= or MemoryKeyResolver.

Store keys in Vault:

vault kv put secret/production/tenuo/agent-2024 \
  key=@signing_key.b64

AWS Secrets Manager

from tenuo.temporal import AWSSecretsManagerKeyResolver

resolver = AWSSecretsManagerKeyResolver(
    secret_prefix="tenuo/keys/",
    region_name="us-west-2",
    cache_ttl=300,
)

Same limit as Vault: this class cannot sign from inside the workflow sandbox. Load the secret before Worker(...) and keep the key in memory.

Store keys in AWS:

aws secretsmanager create-secret \
  --name tenuo/keys/agent-2024 \
  --secret-binary fileb://signing_key.bin \
  --region us-west-2

GCP Secret Manager

from tenuo.temporal import GCPSecretManagerKeyResolver

resolver = GCPSecretManagerKeyResolver(
    project_id="my-project",
    secret_prefix="tenuo-keys-",
    cache_ttl=300,
)

Same limit as Vault: this class cannot sign from inside the workflow sandbox. Load the secret before Worker(...) and keep the key in memory.

Store keys in GCP:

gcloud secrets create tenuo-keys-agent-2024 \
  --data-file=signing_key.bin \
  --project=my-project

Development: Environment Variables

from tenuo.temporal import EnvKeyResolver

resolver = EnvKeyResolver(
    prefix="TENUO_KEY_",
    warn_in_production=True,
)

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
)

EnvKeyResolver maps key_id to environment variables using the convention TENUO_KEY_<key_id>:

# From an existing key file:
export TENUO_KEY_agent1=$(cat signing_key.bin | base64)

# Or generate one inline:
export TENUO_KEY_agent1=$(python -c "from tenuo import SigningKey; import base64; k=SigningKey.generate(); print(base64.b64encode(k.secret_key_bytes()).decode())")

export TENUO_ENV=development   # suppress production warning

TenuoTemporalPlugin calls preload_all() automatically, scanning all TENUO_KEY_* variables into an in-memory cache before the sandbox activates. If using TenuoWorkerInterceptor manually, call resolver.preload_all() before Worker(...) — PoP signing runs inside the workflow sandbox where os.environ is blocked.

Warning: EnvKeyResolver is for development only. A production workflow worker should use signing_key= or a custom resolver whose resolve_sync returns a key already in memory.

KeyResolver and the workflow sandbox

PoP signing runs inside _TenuoWorkflowOutboundInterceptor.start_activity, which is inside the workflow sandbox. The default KeyResolver.resolve_sync submits resolve() to a ThreadPoolExecutor when an event loop is already running. The sandbox blocks that thread, and the cache inside resolve() is only consulted after the thread starts. A filled cache and a long cache_ttl do not avoid the failure.

A blocked call is raised at execute_activity as a non-retryable ApplicationError with type CONTEXT_MISSING (TenuoContextError). The message includes the sandbox’s RestrictedWorkflowAccessError. The Activity is not scheduled, and Temporal does not retry that workflow task. The workflow run fails unless workflow code catches the error.

Note: SigningKey.__repr__ is explicitly redacted (prints SigningKey(public_key=…, secret=[REDACTED])), so a surprise logger.info(f"{sk}") or ApplicationError(str(resolver)) will not leak secret bytes into Temporal history or the Temporal Web UI.

Composite Resolver (Fallback Chain)

from tenuo.temporal import CompositeKeyResolver, EnvKeyResolver

resolver = CompositeKeyResolver(
    resolvers=[
        memory_resolver,   # resolve_sync returns a key loaded before Worker(...)
        EnvKeyResolver(),  # development fallback; preload_all() first
    ],
    warn_on_fallback=True,
)

A Vault, AWS, or GCP child raises inside the sandbox on every call. Composite records that and tries the next child, so those classes cannot be the resolver that serves the workflow worker.

Tenuo Cloud alternative: If you prefer not to operate your own KMS or Vault, Tenuo Cloud provides managed key issuance and rotation.

Worker plugin config (TenuoPluginConfig)

from tenuo.temporal import TenuoPluginConfig

config = TenuoPluginConfig(
    signing_key=holder_signing_key,            # In-memory holder key. See Key Management.
    on_denial="raise",                         # "raise" | "log" | "skip"
    dry_run=False,                             # Shadow mode only; never for production
    trusted_roots=[control_key.public_key],
    strict_mode=True,                          # Fail-fast on ambiguous PoP with named constraints
    require_warrant=True,                      # Fail-closed: deny if no warrant
    block_local_activities=True,               # Prevent local activity bypass
    redact_args_in_logs=True,                  # Prevent secret leaks in logs
    max_chain_depth=10,                        # Max delegation depth
    audit_callback=on_audit,                   # Optional audit event handler
    metrics=TenuoMetrics(),                    # Optional Prometheus metrics
    authorized_signals=["approve"],            # Worker-wide. A name off the list fails the workflow.
    authorized_updates=["update_config"],      # Worker-wide. A name off the list rejects that update.
)

Production hardening: Every worker must supply trusted_roots (or call tenuo.configure(trusted_roots=[...])). Without them, TenuoPluginConfig raises ConfigurationError at construction time.

Denial Handling

# "raise" (default): raise TemporalConstraintViolation
# "log":             log denial and block (return None)
# "skip":            silently block (return None)
config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    on_denial="raise",
)

Dry run (staging only)

dry_run=True records authorization denials but still executes activities. Use only for rollout validation.

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[root_key.public_key],
    dry_run=True,
)

Warning: dry_run=True disables enforcement. Never use in production.

Activity registry (activity_fns) and PoP argument names

Why this matters

Each activity call gets a PoP signature over a payload that includes the tool name and a sorted argument dictionary. When your warrant has named field constraints (e.g. path=Subpath("/data")), the argument dict keys must match the Python parameter names.

Resolution order (function reference)

1. input.fn: supplied by the Temporal Python SDK when available.
2. tenuo_execute_activity(...): Tenuo records the function reference for that call.
3. TenuoPluginConfig.activity_fns: explicit registry (activity type name → function).
4. Fallback: arg0, arg1, … — correct for tool-only capabilities, wrong for named constraints.

What to configure

activities = [read_file, write_file]

interceptor = TenuoWorkerInterceptor(
    TenuoPluginConfig(
        key_resolver=EnvKeyResolver(),
        trusted_roots=[control_key.public_key],
        strict_mode=True,
        activity_fns=activities,
    ),
    task_queue="my-queue",
)

async with Worker(
    client,
    task_queue="my-queue",
    workflows=[MyWorkflow],
    activities=[*activities, *TENUO_TEMPORAL_ACTIVITIES],
    interceptors=[interceptor],
    workflow_runner=...,
):
    ...

Sandbox passthrough explained

Temporal’s Python SDK re-imports workflow code in an isolated sandbox on every task. Tenuo signs PoP inside the sandbox at execute_activity() dispatch time using tenuo_core (a PyO3 Rust extension). Both tenuo and tenuo_core must be declared passthrough.

If you omit the passthrough:

The worker appears healthy while workflow executions are dead. Diagnose via Temporal Web → find the workflow → look for repeated WorkflowTaskFailed events.

Compatibility

Proof-of-Possession

With trusted_roots in place, Tenuo enforces PoP for all warranted activity executions. The challenge is a CBOR-serialized tuple signed with Ed25519:

domain_context = b"tenuo-pop-v1"
window_ts      = (unix_now // 30) * 30          # 30-second bucket
challenge_data = CBOR( (warrant_id, tool, sorted_args, window_ts) )
preimage       = domain_context || challenge_data
signature      = Ed25519.sign(signing_key, preimage)   # 64 bytes

Two patterns for PoP:

AuthorizedWorkflow validates headers at workflow start:

@workflow.defn
class MyWorkflow(AuthorizedWorkflow):
    @workflow.run
    async def run(self, path: str) -> str:
        return await self.execute_authorized_activity(
            read_file, args=[path],
            start_to_close_timeout=timedelta(seconds=30),
        )

tenuo_execute_activity() is a free function for advanced use cases:

from tenuo.temporal import tenuo_execute_activity

return await tenuo_execute_activity(
    read_file, args=[path],
    start_to_close_timeout=timedelta(seconds=30),
)

Both automatically sign PoP; you never call warrant.sign() directly in workflows.

Security considerations

This section covers the full threat model, trust boundaries, PoP windows, dedup, root rotation, revocation, and retry drift.

Temporal’s security vs. Tenuo’s security. Temporal Cloud provides infrastructure-level security: encrypted payloads, RBAC, namespace isolation, SOC 2. Tenuo operates at the authorization layer above that: each Activity is authorized against a signed warrant before it executes, regardless of who has Temporal cluster access. A namespace admin cannot cause an activity to execute outside warrant constraints, because authorization runs on the worker.

In-process enforcement. Tenuo runs entirely within your worker process using tenuo_core. No SaaS call, no network round-trip at verify time. If Tenuo’s distribution infrastructure is unreachable, workers already running continue enforcing normally.

Trust boundaries

Protections

1. Activity without valid warrant: Denied when require_warrant=True (default).
2. Forged or tampered warrant: Chain validation ties delegated warrants back to trusted roots.
3. Execution without holder PoP: PoP binds tool name and argument map to the holder’s key.
4. Arguments outside constraints: Field constraints enforced against the same argument map used for PoP.
5. Over-broad credentials: Short TTLs, delegation / workflow_grant for least privilege, signal/update guards.
6. Mis-signing (named vs positional args): strict_mode=True fails fast on ambiguous PoP.

Clock skew and PoP time windows

Verification checks PoP using multiple aligned time windows around the verifier’s clock:

• pop_window_secs=30, pop_max_windows=5: ~±60 seconds effective skew tolerance.
• clock_tolerance_secs=30: applied to warrant lifetime/expiry, separate from PoP bucketing.

Workflow-side signing uses deterministic timestamps for Temporal replay; workers verify against their wall clock.

Replay and horizontal workers

1. Cryptographic validity: PoP is valid within the window configuration; not a one-time nonce.
2. Dedup: After verification, a dedup key (warrant facet + workflow id + run id + activity id) is recorded for attempt <= 1. Retries with attempt > 1 bypass dedup.

Default dedup is in-memory per process (InMemoryPopDedupStore). For fleet-wide suppression, implement PopDedupStore (e.g. Redis SET NX) and set TenuoPluginConfig.pop_dedup_store.

For strict Nexus replay suppression, a custom shared store should also expose check_pop_replay_for_owner(dedup_key, owner_id, now, ttl_seconds, *, activity_name). Nexus uses the PoP signature as dedup_key and the Nexus request_id as owner_id, allowing same-request redelivery while rejecting the same captured PoP on a different request.

Unlike Activities, Nexus handlers may be invoked more than once for the same operation before the handler’s workflow-id conflict policy or business idempotency runs. Keep workflow-backed Nexus operations idempotent with stable business workflow IDs, and give synchronous Nexus handlers their own idempotency key if duplicate side effects matter. See Temporal Nexus Authorization for the Nexus-specific guidance.

Trusted root rotation

Static trusted_roots require a restart to pick up new issuer keys. For rotation without restarts, use trusted_roots_provider + trusted_roots_refresh_interval_secs. During rotation, return overlapping old and new issuer keys. On refresh failure, the worker retains the previous Authorizer and logs a warning.

Signed revocation lists loaded through Python Temporal config must be signed by one of the configured trusted_roots. This matches the one-argument binding exposed to Python today; the Rust authorizer has an explicit issuer form for deployments that separate warrant-issuing authority from revocation-list signing authority.

Out of scope

• Compromised Temporal service: address with Temporal security, not Tenuo alone.
• Compromised worker host with KeyResolver access: use HSM/KMS and minimal identity.
• Malicious workflow code: Tenuo constrains activities, not arbitrary Python in workflows.
• dry_run=True: disables enforcement; staging only.
• Local activities: bypass the interceptor unless @unprotected and block_local_activities allows it.

Temporal activity retries and PoP time-drift

PoP is signed at workflow.now() when the Activity is first scheduled. Temporal retries reuse that signature from the original ACTIVITY_TASK_SCHEDULED event. The first attempt accepts about ±60 s (pop_max_windows=5). Retries use retry_pop_max_windows, which defaults to 40: 20 windows of 30 s on the past side, about 10 minutes. Temporal’s default backoff (1 s, doubling, capped at 100 s) waits about 7 minutes across ten retries that fail immediately, and that fits. Queue time and attempt runtime use the rest of the window.

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    retry_pop_max_windows=240,   # about 1 hour on the past side
)

Warrant TTL vs. workflow lifetime

The warrant’s expires_at is checked by the activity interceptor (on the activity worker, wall-clock). It is NOT checked inside the workflow sandbox during replay — _TenuoWorkflowInboundInterceptor only enforces signal / update allowlists, and _TenuoWorkflowOutboundInterceptor signs PoP against workflow.now(), which is deterministic across replays. So a Temporal replay 3 days after the fact will re-sign PoP at the original workflow time and never raise WarrantExpired from the replay path itself.

What the TTL does bound is how long activities scheduled by that workflow can continue to dispatch. A workflow that runs for 30 days under a 1-hour warrant will start seeing activity denials (WarrantExpired) ~1 hour in, even though the workflow object itself is still valid. Pick one of:

• Short workflows (< TTL): mint a warrant whose TTL covers the worst-case workflow duration including retries and timer sleeps.
• Long workflows (> a single warrant can safely cover): treat the warrant like a short-lived session token. Use one of:
  • workflow_issue_execution(...) to mint a short-lived execution warrant from a longer-lived issuer warrant, then pass it to tenuo_execute_activity(..., warrant=execution_warrant, key_id=...).
  • workflow_grant(...) to mint a narrower per-phase delegated warrant, then pass it to the same per-Activity override (requires the parent holder key).
  • tenuo_execute_child_workflow(...) to spawn child workflows each with their own freshly-minted warrant.
  • A resolver-side key rotation so retry_pop_max_windows extends the PoP window for durable retries (see previous section).
• Unbounded workflows: structure work as a series of child workflows rather than a single long-lived parent so each fresh warrant is scoped to a bounded unit of work.

Temporal event history overhead

Each activity dispatch and each child-workflow start puts the Tenuo headers on one history event. Temporal warns at 10 MB or 10,240 events and terminates at 50 MB or 51,200 events. Those limits are fixed on Temporal Cloud and are the defaults on a self-hosted server.

x-tenuo-warrant is the leaf warrant, gzip-compressed. A delegated chain is a separate header, x-tenuo-warrant-chain, base64 of the uncompressed stack. The chain is not gzip-compressed. A longer chain grows with that base64 text.

Worked example. 200 Activities with a 3-hop chain add about 0.4–0.55 MB of Tenuo headers. Each Activity writes several events and the headers sit on one of them, so the 10,240-event warning arrives while those bytes are still under 10 MB.

Operational guidance:

1. Keep chains short. Prefer workflow_grant(...) (one issuer hop, attenuates in-process) over passing a delegated warrant through multiple external hops before it hits a worker.
2. Watch workflow.info().history_size_bytes in Temporal ≥ 1.22 and alert at, say, 1 MB; Tenuo headers are one of several contributors but one of the easier to attribute.
3. Structure very-long workflows as parent/child. Each child gets a fresh history budget. Pairs well with the TTL guidance above.

Access revocation

Tenuo Cloud manages root distribution and rotation as a first-class primitive.

Fail-closed defaults

Child Workflow Delegation

Important: workflow.execute_child_workflow() does not propagate Tenuo warrant headers. Use tenuo_execute_child_workflow().

from tenuo.temporal import tenuo_execute_child_workflow

result = await tenuo_execute_child_workflow(
    ChildWorkflow.run,
    tools=["read_file"],
    ttl_seconds=60,
    args=["/data/input"],
    id=f"child-{workflow.info().workflow_id}",
    task_queue=workflow.info().task_queue,
)

Delegation Chain Verification

When warrants are attenuated, the full chain is propagated via x-tenuo-warrant-chain. The activity interceptor calls Authorizer.check_chain() to verify every link back to a trusted root.

Signal & Update Authorization

config = TenuoPluginConfig(
    signing_key=holder_signing_key,
    on_denial="raise",
    trusted_roots=[control_key.public_key],
    authorized_signals=["approve", "reject"],
    authorized_updates=["update_config"],
)

authorized_signals and authorized_updates are optional settings on the worker config. One worker uses one pair of lists for every workflow it runs. Leave them unset (None, the default) and every signal and update name is allowed. Queries are not checked.

With authorized_signals set, a name that is not on the list raises TemporalConstraintViolation during signal handling, including a signal that arrives with the first workflow task. The plugin registers that exception as a workflow failure type, so the signal fails the workflow run. A caller who can send a signal can end the run by choosing a name that is not on the list. Set the list when that outcome is acceptable. Workflows that need different names need separate worker configs.

With authorized_updates set, the same check runs in the update validator when the update defines one, and in the update handler for every update, including an update with no validator. That failure rejects the update.

PoP Replay Protection

The activity interceptor runs dedup after PoP verification. Default: InMemoryPopDedupStore (thread-safe, process-local, 10,000-entry cap, ~3-4 MB). Temporal retries with attempt > 1 bypass dedup.

Memory footprint: ~3-4 MB at cap. For lower footprint, implement a custom PopDedupStore.

Pluggable backend: Set TenuoPluginConfig.pop_dedup_store for fleet-wide replay suppression.

Without a shared PopDedupStore, dedup is single-process only. PoP windows still bound signature age.

Decorators

@tool() - Activity-to-Tool Mapping

from tenuo.temporal import tool

@activity.defn
@tool("read_file")
async def fetch_document(doc_id: str) -> str:
    """Activity name is 'fetch_document', warrant checks 'read_file'."""
    return await storage.get(doc_id)

@unprotected - Local Activities

from tenuo.temporal import unprotected

@activity.defn
@unprotected
async def get_config_value(key: str) -> str:
    """Internal config lookup: no warrant needed."""
    return config[key]

Activities not marked @unprotected that are called via execute_local_activity() will raise LocalActivityError.

current_warrant() and current_key_id()

Read the active warrant and signing key ID from within workflow code:

from tenuo.temporal import current_warrant, current_key_id

warrant = current_warrant()    # Raises TenuoContextError if no warrant
key_id = current_key_id()      # Raises TenuoContextError if no key ID

tool_mappings - Config-Driven Name Mapping

TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    tool_mappings={
        "log_ticket_outcome": "audit_log",
        "send_notification":  "notify",
    },
)

Both tool_mappings and @tool() can coexist; tool_mappings takes precedence.

Human Approval

Gates and approvers live on the warrant. Configure approval_handler on TenuoPluginConfig, or pre-supply approvals per activity.

A call the warrant does not grant is denied before an approver is asked. That check is capability scope. It does not check issuer trust, expiry, revocation, or PoP. If a gate fires, approvals are collected next. A pending handler may raise a retryable ApplicationError, and that attempt does not run the Activity. Once approvals are returned, the activity worker still checks trust, expiry, revocation, and PoP. Waiting does not extend the warrant TTL or the PoP window.

Header encoding: x-tenuo-approvals = JSON array of base64(CBOR) strings — no outer base64 wrapper (differs from FastAPI). See Human Approvals. A malformed header fails closed: the activity is denied with a non-retryable ApplicationError of type "invalid_approval" (it is not silently treated as “no approvals supplied”), so a bad-encoding client bug is distinguishable from a legitimate insufficient-approvals retry.

TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    approval_handler=cli_prompt(approver_key=approver_key),
)

set_activity_approvals() - Pre-Supply Multisig Approvals

Call from a workflow immediately before workflow.execute_activity() when the warrant has approval gates for that activity.

One-shot: approvals attach to the next activity dispatch only, then clear. For parallel activities, call set_activity_approvals before each dispatch — not once before asyncio.gather.

from tenuo.temporal import set_activity_approvals

set_activity_approvals([signed_approval_1, signed_approval_2])
await workflow.execute_activity(
    transfer_funds,
    args=[account, amount],
    start_to_close_timeout=timedelta(seconds=30),
)

On denial, the worker raises non-retryable ApplicationError with type of "approval_required" or "insufficient_approvals". Retry explicitly from the workflow after collecting signatures — Temporal will not auto-retry auth failures.

tenuo_warrant_context() - Warrant Context for Plain Client Calls

Async context manager that sets the active warrant for client.execute_workflow() or client.start_workflow() without using execute_workflow_authorized(). Useful when you need full control over the Temporal client call.

from tenuo.temporal import tenuo_warrant_context

async with tenuo_warrant_context(warrant, key_id="agent1"):
    result = await client.execute_workflow(
        MyWorkflow.run,
        id="wf-123",
        args=["/data/report.txt"],
        task_queue="my-queue",
    )

workflow_grant() - Scoped In-Workflow Grants

from tenuo.temporal import workflow_grant

file_warrant = await workflow_grant(
    "read_file",
    constraints={"path": path},
    ttl_seconds=60,
)

contents = await tenuo_execute_activity(
    read_file,
    args=[path],
    warrant=file_warrant,
    key_id=current_key_id(),
    start_to_close_timeout=timedelta(seconds=30),
)

Constraint keys must already exist in the parent warrant.

The per-Activity override is held in a workflow task-local ContextVar, so parallel workflow tasks cannot consume one another’s warrant. When warrant_chain= is omitted, Tenuo extends the active workflow chain with the override warrant automatically.

Scheduled Workflows

create_scheduled_workflow_with_warrant(...) places Tenuo payloads in the ScheduleActionStartWorkflow.headers field consumed by the worker interceptor. It never stores warrant material in memo. Because a Schedule action is static, the same warrant is used by every trigger; choose a TTL that covers the bounded Schedule lifetime or use an issuer warrant plus per-Activity execution warrants. Temporal action settings such as execution_timeout belong in action_options=. The legacy workflow_kwargs= alias remains temporarily supported with a deprecation warning; workflow input remains positional via workflow_args=.

Async Activity Completion

tenuo_complete_async_activity(...) completes an Activity whose original dispatch was already authorized. Temporal’s completion RPC has no user-header field, so the helper validates locally before releasing the completion:

• exact task-queue worker-config selection;
• trusted-root chain, signature, linkage, and current-time validation; and
• configured revocation-list validation.

This is warrant liveness, not capability scope: the helper does not check whether the warrant authorizes completing this Activity.

Delegated warrants must include warrant_chain=[root, ..., leaf]. Validation is performed before this helper calls the completion handle; it never falls back to an unverified completion. This is a caller-side preflight, not a Temporal enforcement boundary: code that directly invokes AsyncActivityHandle.complete() can bypass it. For compatibility, callers that omit task_queue= may use the sole registered worker config with a deprecation warning. Zero or multiple registrations fail closed. Provider failures or empty results retain the config’s last accepted trusted-root snapshot; revocation-provider failures or None results likewise retain the last accepted signed revocation list. SRL refreshes must be monotonic: lower versions are rejected, and a different revoked-ID set at the same version is rejected. Re-signing the same set at the same version is accepted. TenuoTemporalPlugin validates that the worker has a non-empty task queue during worker setup, so the registry cannot be silently left unconfigured.

An empty, valid signed revocation list is different from None: at a higher version, it explicitly replaces the prior list and clears its revoked-ID set.

Temporal’s async-completion RPC cannot carry user headers. Earlier helper code computed a PoP signature but discarded it, so no downstream validator could observe it. The helper therefore does not resolve or load the holder’s private key merely to compare its public half. The original Activity dispatch remains the worker-boundary PoP authorization.

Audit Events

Every authorization decision emits a TemporalAuditEvent. Supports SOC 2 CC6.8, PCI DSS 10.2, and HIPAA audit controls.

Each event captures: warrant_id, workflow_id, workflow_run_id, tool, arguments (redacted by default), timestamp, decision (ALLOW/DENY), denial_reason.

from tenuo.temporal import TemporalAuditEvent

def on_audit(event: TemporalAuditEvent):
    audit_logger.info(event.to_dict())

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    audit_callback=on_audit,
    audit_allow=True,
    audit_deny=True,
    redact_args_in_logs=True,
)

Tenuo Cloud indexes audit receipts across all workflows and provides a queryable trail.

Observability

Prometheus Metrics

from tenuo.temporal import TenuoMetrics

metrics = TenuoMetrics(prefix="tenuo_temporal")

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    metrics=metrics,
)

# Registers:
# - tenuo_temporal_activities_authorized_total{tool, workflow_type}
# - tenuo_temporal_activities_denied_total{tool, reason, workflow_type}
# - tenuo_temporal_authorization_latency_seconds_bucket{tool}

Suggested Alerts

• Sustained increase in *_activities_denied_total
• Spikes in POP_VERIFICATION_FAILED / replay-related denials
• Key resolver failures (KEY_NOT_FOUND)
• Sudden drop in authorized activity volume

Activity Summaries (Temporal Web UI)

TenuoTemporalPlugin enriches every authorized activity with a human-readable summary in the Event History.

Pass a summary to tenuo_execute_activity() and Tenuo preserves it:

await tenuo_execute_activity(
    read_file,
    args=["/data/report.txt"],
    start_to_close_timeout=timedelta(seconds=30),
    summary="monthly sales report",
)
# UI shows: [tenuo.TenuoTemporalPlugin] read_file: monthly sales report

Summaries are capped at 200 bytes. Avoid sensitive data.

Exceptions

The main authorization exceptions include error_code for wire format compatibility:

from tenuo.temporal import (
    TemporalConstraintViolation,  # error_code: "CONSTRAINT_VIOLATED"
    WarrantExpired,               # error_code: "WARRANT_EXPIRED"
    ChainValidationError,         # error_code: "CHAIN_INVALID"
    PopVerificationError,         # error_code: "POP_VERIFICATION_FAILED"
    LocalActivityError,           # error_code: "LOCAL_ACTIVITY_BLOCKED"
    KeyResolutionError,           # error_code: "KEY_NOT_FOUND"
)

Approval denials surface as ApplicationError(non_retryable=True) with type:

• "approval_required" — gate fired, resubmit with approvals
• "insufficient_approvals" — partial multi-sig, collect more signatures

See Human Approval and Human Approvals.

Failure Semantics

Authorization failures are wrapped in ApplicationError(non_retryable=True) to prevent retrying permanent denials.

Troubleshooting

Constraint Types for AI Agent Workflows

from tenuo import (
    Subpath, UrlSafe, UrlPattern, Exact, Range, OneOf, AnyOf,
    CEL, Wildcard, Regex, NotOneOf, Pattern,
)

Zero-trust mode: When ANY argument is constrained, ALL others must also be declared. Use Wildcard() for unconstrained fields.

Attenuation: Wildcard() can attenuate to any type. Pattern("*") can only narrow to globs.

For the full constraint reference, see docs/constraints.md.

Best Practices

1. Production keys: Vault, AWS Secrets Manager, or GCP Secret Manager; not EnvKeyResolver.
2. Sandbox passthrough: always with_passthrough_modules("tenuo", "tenuo_core").
3. Named fields: set activity_fns or use tenuo_execute_activity() or strict_mode=True.
4. AuthorizedWorkflow: use when missing headers should fail at workflow start.
5. Child workflows: only tenuo_execute_child_workflow().
6. Start path: prefer execute_workflow_authorized(...) (blocks on result) or start_workflow_authorized(...) (returns handle) under concurrency.
7. Audit: wire audit_callback and keep redact_args_in_logs=True.
8. @unprotected: limit to internal, low-risk local activities.
9. TTLs: short for sensitive work; combine with trusted_roots_provider.
10. Never dry_run=True in production.
11. Multi-tenant: separate configs per tenant with distinct trusted_roots.
12. Scale: shared PopDedupStore; retry_pop_max_windows for long retries.

Migration Path (from plain Temporal)

1. Plugin: add TenuoTemporalPlugin to Client.connect(plugins=[...]) — this handles interceptors, sandbox passthrough, and key preloading in one step. (For manual control, use TenuoWorkerInterceptor + SandboxedWorkflowRunner instead; see examples README.)
2. Client: start workflows with execute_workflow_authorized(...) (or start_workflow_authorized(...) for the signal/query pattern).
3. Children: replace execute_child_workflow() with tenuo_execute_child_workflow().
4. Rollout: one task queue or tenant first, then expand.

Rollback

Route traffic to an unprotected queue while preserving workflow code. Keep as an operational fallback, not steady state.

Examples

Per-stage pipeline (from delegation.py)

# Ingest warrant: read-only
ingest_warrant = (
    Warrant.mint_builder()
    .holder(ingest_key.public_key)
    .capability("read_file", path=Subpath("/data/source"))
    .capability("list_directory", path=Subpath("/data/source"))
    .ttl(600)
    .mint(control_key)
)

# Transform warrant: write-only
transform_warrant = (
    Warrant.mint_builder()
    .holder(transform_key.public_key)
    .capability("write_file", path=Subpath("/data/output"), content=Wildcard())
    .ttl(600)
    .mint(control_key)
)

Integration QA Coverage

• tests/e2e/test_temporal_live.py, test_temporal_replay.py: in-process Temporal test server, serialization, delegation, continue-as-new, replay
• tests/e2e/test_temporal_e2e.py: mocked Temporal with real Tenuo objects: interceptors, PoP, constraints, child headers

More Information

• Temporal Documentation
• Tenuo Core Concepts
• Security Model
• Example Code