Temporal Workflow Engine

Why We Replaced the v0.2 Task Queue

The v0.2 in-memory task queue was sufficient for single-session, short-duration workflows. It was not sufficient for workflows spanning multiple hours or days (e.g., waiting for human approval), fault recovery (if the orchestration service restarted, in-flight workflows were lost), or parallel branching (the queue was linear; branching required custom logic in every workflow definition).

Temporal solves all three. It provides durable execution state, automatic activity retry with configurable backoff, workflow versioning, and a Web UI for workflow visibility.

Key Temporal Capabilities Now Available

Deployment Configuration

# agento-kernel-config.yaml
temporal:
  mode: cloud         # or "self-hosted"
  endpoint: "your-namespace.tmprl.cloud:7233"
  namespace: "agento-production"
  tls: true
  api_key: ${TEMPORAL_API_KEY}

Temporal is deployed as a sidecar to the Agento Kernel. The default configuration uses Temporal Cloud. Self-hosted Temporal (Docker or Kubernetes, Server v1.24+) is supported from v0.3.1.

Skill Registry

The skill registry is a versioned catalogue of agent capabilities. Each "skill" is a self-contained unit of agent behaviour: a prompt template, a set of allowed tools, an output schema, and an execution config. Skills can be published (available to all workflows), scoped (restricted to specific projects or teams), or versioned (pinned in a workflow definition to prevent silent behaviour changes on updates).

from agento.skills import SkillRegistry

registry = SkillRegistry()

registry.register(
    skill_id="bess_commissioning_retrieval",
    version="1.2.0",
    description="Retrieves BESS commissioning test data from ACC",
    allowed_connectors=["acc", "sharepoint"],
    prompt_template="skills/bess_retrieval.jinja2",
    output_schema=BESSTestDataSchema,
    rbac_scope="engineering.commissioning"
)

# Invoke from a workflow activity
@activity.defn
async def retrieve_test_data(input):
    skill = registry.get("bess_commissioning_retrieval",
                         version="1.2.0")
    return await skill.execute(input)

Skills can be updated without restarting the Kernel service. Running workflows continue on their pinned version; new executions pick up the latest (or a specified version if pinned in the workflow definition).

RBAC Audit Layer

Every agent action in v0.3 produces a structured audit event written to the Agento Audit API and any configured sink integrations:

{
  "event_id": "evt_01HVKM3X9XQPZ8R7N4T",
  "timestamp": "2026-04-15T09:23:41.842Z",
  "workflow_id": "bess-commissioning-wf-2026-04-15",
  "activity_id": "retrieve_acc_test_data",
  "agent_id": "retrieval-agent-v1",
  "action": "document_read",
  "resource": {
    "connector": "acc",
    "project_id": "proj_abc123",
    "document_id": "doc_xyz789",
    "document_name": "BESS-TR-001-TestSheet-Rev2.pdf"
  },
  "rbac_scope": "engineering.commissioning",
  "user_context": {
    "initiated_by": "user_j_morrison",
    "on_behalf_of": "project_alpha"
  },
  "result": "success",
  "latency_ms": 342
}

Permissions are enforced at three levels:

Audit log sinks: Datadog (GA), CloudWatch (GA), Splunk (beta, see known limitations), and generic webhook.

Migration Guide: v0.2 → v0.3

Task Queue → Temporal

Before (v0.2):

@agento.task
async def my_workflow(input: WorkflowInput):
    result_a = await task_queue.run(step_a, input)
    result_b = await task_queue.run(step_b, result_a)
    return result_b

After (v0.3):

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, input: WorkflowInput):
        result_a = await workflow.execute_activity(
            step_a, args=[input],
            start_to_close_timeout=timedelta(minutes=5)
        )
        result_b = await workflow.execute_activity(
            step_b, args=[result_a],
            start_to_close_timeout=timedelta(minutes=5)
        )
        return result_b

Breaking Changes

Known Limitations in v0.3

Coming in v0.4