# Temporal Integration

Source: https://tenuo.ai/temporal

## What is Temporal?

[Temporal](https://temporal.io/) is a platform for **durable execution**: you write **Workflows**, deterministic orchestration code whose progress is replayed from event history and survives process restarts, retries, and long waits, and **Activities**, the non-deterministic work such as tool calls, HTTP requests, or model inference. The service assigns tasks from **task queues** to **Workers**, persists workflow state, and gives you reliable, observable automation for multi-step AI agents without hand-rolling sagas or bespoke recovery logic.

For a guided introduction, see [Understanding Temporal](https://learn.temporal.io/getting_started/).

## How Tenuo fits

Access control in Temporal typically relies on worker identity or task queue tokens, which grant the same permissions to every workflow running on that worker. As agents take on more consequential actions, you need finer control: which tools each agent may use for a given task, with what arguments, and on whose authority.

Tenuo adds that task-scoped authorization layer. A signed warrant travels with each workflow and is verified by the worker interceptor before every Activity runs. Agents are **constrained by design**: an agent can only execute what its warrant permits, and every action is cryptographically attributable to the entity that authorized it. Activity definitions require no changes, and verification runs in-process with no external service dependency.

For cross-namespace workflows, Tenuo also supports
[Temporal Nexus authorization](./temporal-nexus.md). Nexus lets one team expose
a durable service contract through a named Endpoint; Tenuo carries delegated
warrants across that Endpoint so the handler can verify the exact operation,
input, holder key, expiry, approvals, and revocation state instead of relying
only on namespace-level access. For customer-facing scenarios, see
[Tenuo for Temporal Nexus](./temporal-nexus-use-cases.md).

## Prerequisites

- Familiarity with Temporal Workflows, Activities, and Workers ([What is Temporal](#what-is-temporal) above, or [Temporal learning resources](https://learn.temporal.io/getting_started/)).
- A running Temporal cluster (local `temporal server start-dev` or Temporal Cloud).
- Python 3.10+ (inherited from `temporalio>=1.23.0`, which provides `SimplePlugin`).

If you want the shortest copy-paste path first, start with the
[Temporal Quickstart](./temporal-quickstart.md). It runs one authorized
workflow locally and shows both an allowed Activity and a denied Activity.

## Install

```bash
uv pip install "tenuo[temporal]"
```

This installs `temporalio>=1.23.0` and `tenuo_core`, a compiled Rust extension with prebuilt wheels for common platforms.

## Configure Workers to use Tenuo

Add the `TenuoTemporalPlugin` to your Client. The plugin wires client interceptors, worker interceptors, and the workflow sandbox runner in one step.

```python
from temporalio.client import Client
from temporalio.worker import Worker
from tenuo import SigningKey
from tenuo.temporal import TenuoTemporalPlugin, TenuoPluginConfig, EnvKeyResolver

# For local development — generate a key pair:
control_key = SigningKey.generate()
issuer_public_key = control_key.public_key

plugin = TenuoTemporalPlugin(
    TenuoPluginConfig(
        key_resolver=EnvKeyResolver(),
        trusted_roots=[issuer_public_key],
    )
)

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

`TenuoPluginConfig` requires two things:

- **`trusted_roots`** — public keys of warrant issuers (for verification on the activity worker)
- **`key_resolver` or `signing_key`** — the holder key the workflow worker uses to sign PoP. The call runs inside the workflow sandbox, so it has to return a key already in memory. `signing_key=` does that. `EnvKeyResolver` does too, after preload, and is for development. `VaultKeyResolver`, `AWSSecretsManagerKeyResolver`, and `GCPSecretManagerKeyResolver` start a thread the sandbox blocks, including when their cache is full.

`EnvKeyResolver` maps `key_id` to environment variables using the convention **`TENUO_KEY_<key_id>`** with **base64-encoded** signing key bytes:

| `key_id` passed to warrant | Environment variable | Format |
|---|---|---|
| `"agent1"` | `TENUO_KEY_agent1` | Base64 or hex |
| `"my-service"` | `TENUO_KEY_my-service` | Base64 or hex |

```bash
# Generate and export a key for local development:
export TENUO_KEY_agent1=$(python -c "from tenuo import SigningKey; import base64; k=SigningKey.generate(); print(base64.b64encode(k.secret_key_bytes()).decode())")
```

`TenuoTemporalPlugin` automatically preloads all `TENUO_KEY_*` variables into an in-memory cache so that key resolution never touches `os.environ` inside the workflow sandbox. For production, pass `signing_key=` or a resolver whose `resolve_sync` returns a key already in memory. See the [reference](./temporal-reference.md#key-management-required). **[Tenuo Cloud](https://cloud.tenuo.ai)** handles key issuance, warrant minting, rotation, and audit for teams that prefer a managed control plane.

> **Important:** Pass the plugin on `Client.connect(plugins=[plugin])` only. Workers created from that client automatically merge client plugins — do not duplicate.

> **About the names.** Two public classes have similar names on purpose — they are *not* the same:
>
> | Class | Type | When to use |
> |---|---|---|
> | `tenuo.temporal_plugin.TenuoTemporalPlugin` | Temporal SDK **`SimplePlugin`** | **Default.** Pass to `Client.connect(plugins=[...])`. Wires the client interceptor, worker interceptor, and sandboxed workflow runner in one step. |
> | `tenuo.temporal.TenuoWorkerInterceptor` | Temporal SDK **`WorkerInterceptor`** | Advanced only. Use when you are hand-composing your own `Plugin` / `SimplePlugin` and just want Tenuo's authorization interceptor. |

## Define activities and workflows

Activity definitions stay unchanged — no Tenuo imports needed:

```python
from pathlib import Path
from temporalio import activity, workflow
from datetime import timedelta

@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"
```

Use `AuthorizedWorkflow` to fail fast if warrant headers are missing, or use plain `workflow.execute_activity()` — the interceptor handles authorization either way:

```python
from tenuo.temporal import AuthorizedWorkflow

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

## Start an authorized workflow

Pass a warrant and key ID when starting a workflow. The warrant defines what the agent is allowed to do — your control plane or policy layer mints it.

```python
from tenuo.temporal import execute_workflow_authorized

result = await execute_workflow_authorized(
    client=client,
    workflow_run_fn=MyWorkflow.run,
    workflow_id="process-001",
    warrant=warrant,
    key_id="agent1",
    args=["/data/input/report.txt"],
    task_queue="my-queue",
)
```

For long-running workflows where you need a handle to signal or query later, use `start_workflow_authorized()`:

```python
from tenuo.temporal import start_workflow_authorized

handle = await start_workflow_authorized(
    client=client,
    workflow_run_fn=ApprovalWorkflow.run,
    workflow_id="approval-001",
    warrant=warrant,
    key_id="agent1",
    args=[request_data],
    task_queue="my-queue",
)

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

### Use a fresh warrant for one Activity

Long-running workflows can keep a longer-lived issuer warrant as their base
authority while issuing a short-lived execution warrant for a consequential
dispatch. The override is scoped to this call and is safe when workflow tasks
schedule Activities concurrently:

```python
from tenuo.temporal import (
    current_key_id, tenuo_execute_activity, workflow_issue_execution,
)

execution_warrant = await workflow_issue_execution(
    "transfer_funds",
    constraints={"account": account, "amount": amount},
    ttl_seconds=60,
)
await tenuo_execute_activity(
    transfer_funds,
    args=[account, amount],
    warrant=execution_warrant,
    key_id=current_key_id(),
    start_to_close_timeout=timedelta(seconds=30),
)
```

### Scheduled workflows

Use `create_scheduled_workflow_with_warrant()` so each scheduled workflow start
carries Tenuo headers—not warrant material in a memo:

```python
from tenuo.temporal import create_scheduled_workflow_with_warrant

await create_scheduled_workflow_with_warrant(
    client,
    "nightly-report",
    ReportWorkflow.run,
    issuer_warrant,
    "report-agent",
    schedule_spec,
    workflow_args=[tenant_id],
    task_queue="reports",
)
```

The Schedule action reuses this warrant on every trigger. Its TTL must cover
the Schedule lifetime. For unbounded recurring Schedules, use a longer-lived,
narrowly scoped issuer warrant and mint short-lived per-Activity execution
warrants as shown above.

## Capability constraints

Warrants use a **closed-world (zero-trust) model**: every argument the activity receives must be declared in the capability, even if unconstrained. Use `Wildcard()` for arguments that can take any value:

```python
from tenuo import Warrant, SigningKey, Subpath, UrlSafe, Wildcard

warrant = (
    Warrant.mint_builder()
    .holder(agent_key.public_key)
    .capability("read_file", path=Subpath("/data"))       # path must be under /data
    .capability("search", query=Wildcard())               # query can be anything
    .capability("fetch_url",
        url=UrlSafe(allow_schemes=["https"],
                    allow_domains=["api.example.com"],
                    block_private=True),
        timeout=Wildcard(),                               # unconstrained but declared
    )
    .ttl(3600)
    .mint(control_key)
)
```

If an activity argument is not listed in the capability, the interceptor rejects the call with `TemporalConstraintViolation` — even if the value would otherwise be valid. This prevents accidental exposure of undeclared parameters.

> **Common mistake:** listing only the constrained fields. If your activity has parameters `path` and `encoding`, the capability needs both — e.g. `.capability("read_file", path=Subpath("/data"), encoding=Wildcard())`.

See [Temporal Integration Reference — Constraint Types](./temporal-reference.md#constraint-types-for-ai-agent-workflows) for the full list of constraint types (`Subpath`, `UrlSafe`, `Exact`, `Pattern`, `Range`, `AnyOf`, etc.).

## How it works

```mermaid
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_id, tool, sorted_args, window_ts)
    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.

## Child workflow delegation

Attenuate warrants when spawning child workflows so children get least-privilege access:

```python
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 a shorter TTL.
        return 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,
        )
```

> **Important:** `workflow.execute_child_workflow()` does **not** propagate warrant headers. Always use `tenuo_execute_child_workflow()` for authorized children.

## Cross-namespace calls with Temporal Nexus

When workflows in one Namespace need to invoke capabilities owned by another
team or Namespace, use [Temporal Nexus authorization](./temporal-nexus.md).
The Nexus helpers mirror the Activity helper shape:

```python
await tenuo_execute_nexus_operation(
    nexus_client,
    PaymentService.refund,
    RefundInput(order_id="ord_123", amount_cents=5000),
    warrant=refund_warrant,
    key_id="agent1",
)
```

On the handler side, `@tenuo_nexus_operation(config, endpoint="payments-prod")`
or `verify_nexus_operation(...)` verifies the signed warrant, proof of
possession, endpoint/service/operation binding, input constraints, approvals,
expiry, and revocation before user code runs.

For workflow-backed Nexus operations, use the explicit envelope/bootstrap
helpers documented in the Nexus guide. They carry verified or attenuated
authority into the backing workflow without exposing handler Namespace
credentials to the caller Namespace.

When Nexus becomes a shared enterprise surface across teams, accounts, or
organizations, the remaining challenge is operational control: issuer
ownership, trusted-root distribution, revocation rollout, approvals, key
rotation, and audit search across the fleet. Tenuo can be self-hosted for those
pieces; teams that want that control plane operated centrally can use a managed
Tenuo deployment. See [Tenuo for Temporal Nexus](./temporal-nexus-use-cases.md#enterprise-scale-control-across-teams),
or [schedule a demo](https://tenuo.ai/#talk) to map this onto your
Temporal topology.

## Security

**Fail-closed by default.** Missing or invalid warrants block execution. Each activity dispatch includes a Proof-of-Possession (PoP) signature binding the tool name and arguments to the holder key. Enforcement is in-process (no Tenuo network hop at verify time).

**Private keys never leave your infrastructure.** Only the `key_id` and warrant material travel in Temporal headers. The workflow worker signs with a holder key it already has in memory (`signing_key=`, or a `KeyResolver` whose `resolve_sync` does not leave the process). `EnvKeyResolver` is the development form of that, after preload. No private key material is transmitted to the Temporal cluster or any Tenuo endpoint.

**Warrant chain verification.** When warrants are attenuated (e.g. for child workflows), the full delegation chain is validated back to trusted roots, ensuring no intermediate warrant was forged or widened.

| Check | Missing / invalid | Default behavior |
|-------|-------------------|------------------|
| Warrant header | Missing | Denied (`require_warrant=True`) |
| Warrant expired | Expired | `WarrantExpired` |
| Tool / constraints | Args outside scope | `TemporalConstraintViolation` |
| PoP signature | Missing or invalid | `PopVerificationError` |

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

**Trust boundaries:**

| Component | Role |
|-----------|------|
| **Issuer / control plane** | Mints warrants; public keys configured as `trusted_roots` on workers |
| **Temporal service** | Schedules tasks and carries headers; Tenuo does not replace Temporal's own security |
| **Workflow workers** | Sign PoP with a holder key already in memory; sandbox passthrough required for `tenuo_core` |
| **Activity workers** | Verify warrants, PoP, and constraints before running activities |

For the full threat model, PoP time windows, replay protection, root rotation, and revocation, see [Temporal Integration Reference](./temporal-reference.md#security-considerations).

## Activity summaries in the Temporal Web UI

The plugin enriches every authorized activity with a human-readable summary in the Temporal Web UI's Event History:

| Activity kind | Summary in UI |
|---|---|
| User activity | `[tenuo.TenuoTemporalPlugin] read_file` |
| With user summary | `[tenuo.TenuoTemporalPlugin] read_file: monthly report` |
| Internal warrant mint | `[tenuo.TenuoTemporalPlugin] attenuate(read_file, list_directory)` |

```python
from tenuo.temporal import tenuo_execute_activity

await tenuo_execute_activity(
    read_file,
    args=["/data/report.txt"],
    start_to_close_timeout=timedelta(seconds=30),
    summary="monthly sales report",
)
```

## Runnable examples

These scripts under [`tenuo-python/examples/temporal/`](https://github.com/tenuo-ai/tenuo/tree/main/tenuo-python/examples/temporal) are the fastest path to a working demo. Run `temporal server start-dev` in one terminal, then run the Python file in another.

| Example | What it shows |
|---------|---------------|
| [`demo.py`](https://github.com/tenuo-ai/tenuo/tree/main/tenuo-python/examples/temporal/demo.py) | **Start here.** Transparent `execute_activity()` and `AuthorizedWorkflow` in one place |
| [`delegation.py`](https://github.com/tenuo-ai/tenuo/tree/main/tenuo-python/examples/temporal/delegation.py) | Per-stage pipeline with least-privilege warrants |
| [`multi_warrant.py`](https://github.com/tenuo-ai/tenuo/tree/main/tenuo-python/examples/temporal/multi_warrant.py) | Multi-tenant isolation: same workflow, different warrants |
| [`cloud_iam_layering.py`](https://github.com/tenuo-ai/tenuo/tree/main/tenuo-python/examples/temporal/cloud_iam_layering.py) | Temporal + MCP + S3 with per-tenant prefixes |
| [`temporal_mcp_layering.py`](https://github.com/tenuo-ai/tenuo/tree/main/tenuo-python/examples/temporal/temporal_mcp_layering.py) | Temporal + MCP over stdio |

## Next steps

- **[Temporal Quickstart](./temporal-quickstart.md)** — copy-paste local
  workflow with one allowed Activity and one denied Activity.
- **[Temporal Integration Reference](./temporal-reference.md)** — production checklist, key management (Vault, AWS, GCP), sandbox details, PoP mechanics, configuration reference, constraint types, troubleshooting, and the full threat model.
- **[Tenuo for Temporal Nexus](./temporal-nexus-use-cases.md)** — practical cross-team, multi-hop, and cross-organization examples.
- **[Temporal Nexus Authorization](./temporal-nexus.md)** — cross-namespace authorization for Nexus operations and workflow-backed handlers.
- [Tenuo Core Concepts](./concepts.md)
- [Security Model](./security.md)
- [Example Code](https://github.com/tenuo-ai/tenuo/tree/main/tenuo-python/examples/temporal)