Tenuo Temporal Integration

Available in tenuo[temporal]

Overview

Tenuo integrates with Temporal so each activity run is checked against a cryptographically signed warrant before your activity code executes. Activity definitions stay unchanged: no Tenuo imports inside @activity.defn functions. Workflow code must start the run with Tenuo headers (see Client below) and either call normal workflow.execute_activity() (with activity_fns configured when warrants use named argument constraints) or subclass AuthorizedWorkflow and use execute_authorized_activity() for early failure if headers are missing.

The integration runs entirely in your workers (Rust tenuo_core in-process). Vault, AWS Secrets Manager, GCP Secret Manager, or env vars supply holder keys. Tenuo Cloud is optional: managed issuance, keys, audit, and root rotation for teams that want that, not a requirement to enforce warrants.

Key Features:

• Activity-level authorization: Each activity execution is authorized against warrant constraints
• Proof-of-Possession (PoP): Ed25519 signature verification for every warranted activity (requires trusted_roots on the worker config)
• Warrant propagation: Warrants flow through workflow headers automatically
• Child workflow delegation: Attenuate warrants when spawning child workflows via tenuo_execute_child_workflow()
• Delegation chain verification: Full chain-of-trust validation back to trusted roots
• Signal & update guards: Restrict which signals and updates a workflow accepts
• Nexus header propagation: Warrant context flows through Nexus operations
• PoP replay protection: Default in-process dedup plus PoP time-window verification (clock skew); optional PopDedupStore for fleet-wide dedup (Security considerations)
• Continue-as-new support: Warrant headers survive workflow continuation
• Fail-closed: Missing or invalid warrants block execution by default
• Secure key management: Private keys NEVER transmitted in headers - resolved from Vault/KMS/Secret Manager
• Enterprise key resolvers: VaultKeyResolver, AWSSecretsManagerKeyResolver, GCPSecretManagerKeyResolver, CompositeKeyResolver

Temporal ensures your workflows survive failures. Tenuo ensures every activity your workflow dispatches is authorized against the warrant the issuer approved. Together they give you durable execution with cryptographic least privilege.

Security posture in brief: Activity execution is fail-closed by default: missing or invalid warrants are denied without changes to activity definitions. Each dispatch includes a Proof-of-Possession signature binding tool name and arguments to the holder key. Enforcement is in-process (no Tenuo network hop at verify time); private keys stay in your infrastructure. See Security considerations for threat model, PoP windows, revocation, retries, and dedup.

Installation

uv pip install "tenuo[temporal]"

Requires a Temporal cluster (local temporal server start-dev or production).

Temporal plugin (TenuoTemporalPlugin)

For Temporal’s plugin model (AI Partner Ecosystem–style registration), use tenuo.TenuoTemporalPlugin: a SimplePlugin that wires client interceptors, worker interceptors (TenuoPlugin), and a sandboxed workflow runner with tenuo / tenuo_core passthrough (required for PyO3).

from temporalio.client import Client
from temporalio.worker import Worker

from tenuo import SigningKey
from tenuo.temporal import TenuoPluginConfig, EnvKeyResolver
from tenuo.temporal_plugin import TenuoTemporalPlugin

control = SigningKey.generate()
plugin = TenuoTemporalPlugin(
    TenuoPluginConfig(
        key_resolver=EnvKeyResolver(),
        trusted_roots=[control.public_key],
    )
)

client = await Client.connect("localhost:7233", plugins=[plugin])
worker = Worker(client, task_queue="my-queue", workflows=[...], activities=[...])

Pass the plugin on Client.connect(..., plugins=[plugin]) only: workers created from that client merge client plugins, so you should not duplicate the same plugin on Worker(..., plugins=[...]) unless the client was created without plugins.

Manual setup (interceptors=[...] + workflow_runner=SandboxedWorkflowRunner(...)) remains supported; for passthrough without the plugin, use ensure_tenuo_workflow_runner from tenuo.temporal_plugin (or from tenuo.temporal import ensure_tenuo_workflow_runner).

Runnable examples

These scripts under tenuo-python/examples/temporal/ are the fastest path from “I get the model” to a working worker + client. Run temporal server start-dev in one terminal, then cd tenuo-python/examples/temporal and run the Python file in another. See the examples README for prerequisites such as boto3 or tenuo[temporal,mcp].

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: worker's KeyResolver (Vault, etc.)
          or Tenuo Cloud
Used to: mint warrants               Used to: sign PoP on each activity dispatch
If compromised: rotate trusted root  If compromised: rotate key_id + re-issue 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.

Onboarding checklist

Follow this order the first time you integrate Tenuo with Temporal. The Quick Start and Configuration sections below go deeper; the tenuo.temporal module docstring has a Troubleshooting section for common failures.

1. Install: uv pip install "tenuo[temporal]" (see Installation). This installs tenuo_core, a compiled Rust extension. Prebuilt wheels are available for common platforms; if yours isn’t covered, build from source with maturin develop in tenuo-python.

2. Temporal server: Run a dev server (e.g. temporal server start-dev). See examples/temporal README.

3. Keys for PoP: Proof-of-Possession (PoP) is how each activity dispatch proves it was authorised by the workflow that holds the warrant. It needs two Ed25519 key-pairs:
   • Issuer key (control_key): signs warrants; stays with policy owners (Vault, KMS, CI, or Tenuo Cloud). Never on the worker. Tenuo Cloud can hold and rotate the issuer key if you use it.
   • Holder key (agent_key): signs a short-lived challenge on each execute_activity() call, binding the exact tool and arguments. Lives on the worker via a KeyResolver.

   Create both with SigningKey.generate(). Expose the holder key to the worker via KeyResolver (development: EnvKeyResolver + TENUO_KEY_<key_id>: see Development: Environment Variables).

4. Mint a warrant: Use Warrant.mint_builder() (or Warrant.issue) so capabilities match your activity names and argument constraints (e.g. path=Subpath(...)). In production, warrants are typically minted by Tenuo Cloud on behalf of your workflows, separating authorization policy from application code and giving your security team control over what gets issued without touching workflow definitions.

5. Worker config: Worker(..., interceptors=[TenuoPlugin(TenuoPluginConfig(...))]) with:
   • key_resolver: resolves key_id from headers to the holder signing key. If you use EnvKeyResolver, call resolver.preload_keys(["<key_id>", ...]) for every holder key_id before you construct the worker. PoP signing runs inside the workflow sandbox and calls resolve_sync(); the sandbox blocks os.environ as non-deterministic, so keys must be cached first.
   • trusted_roots: required (e.g. [control_key.public_key], or set tenuo.configure(trusted_roots=[...]) before constructing the config)
   • activity_fns: required when the warrant uses named field constraints and you call workflow.execute_activity() without a reliable activity function reference; use the same callables as Worker(activities=[...])
   • strict_mode=True: recommended when using named constraints with transparent execute_activity (fail-fast instead of only logging)

6. Workflow sandbox passthrough (required): Use SandboxedWorkflowRunner with SandboxRestrictions.default.with_passthrough_modules("tenuo", "tenuo_core"). Omitting this causes ImportError: PyO3 modules may only be initialized once....

7. Child workflows: For any child that must run under Tenuo, use tenuo_execute_child_workflow() only. The SDK’s workflow.execute_child_workflow() does not propagate warrant headers; children started that way have no authorization context.

8. Client: Attach TenuoClientInterceptor to Client.connect(..., interceptors=[...]). Bind headers before start with execute_workflow_authorized(...) (best under concurrency) or set_headers_for_workflow(workflow_id, tenuo_headers(warrant, key_id)) then execute_workflow.

9. Run the samples: See the Runnable examples table (demo.py first, then cloud_iam_layering.py, multi_warrant.py, delegation.py, and optionally temporal_mcp_layering.py).

10. Verify with tests: Without a Temporal server: cd tenuo-python && pytest tests/e2e/test_temporal_e2e.py. With the in-process Temporal test server (CI-style): pytest tests/e2e/test_temporal_live.py tests/e2e/test_temporal_replay.py -m temporal_live (see the temporal-integration job in .github/workflows/ci.yml).

Quick Start

Basic Workflow Protection

Development quick start: EnvKeyResolver reads holder keys from environment variables (no Vault). For production, use VaultKeyResolver, AWSSecretsManagerKeyResolver, or GCPSecretManagerKeyResolver (Key Management).

This snippet uses AuthorizedWorkflow and execute_authorized_activity() so a missing warrant fails at workflow start. The same warrant and interceptor setup also supports plain workflow.execute_activity() with no workflow base class; then set activity_fns when your warrant uses named field constraints (registry section). See demo.py for both styles.

from datetime import timedelta
from pathlib import Path
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.common import RetryPolicy
from temporalio.worker import Worker

from tenuo import SigningKey, Warrant
from tenuo_core import Subpath
from tenuo.temporal import (
    AuthorizedWorkflow,
    TenuoPlugin,
    TenuoPluginConfig,
    TenuoClientInterceptor,
    EnvKeyResolver,
    execute_workflow_authorized,
    tenuo_headers,
)

# Define protected activities (no Tenuo-specific code needed)
@activity.defn
async def read_file(path: str) -> str:
    return Path(path).read_text()

@activity.defn
async def write_file(path: str, content: str) -> str:
    Path(path).write_text(content)
    return f"Wrote {len(content)} bytes"

# Define workflow: AuthorizedWorkflow fails fast if warrant headers are missing
@workflow.defn
class DataProcessingWorkflow(AuthorizedWorkflow):
    @workflow.run
    async def run(self, input_path: str, output_path: str) -> str:
        # Automatic PoP signature generation via self.execute_authorized_activity
        data = await self.execute_authorized_activity(
            read_file,
            args=[input_path],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=1),
        )

        processed = data.upper()
        
        await self.execute_authorized_activity(
            write_file,
            args=[output_path, processed],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=1),
        )

        return f"Processed {len(data)} bytes"

# Setup
async def main():
    # Generate issuer (control) key and holder (agent) key.
    # In production: control_key belongs to your authorization team (stored in Vault/KMS,
    # used only to mint warrants). agent_key belongs to this worker process (stored in
    # the worker's KeyResolver, used to sign PoP on each activity dispatch).
    control_key = SigningKey.generate()
    agent_key = SigningKey.generate()

    # EnvKeyResolver reads TENUO_KEY_<key_id> from environment.
    # Register the agent key before starting the worker:
    import os, base64
    os.environ["TENUO_KEY_agent-key-1"] = base64.b64encode(bytes(agent_key.secret_key_bytes())).decode()

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

    # Issue warrant using the builder API
    warrant = (
        Warrant.mint_builder()
        .holder(agent_key.public_key)
        .capability("read_file", path=Subpath("/data/input"))
        .capability("write_file", path=Subpath("/data/output"))
        .ttl(3600)
        .mint(control_key)
    )
    # In production, warrants are typically minted by Tenuo Cloud on behalf of your
    # workflows: scoped to the specific task, delegated to the correct holder key,
    # and managed without embedding issuance logic in application code. This separates
    # authorization policy from application code and gives your security team
    # visibility and control over what gets issued.

    # Configure worker interceptor with full PoP verification
    key_resolver = EnvKeyResolver()
    key_resolver.preload_keys(["agent-key-1"])  # cache holder key before Worker(...); sandbox PoP cannot use os.environ

    interceptor = TenuoPlugin(
        TenuoPluginConfig(
            key_resolver=key_resolver,
            on_denial="raise",
            trusted_roots=[control_key.public_key],  # required: Authorizer + PoP
            strict_mode=True,  # optional: fail-fast on ambiguous PoP with named constraints
        )
    )

    # Start worker with interceptor and sandbox passthrough
    from temporalio.worker.workflow_sandbox import (
        SandboxedWorkflowRunner, SandboxRestrictions,
    )

    async with Worker(
        client,
        task_queue="data-processing",
        workflows=[DataProcessingWorkflow],
        activities=[read_file, write_file],
        interceptors=[interceptor],
        workflow_runner=SandboxedWorkflowRunner(
            restrictions=SandboxRestrictions.default.with_passthrough_modules(
                "tenuo", "tenuo_core",  # Required for PoP signing
            )
        ),
    ):
        result = await execute_workflow_authorized(
            client=client,
            client_interceptor=client_interceptor,
            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",
        )

Security: Private keys are NEVER transmitted in headers. Only the key_id is sent. Workers use KeyResolver to fetch the actual signing key from secure storage (Vault, AWS Secrets Manager, GCP Secret Manager, etc.).

What happens:

sequenceDiagram
    participant C as Client
    participant T as Temporal
    participant WW as Workflow Worker
    participant KR as KeyResolver
    participant AW as Activity Worker

    C->>T: execute_workflow(headers: warrant + key_id)
    T->>WW: workflow task
    WW->>KR: resolve(key_id)
    KR-->>WW: signing_key  ← never transmitted
    Note over WW: PoP = sign(warrant, tool, args, workflow.now())
    WW->>AW: activity headers (warrant + PoP)
    Note over AW: verify warrant chain → trusted_roots
    Note over AW: verify PoP signature + constraints
    AW->>AW: execute activity (authorized)

The signing key is resolved on the worker and never leaves it. PoP is computed at schedule time (binding exact tool and args), then verified on the activity worker before execution.

This works in both single-process demos and distributed deployments where client and worker run in separate processes.

Required passthrough modules: tenuo and tenuo_core must be configured as passthrough modules in Temporal’s workflow sandbox. Without this, the worker starts and connects normally but every workflow execution fails on its first task. No activities are ever scheduled. The worker continues polling and appears healthy from the outside; the failure only appears as workflow task errors in Temporal Web.

Error you will see in Temporal Web:

ImportError: PyO3 modules may only be initialized once per interpreter process

Why this is necessary: In Temporal’s Python SDK today, OpenTelemetry’s TracingInterceptor is pure Python and does not need workflow sandbox passthrough. Tenuo is different: it computes the Proof-of-Possession signature inside the workflow sandbox at execute_activity() dispatch time so the exact tool name and arguments are committed at dispatch. Moving signing outside the sandbox would break that guarantee. tenuo_core is a PyO3 Rust extension that cannot be re-imported in a sub-interpreter, so both tenuo and tenuo_core must be declared passthrough. (If OTel’s interceptor ever pulled native code into the sandbox, the same passthrough consideration would apply—today it does not.) See Sandbox passthrough explained for the full failure sequence and diagnostic steps.

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 holder key is resolved on the worker via a production KeyResolver (Vault, AWS Secrets Manager, or GCP Secret Manager), not EnvKeyResolver.
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 — Every workflow worker must use 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.

Then walk the Onboarding checklist once for your repo.

API Ergonomics

Use one of these patterns based on your needs:

Recommended (default): execute_workflow_authorized(...)

The safest way to start authorized workflows. Binds headers to a specific workflow ID and executes immediately.

result = await execute_workflow_authorized(
    client=client,
    client_interceptor=client_interceptor,
    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",
)

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), the integration contract is:

If any required part is missing, execution fails closed.

Configuration

Key Management (REQUIRED)

Security Requirement: Tenuo NEVER transmits private keys in headers. Workers must be configured with a KeyResolver to fetch signing keys from secure storage.

Production: Vault

from tenuo.temporal import VaultKeyResolver, TenuoPluginConfig

resolver = VaultKeyResolver(
    url="https://vault.company.com:8200",
    path_template="production/tenuo/{key_id}",  # e.g., "production/tenuo/agent-2024"
    token=None,  # Uses VAULT_TOKEN env var
    mount="secret",  # KV secrets engine mount
    cache_ttl=300,  # Cache keys for 5 minutes
)

config = TenuoPluginConfig(
    key_resolver=resolver,  # REQUIRED
    trusted_roots=[root_key.public_key],
    strict_mode=True,  # optional: fail-fast on ambiguous PoP with named constraints
)

Store keys in Vault:

# Store signing key in Vault (field name must match what VaultKeyResolver reads)
vault kv put secret/production/tenuo/agent-2024 \
  key=@signing_key.b64  # base64-encode the 32-byte key file first

Production: AWS Secrets Manager

from tenuo.temporal import AWSSecretsManagerKeyResolver

resolver = AWSSecretsManagerKeyResolver(
    secret_prefix="tenuo/keys/",  # e.g., "tenuo/keys/agent-2024"
    region_name="us-west-2",
    cache_ttl=300,
)

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[root_key.public_key],
    strict_mode=True,  # optional: fail-fast on ambiguous PoP with named constraints
)

Store keys in AWS:

# Store signing key in AWS Secrets Manager
aws secretsmanager create-secret \
  --name tenuo/keys/agent-2024 \
  --secret-binary fileb://signing_key.bin \
  --region us-west-2

Production: GCP Secret Manager

from tenuo.temporal import GCPSecretManagerKeyResolver

resolver = GCPSecretManagerKeyResolver(
    project_id="my-project",
    secret_prefix="tenuo-keys-",  # e.g., "tenuo-keys-agent-2024"
    cache_ttl=300,
)

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[root_key.public_key],
    strict_mode=True,  # optional: fail-fast on ambiguous PoP with named constraints
)

Store keys in GCP:

# Store signing key in GCP Secret Manager
gcloud secrets create tenuo-keys-agent-2024 \
  --data-file=signing_key.bin \
  --project=my-project

Development: Environment Variables

from tenuo.temporal import EnvKeyResolver

# DEVELOPMENT ONLY - DO NOT USE IN PRODUCTION
# A one-time WARNING is emitted at first key resolution unless TENUO_ENV looks like a dev
# environment (see below).
resolver = EnvKeyResolver(
    prefix="TENUO_KEY_",
    warn_in_production=True,  # Default; set False to suppress explicitly
)

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],  # required (or call tenuo.configure(trusted_roots=[...]) first)
    # strict_mode: optional: set True if you use named warrant constraints with transparent execute_activity
)

Set environment variable:

# Export base64-encoded signing key
export TENUO_KEY_agent1=$(cat signing_key.bin | base64)
# Suppress the production warning in local dev (any of: development, dev, test, testing, local):
export TENUO_ENV=development

What the warning looks like: On first resolve() when warn_in_production=True (default) and TENUO_ENV is not one of the dev values above, the worker logs (Python WARNING, logger name from tenuo.temporal):

EnvKeyResolver is designed for development and testing only. In production, use VaultKeyResolver, AWSSecretsManagerKeyResolver, or GCPSecretManagerKeyResolver to fetch keys from secure storage. Set TENUO_ENV=development to suppress this warning in local environments.

Before starting a worker that runs workflows with Tenuo, call resolver.preload_keys([...]) with the same key_id values you pass to tenuo_headers(...) / execute_workflow_authorized(..., key_id=...). That loads TENUO_KEY_* into an in-process cache so the sandbox can resolve holder keys without touching the environment.

Warning: EnvKeyResolver is for development only. In production, use Vault, AWS Secrets Manager, or GCP Secret Manager.

Composite Resolver (Fallback Chain)

from tenuo.temporal import CompositeKeyResolver, VaultKeyResolver, EnvKeyResolver

resolver = CompositeKeyResolver(
    resolvers=[
        VaultKeyResolver(url="https://vault.company.com"),  # Try Vault first
        EnvKeyResolver(),                                    # Fallback to env vars
    ],
    warn_on_fallback=True,  # Log a WARNING whenever a fallback resolver is used
)

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[root_key.public_key],
    strict_mode=True,  # optional: fail-fast on ambiguous PoP with named constraints
)

Tenuo Cloud alternative: If you prefer not to operate your own KMS or Vault deployment, Tenuo Cloud provides managed key issuance and rotation. Signing keys are created, scoped, and rotated in the Cloud dashboard; workers resolve them without any additional key infrastructure on your side.

Worker plugin config (TenuoPluginConfig)

from tenuo.temporal import TenuoPluginConfig

config = TenuoPluginConfig(
    key_resolver=EnvKeyResolver(),        # Required: key resolution strategy
    on_denial="raise",                    # "raise" | "log" | "skip"
    dry_run=False,                        # Shadow mode only; never for production
    trusted_roots=[control_key.public_key],  # Enables Authorizer + PoP verification
    strict_mode=True,                     # Fail-fast on ambiguous PoP when using 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"],       # Optional signal allowlist
    authorized_updates=["update_config"], # Optional update allowlist
)

Production hardening: Every Temporal worker must supply trusted_roots (or set them once via tenuo.configure(trusted_roots=[...])). Without them, TenuoPluginConfig raises ConfigurationError at construction time. Use strict_mode=True to fail fast when PoP signing would use positional args while the warrant has named field constraints.

Denial Handling

Control what happens when authorization fails:

# "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, not production)

Use dry_run=True to run in shadow mode while you validate policies. In this mode, authorization denials are recorded (audit/log), but activities are still executed.

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[root_key.public_key],
    dry_run=True,   # shadow mode for rollout validation only
    on_denial="raise",  # ignored for authorization denials while dry_run=True
)

Warning: dry_run=True disables enforcement for authorization denials. Use only in non-production environments.

Activity registry (activity_fns) and PoP argument names

Why this matters

Each activity call gets a Proof-of-Possession (PoP) signature over a canonical payload that includes the tool name and a sorted argument dictionary. Warrant field constraints (for example path=Subpath("/data") on read_file) are checked against that same dictionary. The keys in the dict must therefore match the Python parameter names of the activity (e.g. path), not generic placeholders.

When your workflow calls workflow.execute_activity(...), the outbound interceptor must build args_dict from the activity’s positional args tuple. It does that by resolving the activity function and using inspect.signature to map positions to names.

Resolution order (function reference)

The worker resolves the callable in this order:

1. input.fn: supplied by the Temporal Python SDK on some versions/paths when using the real execute_activity pipeline.
2. tenuo_execute_activity(...): Tenuo records the function reference for that call.
3. TenuoPluginConfig.activity_fns: explicit registry: activity type name (e.g. read_file) → the same function object you registered on the worker.
4. Fallback: arg0, arg1, …: used only when (1) through (3) are all unavailable.

Step (4) is correct when the warrant only allows the tool without per-field constraints (signing and verification both use arg0, …). Step (4) is wrong when the warrant has named constraints: verification expects path, but signing used arg0, so PoP/constraint checks do not line up with the warrant.

What Tenuo does at runtime

If the interceptor would sign with only arg0/arg1/… and the warrant has non-empty field constraints for that activity type, the worker:

• Logs a warning (default), telling you to set activity_fns or use tenuo_execute_activity.
• Raises TenuoContextError (fail-fast) when strict_mode=True on TenuoPluginConfig, so misconfigured production workers fail immediately instead of issuing bad PoP material.

What you should configure

Example (activity_fns aligned with the worker)

from temporalio.worker import Worker
from tenuo.temporal import TenuoPlugin, TenuoPluginConfig, EnvKeyResolver

activities = [read_file, write_file]

interceptor = TenuoPlugin(
    TenuoPluginConfig(
        key_resolver=EnvKeyResolver(),
        trusted_roots=[control_key.public_key],
        strict_mode=True,
        activity_fns=activities,  # same objects as Worker(activities=...)
    )
)

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

For full narrative and troubleshooting text, see the module docstring in tenuo.temporal (Activity registry (activity_fns) and PoP argument names and Troubleshooting).

Sandbox passthrough explained

Temporal’s Python SDK re-imports all workflow code in an isolated sandbox on every worker task to enforce replay determinism. Modules declared as passthrough are shared from the host process instead of being re-imported.

Why Tenuo needs it: Tenuo signs the Proof-of-Possession challenge inside the workflow sandbox at execute_activity() dispatch time, committing the exact tool and arguments the workflow authorised, using the deterministic workflow.now() clock. This lets the activity worker detect argument tampering in transit. Because this signing uses tenuo_core (a PyO3 Rust extension), and PyO3 cannot be re-initialised in a sub-interpreter, both modules must be declared passthrough.

If you omit the passthrough, the failure is not at startup. The worker connects and polls normally:

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

Compatibility

Feature availability may depend on SDK surface area. Core activity/workflow authorization and child-workflow delegation are primary supported paths.

Proof-of-Possession

With trusted_roots in place (required for workers), Tenuo enforces PoP verification for all activity executions that carry a warrant. The challenge is a CBOR-serialized tuple of (warrant_id, tool, sorted_args, window_ts) signed with the holder’s Ed25519 key.

Two patterns for PoP

AuthorizedWorkflow (recommended) validates headers at workflow start and provides self.execute_authorized_activity():

@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 (multi-warrant workflows, per-stage delegation) where you need explicit control:

from datetime import timedelta
from temporalio.common import RetryPolicy
from tenuo.temporal import tenuo_execute_activity

@workflow.defn
class PipelineWorkflow:
    @workflow.run
    async def run(self, path: str) -> str:
        return await tenuo_execute_activity(
            read_file,
            args=[path],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

Both automatically sign PoP challenges; you never need to call warrant.sign() directly in Temporal workflows.

PoP Challenge Format

The PoP signature is computed deterministically by the Rust core:

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

In Python, this is a single call. Only needed for custom tooling outside of Temporal; tenuo_execute_activity() handles it automatically inside workflows:

import time
pop_signature = warrant.sign(signing_key, "read_file", {"path": "/data/file.txt"}, int(time.time()))
# Returns 64 raw bytes; verifier accepts multiple 30s-aligned windows (default: 5 windows, ~±60s skew band)

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    retry_pop_max_windows=120,   # 120 × 30s = 3600s: covers up to 1 hour of retries
)

Activity '...' is a retry (attempt=2). If this fails with PopVerificationError,
set TenuoPluginConfig.retry_pop_max_windows to accommodate Temporal's retry time offset.

Child Workflow Delegation

Important: workflow.execute_child_workflow() does not propagate Tenuo warrant headers. Use tenuo_execute_child_workflow() instead: it attenuates the parent warrant and injects headers via the outbound interceptor. A plain SDK child has no warrant context and no Tenuo authorization.

Attenuate warrants when spawning child workflows with tenuo_execute_child_workflow():

from tenuo.temporal import tenuo_execute_child_workflow

@workflow.defn
class ParentWorkflow:
    @workflow.run
    async def run(self) -> str:
        # Parent has: read_file + write_file

        # Child gets only read_file with reduced TTL
        result = await tenuo_execute_child_workflow(
            ChildWorkflow.run,
            tools=["read_file"],   # Subset of parent tools
            ttl_seconds=60,        # Shorter than parent
            args=["/data/input"],
            id=f"child-{workflow.info().workflow_id}",
            task_queue=workflow.info().task_queue,
        )
        return result

The wrapper calls attenuated_headers() internally and injects the attenuated warrant via the outbound workflow interceptor. Temporal’s execute_child_workflow() does not accept a headers kwarg directly.

Delegation Chain Verification

When warrants are attenuated, the full delegation chain is propagated via the x-tenuo-warrant-chain header. The activity interceptor calls Authorizer.check_chain() to verify every link in the chain back to a trusted root, ensuring no intermediate warrant was forged or widened.

Signal & Update Authorization

Control which signals and workflow updates are allowed:

config = TenuoPluginConfig(
    key_resolver=EnvKeyResolver(),
    on_denial="raise",
    trusted_roots=[control_key.public_key],
    authorized_signals=["approve", "reject"],     # Only these signals allowed
    authorized_updates=["update_config"],          # Only these updates allowed
)

Unrecognized signals raise TemporalConstraintViolation. Unrecognized updates are rejected at the validator stage before the handler runs. When set to None (default), all signals and updates pass through for backward compatibility.

Nexus Operation Headers

When starting Nexus operations from a Tenuo-protected workflow, the outbound interceptor automatically propagates warrant headers to the Nexus service. Headers are base64-encoded into Nexus’s string-based header format.

PoP Replay Protection

The activity interceptor runs dedup after PoP verification. The default store is InMemoryPopDedupStore: a thread-safe, process-local map. Each dedup key includes the warrant’s logical facet (via dedup_key(tool, args)), workflow_id, workflow_run_id, and activity_id. Temporal retries with attempt > 1 bypass dedup so legitimate redelivery is not blocked. The default store evicts periodically (every 60 seconds) and caps size at 10,000 entries.

Memory footprint: Each entry is a string key (on the order of 100 to 180 bytes for typical Temporal IDs) mapped to a float timestamp. With Python dict overhead, the store uses roughly 3 to 4 MB at the 10,000-entry cap, which is small for typical workers. To lower the cap, implement a custom PopDedupStore with a smaller internal size limit.

Pluggable backend: Set TenuoPluginConfig.pop_dedup_store to a shared implementation of PopDedupStore when you need fleet-wide replay suppression (see Security considerations).

Distributed deployments: Without a shared PopDedupStore, dedup state is not replicated across worker pods. Treat that as an explicit trade-off: cryptographic PoP windows still bound signature age, but duplicate first attempts on different replicas within the dedup TTL are not suppressed by the default store.

Decorators

@tool() - Activity-to-Tool Mapping

Map activity names to different tool names in warrants:

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

By default, TenuoPlugin blocks all activities used as local activities unless they are explicitly opted out. Use @unprotected to declare that a specific activity intentionally runs without a warrant, typically for internal, non-sensitive operations (config lookups, metrics, logging helpers). Every @unprotected activity is a deliberate hole in your authorization perimeter; document the reason at the call site.

from tenuo.temporal import unprotected

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

# Can be used as local activity (bypasses worker interceptor)
await workflow.execute_local_activity(
    get_config_value,
    args=["database_url"],
)

Activities not marked @unprotected that are called via workflow.execute_local_activity() will raise LocalActivityError at runtime. Intentional fail-closed behaviour: local activities bypass the inbound interceptor, so Tenuo cannot enforce warrant constraints without this explicit opt-out.

workflow_grant() - Scoped In-Workflow Grants

Issue a narrowed warrant for a single tool within a running workflow:

from tenuo.temporal import workflow_grant

@workflow.defn
class MyWorkflow(AuthorizedWorkflow):
    @workflow.run
    async def run(self, path: str) -> str:
        # Issue a 60-second, read-only grant for exactly one tool,
        # narrower than the workflow's own warrant.
        file_warrant = workflow_grant(
            "read_file",
            constraints={"path": path},  # Must be keys the parent already has
            ttl_seconds=60,
        )
        # file_warrant is a Warrant object; use it in a custom activity call
        # or pass it via tenuo_headers() to an external service.
        ...

workflow_grant() is useful when you need to hand off a scoped credential to a sub-process or external call. Constraint keys in constraints must already exist in the parent warrant; introducing new keys raises TemporalConstraintViolation.

Audit Events

Every authorization decision emits a structured TemporalAuditEvent with full context. This supports per-action audit trails aligned with common control frameworks (for example SOC 2 CC6.8 logical access logging, PCI DSS 10.2 privileged access audit trails, and HIPAA Security Rule audit controls for PHI).

Each event captures:

• Who: warrant_id, workflow_id, workflow_run_id: identifies the agent and the specific execution
• What: tool, arguments (redacted by default), warrant_capabilities: the specific action and scope
• When: timestamp (UTC)
• Decision: ALLOW or DENY, with denial_reason and constraint_violated for denials
• Context: workflow_type, activity_id, task_queue, tenuo_version

from tenuo.temporal import TemporalAuditEvent

def on_audit(event: TemporalAuditEvent):
    # Structured logging: forward to Splunk, Datadog, CloudWatch, etc.
    record = event.to_dict()
    audit_logger.info(record)

    # Or handle allow/deny separately
    if event.decision == "ALLOW":
        logger.info(
            f"Allowed: {event.tool} in {event.workflow_type} "
            f"(warrant: {event.warrant_id})"
        )
    else:
        logger.warning(
            f"Denied: {event.tool} in {event.workflow_type} - "
            f"{event.denial_reason}"
        )

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    audit_callback=on_audit,
    audit_allow=True,   # Log allowed actions (recommended for compliance)
    audit_deny=True,    # Log denied actions
    redact_args_in_logs=True,  # Replace argument values with "[REDACTED]" in logs
)

TemporalAuditEvent.to_dict() returns a plain dict suitable for structured log ingestion. Set redact_args_in_logs=True (default) to prevent argument values from appearing in log pipelines when processing sensitive data.

At scale: Tenuo Cloud indexes receipts across all your workflows and provides a queryable audit trail: which agent invoked which tool, under which warrant, through which delegation chain, at what time. For compliance-sensitive deployments this replaces custom audit log infrastructure and makes it straightforward to answer “what did this agent do last Tuesday?” across an entire fleet.

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 with prometheus_client (requires uv pip install prometheus-client):
# - tenuo_temporal_activities_authorized_total{tool, workflow_type}
# - tenuo_temporal_activities_denied_total{tool, reason, workflow_type}
# - tenuo_temporal_authorization_latency_seconds_bucket{tool}
# Expose via your app's existing /metrics endpoint or prometheus_client.start_http_server()

Suggested Alerts

For production integration monitoring, alert on:

• sustained increase in *_activities_denied_total
• spikes in POP_VERIFICATION_FAILED / replay-related denials
• key resolver failures (KEY_NOT_FOUND, resolver exceptions)
• sudden drop in authorized activity volume

Security Model

See Security considerations for the full threat model, trust boundaries, PoP time windows, dedup semantics, and root rotation. The Failure Semantics table maps exceptions to integration handling.

Exceptions

All 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"
)

Failure Semantics (Integrator View)

When dry_run=True, these authorization denials are converted to audit/log-only signals and execution continues. Use this only for staging validation and rollout analysis, never as a production steady state.

Troubleshooting

For the full module-level troubleshooting entries, see the tenuo.temporal module docstring. The table below covers the most common integration errors.

Best Practices

1. Production keys: VaultKeyResolver, AWSSecretsManagerKeyResolver, or GCPSecretManagerKeyResolver; not EnvKeyResolver. Optional: Tenuo Cloud for managed issuance and rotation.
2. Sandbox passthrough: always with_passthrough_modules("tenuo", "tenuo_core") on workflow workers.
3. Named warrant fields: set activity_fns to match Worker(activities=[...]), or call tenuo_execute_activity(), or use strict_mode=True to catch arg0-style PoP early.
4. AuthorizedWorkflow: use when you want missing warrant headers to fail at workflow start; otherwise plain workflow.execute_activity() is fine if activity_fns is set when needed.
5. Child workflows: only tenuo_execute_child_workflow(); never raw execute_child_workflow() for authorized children.
6. Start path: prefer execute_workflow_authorized(...) under concurrent clients.
7. Audit: wire audit_callback and forward TemporalAuditEvent.to_dict() to your log or SIEM; keep redact_args_in_logs=True for sensitive args.
8. @unprotected: limit to internal, low-risk local activities; each one is an explicit bypass. Tenuo Cloud can aggregate unprotected vs authorized volume if you use it.
9. TTLs: keep warrants short for sensitive work; combine with trusted_roots_provider and a short refresh interval for fast issuer rotation.
10. Never dry_run=True in production (staging only).
11. Multi-tenant fleets: separate TenuoPluginConfig per tenant (or task queue) with distinct trusted_roots so issuers do not cross tenants.
12. Scale: shared PopDedupStore across replicas; retry_pop_max_windows when Temporal retries exceed the default PoP window.

Migration Path (from plain Temporal)

1. Worker: add TenuoPlugin and sandbox passthrough (tenuo, tenuo_core).
2. Client: start workflows with execute_workflow_authorized(...) or set_headers_for_workflow + execute_workflow.
3. Children: replace workflow.execute_child_workflow() with tenuo_execute_child_workflow() wherever the child needs a warrant.
4. Rollout: one task queue or tenant first, then expand.

Rollback

If needed, you can temporarily route traffic to an unprotected queue while preserving the existing workflow code. Keep this as an operational fallback, not a steady-state mode.

Integration QA Coverage

Current test coverage is split across:

• tenuo-python/tests/e2e/test_temporal_live.py and test_temporal_replay.py: in-process Temporal test server, serialization/header propagation, delegation, continue-as-new, replay (pytest -m temporal_live, as in CI)
• tenuo-python/tests/e2e/test_temporal_e2e.py: mocked Temporal infrastructure with real Tenuo objects: outbound/inbound interceptors, PoP, constraints, child headers, Nexus header path

Together these cover protocol-level behavior and worker/client integration without requiring a manually started Temporal cluster for the bulk of the suite.

Examples

The runnable scripts are summarized in the Runnable examples table near the top of this page.

Per-stage pipeline (from delegation.py)

Each pipeline stage gets its own tightly-scoped warrant:

# 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=Pattern("*"))
    .ttl(600)
    .mint(control_key)
)

# Switch warrant between stages
client_interceptor.set_headers_for_workflow(
    "ingest-run-001",
    tenuo_headers(ingest_warrant, "ingest"),
)
data = await client.execute_workflow(IngestWorkflow.run, id="ingest-run-001", ...)

client_interceptor.set_headers_for_workflow(
    "transform-run-001",
    tenuo_headers(transform_warrant, "transform"),
)
await client.execute_workflow(TransformWorkflow.run, id="transform-run-001", ...)

Comparison with Other Integrations

Temporal, Tenuo, and observability tools operate at different layers of an agentic system. None substitutes for the other.

Temporal makes agentic workflows durable. Tenuo authorizes each activity before it runs. Activity definitions stay unchanged; you add client headers, worker interceptors, and (when needed) activity_fns or AuthorizedWorkflow as in Overview.

For reference, Tenuo’s authorization model across integrations:

The Temporal integration is the only path where PoP is mandatory: durable workflows have the strongest threat model (history replay, distributed execution, long-lived warrant exposure).

More Information

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