Temporal Integration Reference

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: 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.

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 — 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.

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. 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}",
    token=None,        # Uses VAULT_TOKEN env var
    mount="secret",
    cache_ttl=300,
)

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

Store keys in Vault:

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

Production: AWS Secrets Manager

from tenuo.temporal import AWSSecretsManagerKeyResolver

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

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

Store keys in AWS:

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-",
    cache_ttl=300,
)

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

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. In production, use Vault, AWS Secrets Manager, or GCP Secret Manager.

KeyResolver and the workflow sandbox

PoP signing runs inside _TenuoWorkflowOutboundInterceptor.start_activity, which is inside the workflow sandbox. That means the sandbox determinism and I/O restrictions apply to whatever the resolver’s resolve_sync does on each call. Pure-memory resolvers are safe; I/O-bound resolvers must be preloaded at worker startup and must return from their in-memory cache inside the sandbox.

A sandbox violation surfaces as temporalio.worker.workflow_sandbox.RestrictedWorkflowAccessError, wrapped by our interceptor as a non-retryable TenuoContextError. If you see this on a live workflow, the fix is almost always “preload before the sandbox activates” or “extend cache_ttl” — not “turn off the sandbox”.

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, VaultKeyResolver, EnvKeyResolver

resolver = CompositeKeyResolver(
    resolvers=[
        VaultKeyResolver(url="https://vault.company.com"),
        EnvKeyResolver(),
    ],
    warn_on_fallback=True,
)

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(
    key_resolver=EnvKeyResolver(),
    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"],            # Optional signal allowlist
    authorized_updates=["update_config"],      # Optional update allowlist
)

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.

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.

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 headers from the original ACTIVITY_TASK_SCHEDULED event. The first-attempt verifier uses pop_max_windows=5 (~±60 s); retries use the laxer retry_pop_max_windows, which defaults to 40 (±20 min) — sized against Temporal’s default retry policy (initial_interval=1s, backoff_coefficient=2, max_interval=100s) so ten retries at ~13 min still verify.

config = TenuoPluginConfig(
    key_resolver=resolver,
    trusted_roots=[issuer_public_key],
    retry_pop_max_windows=120,   # 120 × 30s = 3600s
)

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_grant(...) to mint a narrower per-phase warrant inside the workflow (requires the parent warrant to be an issuer).
  • 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 injects the Tenuo headers into the event payload, so every Tenuo-protected call adds per-event overhead to Temporal’s event history (capped at 50,000 events / 2 MB by default, up to ~50 MB absolute depending on server config).

Approximate size per activity, with gzip compression enabled (the default):

Worked example. A single workflow that dispatches 200 activities with a 3-hop warrant: 200 × (~1.2 KB warrant + ~100 B misc + ~90 B PoP) ≈ 280 KB of Tenuo overhead in history. Well under the 2 MB limit, but non-trivial for archival replay costs and for workflows that also carry large user payloads.

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.

Planned for v0.2 — warrant_hash + worker-side LRU cache. Activities would carry only the PoP signature plus a warrant_hash reference; the receiving worker resolves the full warrant from an in-process LRU (falling back to re-fetching from the source, e.g. Tenuo Cloud). This keeps per-event headers flat (~200 B) regardless of chain depth. Tracking: temporal/warrant-cache-reference.

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(
    key_resolver=EnvKeyResolver(),
    on_denial="raise",
    trusted_roots=[control_key.public_key],
    authorized_signals=["approve", "reject"],
    authorized_updates=["update_config"],
)

Unrecognized signals raise TemporalConstraintViolation. When set to None (default), all signals and updates pass through.

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.

set_activity_approvals() - Pre-Supply Multisig Approvals

Call from a workflow before workflow.execute_activity() when the warrant has guards that require approval. The outbound interceptor attaches the approvals to the next activity dispatch.

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),
)

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,
)

Constraint keys must already exist in the parent warrant.

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

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