Merged
Conversation
Adds `temporalio.contrib.tenuo`, a SimplePlugin integration for
Tenuo warrant-based authorization in Temporal workflows.
The plugin (`TenuoPlugin`) wires client interceptors, worker
interceptors, and workflow sandbox passthrough in a single line:
from temporalio.contrib.tenuo import TenuoPlugin
plugin = TenuoPlugin(config)
client = await Client.connect("localhost:7233", plugins=[plugin])
Key design decisions:
- Thin adapter: only TenuoPlugin, TENUO_PLUGIN_NAME, and
ensure_tenuo_workflow_runner are exported from the contrib module.
All other types (TenuoPluginConfig, EnvKeyResolver, etc.) are
imported directly from tenuo.temporal.
- No private imports: all tenuo.temporal internals used by the plugin
are exposed through public lazy-loaded names.
- No re-exports of external package types, matching the pattern
established by openai_agents and other contrib modules.
Files:
- temporalio/contrib/tenuo/__init__.py — public API (3 exports)
- temporalio/contrib/tenuo/_plugin.py — TenuoPlugin SimplePlugin subclass
- temporalio/contrib/tenuo/README.md — multi-agent delegation example
- tests/contrib/tenuo/test_tenuo.py — unit + live integration tests
- tests/contrib/tenuo/test_tenuo_replay.py — record-and-replay tests
- pyproject.toml — tenuo optional dependency
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds
temporalio.contrib.tenuo, aSimplePluginthat wires Tenuo warrant-based authorization into Temporal workflows. Agents (workflows) carry signed warrants specifying which tools (activities) they can call and with what argument constraints. Sub-agents (child workflows) receive attenuated warrants — capabilities can only shrink, never expand.TenuoPlugin— registers client interceptor (warrant header injection), worker interceptors (PoP signing + authorization verification), and sandbox passthrough for thetenuonative extension.TenuoPlugin,TENUO_PLUGIN_NAME,ensure_tenuo_workflow_runner). All other types are imported fromtenuo.temporal, matching the pattern established byopenai_agentsand other contrib modules.tenuo.temporalinternals used by the plugin are exposed through public lazy-loaded names.Files
temporalio/contrib/tenuo/__init__.pytemporalio/contrib/tenuo/_plugin.pyTenuoPluginSimplePlugin subclasstemporalio/contrib/tenuo/README.mdtests/contrib/tenuo/test_tenuo.pytests/contrib/tenuo/test_tenuo_replay.pypyproject.tomltenuooptional dependencyReplay safety
Replay determinism is verified at two levels:
workflow.now()(nottime.time()), nodatetime.now(), noos.urandom/random/uuid4, notime.sleep, nothreading.Thread.fetch_history(), and a freshTenuoPlugininstance replays viaReplayer. Tests cover single-tool and multi-tool (sequential PoP ordering) scenarios.Integration tests
test_authorized_activity_succeeds— full warrant → PoP → authorization flowtest_start_workflow_authorized—start_workflow_authorizedreturns a handletest_unauthorized_activity_is_non_retryable— unauthorized tool call producesWorkflowFailureErrorwithApplicationError(non_retryable=True)test_duplicate_registration_raises— same plugin instance on two workers raisesRuntimeErrorTest plan
pytest tests/contrib/tenuo/test_tenuo.py -v— unit + integration tests passpytest tests/contrib/tenuo/test_tenuo_replay.py -v— replay determinism tests passruff check temporalio/contrib/tenuo/ tests/contrib/tenuo/— no lint errors