Home

Zelos User Manual v0.9.0

Open Multi-Agent Orchestration Runtime — The Runtime that executes, coordinates, and governs intelligent agents.

Table of Contents

1. Quick Start

Installation

# Install from PyPI
pip install zelos-runtime

# Verify (core has zero external dependencies)
python3 -c "from zelos.runtime import ZelosRuntime; print('✅ Zelos Runtime OK')"

Or clone the repository for development:

git clone https://github.com/AI-Zelos/zelos.git
cd zelos/
pip install -e ".[dev]"

5-Minute Quick Start (Copy-Paste Ready)

# Save this as zelos_quickstart.py and run: python3 zelos_quickstart.py
from zelos.runtime import ZelosRuntime

# 1. Create and start Runtime
rt = ZelosRuntime()
rt.start()

# 2. Register an agent that generates Python code
rt.add_agent(
    name="Coder",
    entrypoint="my_agent:CodeAgent",
    capabilities=[{"name": "code-generation.python", "version": "1.0.0"}],
)

# 3. Submit a goal — the Runtime handles everything else
goal = rt.submit_goal(description="Write a function to reverse a string in Python")
print(f"Goal submitted: {goal['goal_id'][:8]}... → {goal['status']}")

# 4. Check status
status = rt.get_goal_status(goal["goal_id"])
print(f"Progress: {status.get('progress', {}).get('percent_complete', 0)}%")

# 5. Check health
health = rt.get_health()
print(f"Runtime: {health['status']}, version={health['version']}")

rt.shutdown()

Running this prints:

Goal submitted: a1b2c3d4... → planned
Progress: 0%
Runtime: healthy, version=0.9.0

Your First Goal

rt.add_agent( name="CodeAgent", entrypoint="my_agents.coder:CodingAgent", capabilities=[ {"name": "code-generation.python", "version": "1.0.0"} ], )

── 3. Submit a Goal ──

goal = rt.submit_goal( description="Write a function that checks if a string is a palindrome", priority="high", ) print(f"Goal: {goal['goal_id']} → {goal['status']}")

── 4. Wait for completion ──

result = rt.wait_for_goal(goal["goal_id"], timeout_seconds=30) print(f"Result: {result['status']} — {result['progress']['percent_complete']:.0f}%")

── 5. Shutdown ──

rt.shutdown()

### With zelos.yaml

```yaml
# zelos.yaml
runtime:
  api:
    host: "127.0.0.1"
    port: 9876

plugins:
  - id: "llm-planner"
    type: "planner"
    entrypoint: "zelos.planner.LLMPlanner"
    config:
      provider: "openai"
      model: "gpt-4o"
      api_key: "${OPENAI_API_KEY}"

rt = ZelosRuntime.from_yaml("zelos.yaml")
# Everything loaded from config — Planner, Verifier, Policy, Memory
rt.start()

1a. Dashboard

Zelos 内置一个 Web Dashboard，通过 HTTP Adapter 自动提供服务。无需安装任何额外依赖。

启动 Runtime + Dashboard

# 方式 1：一行命令启动（推荐）
python3 -c "
from zelos.runtime import ZelosRuntime
from zelos.http_adapter import HTTPAdapter

rt = ZelosRuntime({'plugins': []})
rt.start()

adapter = HTTPAdapter(rt, host='127.0.0.1', port=9876)
adapter.start()
print(f'✅ Dashboard: {adapter.url}')
# 保持运行，Ctrl+C 停止
import time
try:
    while True: time.sleep(1)
except KeyboardInterrupt:
    adapter.stop(); rt.shutdown()
"

# 方式 2：使用启动脚本
python3 start.py

访问 Dashboard

启动后在浏览器打开 http://127.0.0.1:9876/，即可看到：

• Dashboard — Runtime 状态、活跃 Goals、活跃 Agents、事件数、租户数
• Goals — 所有 Goal 的状态和进度条
• Agents — 所有 Agent 的连接状态
• Audit Log — 审计事件（时间、操作者、操作、资源、结果）
• Tenants — 租户列表和配额使用情况
• Cluster — 集群节点状态和领导者信息
• Health & Metrics — 原始 JSON 数据
• API Reference — 完整 REST API 文档

页面每 5 秒自动刷新，无需手动操作。

REST API 端点

Dashboard 调用的所有 API 也可直接用 curl 访问：

curl http://127.0.0.1:9876/api/v1/health
curl http://127.0.0.1:9876/api/v1/metrics
curl http://127.0.0.1:9876/api/v1/goals
curl http://127.0.0.1:9876/api/v1/agents
curl http://127.0.0.1:9876/api/v1/audit
curl http://127.0.0.1:9876/api/v1/tenants
curl http://127.0.0.1:9876/api/v1/cluster
curl http://127.0.0.1:9876/api/v1/approvals/pending

HTTPS / mTLS

from zelos.security import TLSConfig

tls = TLSConfig(
    cert_file="/etc/zelos/certs/server.crt",
    key_file="/etc/zelos/certs/server.key",
    ca_file="/etc/zelos/certs/ca.crt",
    require_client_cert=True,
)

adapter = HTTPAdapter(rt, host='0.0.0.0', port=8443, tls_config=tls)
adapter.start()
# Dashboard: https://your-host:8443/

2. Core Concepts

Architecture

Goal → Planner → ExecutionPlan → TaskGraph → Scheduler → ExecutionEngine → Agent
  │                                                                              │
  └────────────────────────── Event Bus (everything) ────────────────────────────┘
                                   │
                    Memory / Policy / Verifier / Observability

Design Principles

Agent Lifecycle

Register → Connect → Heartbeat → Dispatch → Execute → Return → Repeat
   │                                              │
   └────────────── Hot-Join / Hot-Leave ──────────┘

3. Runtime

Module: zelos.runtime | Class: ZelosRuntime

The central entry point. Owns the complete lifecycle of the Kernel and all Agents.

Basic Usage

from zelos.runtime import ZelosRuntime

# From dict config
rt = ZelosRuntime({"plugins": [...]})

# From YAML file
rt = ZelosRuntime.from_yaml("zelos.yaml")

# Lifecycle
rt.start()       # Initialize all components + start agents
# ... use the runtime ...
rt.shutdown()    # Graceful shutdown

Agent Management

# Register (hot-join if Runtime is running)
agent_id = rt.add_agent(
    name="ClaudeCode-v2",
    entrypoint="my_agents.coder:CodingAgent",
    capabilities=[
        {"name": "code-generation.python", "version": "1.0.0"},
        {"name": "code-generation.typescript", "version": "1.0.0"},
    ],
    max_concurrent_tasks=5,
    auth_context={"role": "admin", "tenant_id": "eng"},
)

# List agents (tenant-filtered)
agents = rt.list_agents(auth_context={"role": "operator", "tenant_id": "eng"})

# Get agent details
agent = rt.get_agent("ClaudeCode-v2")

# Remove (hot-leave)
rt.remove_agent("ClaudeCode-v2", auth_context={"role": "admin"})

Goal Lifecycle

# Submit — Planner decomposes into Task DAG
goal = rt.submit_goal(
    description="Build a REST API for user management",
    budget=50.0,
    priority="high",
    auth_context={"role": "admin", "tenant_id": "eng"},
    require_approval=True,          # Phase 3: HITL gate
    approvers=["alice", "bob"],     # Phase 3: multi-approver
)

# Check progress
status = rt.get_goal_status(goal["goal_id"])

# Block until done
result = rt.wait_for_goal(goal["goal_id"], timeout_seconds=600)

# Cancel
rt.cancel_goal(goal["goal_id"], auth_context={"role": "admin"})

Phase 3: Advanced APIs

# ── Security ──
key = rt.generate_api_key("agent", "CI pipeline", ttl_seconds=86400,
                          auth_context={"role": "admin"})
audit = rt.get_audit_log(actor="alice", action="goal.submit")

# ── Dynamic Plan Modification ──
rt.modify_plan(plan_id, "add_task",
               task_id="extra-1", description="Additional validation",
               required_capability="code-review.security")

# ── Sub-Goal Spawning ──
sub = rt.spawn_sub_goal("task-123", "Investigate alternatives",
                        budget=25.0, num_tasks=3)

# ── HITL Approval ──
rt.approve_task("approval-abc123", "alice", "LGTM, ship it")
rt.reject_task("approval-abc123", "bob", "Needs more tests")

# ── Hot Reload ──
rt.reload_plugin("verifier", "zelos.verifier_v3:FullVerifier", "3.0.0")
rt.rollback_plugin("verifier", "2.0.0")
rt.set_upgrade_strategy("canary")

# ── Multi-tenancy ──
rt.register_tenant("fintech", "Fintech Division",
                   quotas={"max_goals": 100, "budget_per_goal": 5000})
usage = rt.get_tenant_usage()

# ── Distributed ──
is_leader = rt.is_leader()
cluster = rt.get_cluster_status()

API Reference

4. Event Bus

Module: zelos.event_bus | Classes: EventBus, Event, InMemoryEventStore

The central communication backbone. Every state change in Zelos is an Event. Components never call each other directly.

Key Concepts

from zelos.event_bus import EventBus, Event, HandlerResult

bus = EventBus(max_events=10000)

# ── Immutable Events ──
event = Event(
    event_id="evt-001",
    event_type="task.completed",
    source="execution-engine",
    timestamp=time.time(),
    correlation_id="goal-g-001",
    payload={"task_id": "t-1", "agent_id": "agent-1", "duration_ms": 1234},
)
bus.publish(event)

# ── Subscribe by exact type ──
def on_task_completed(event: Event):
    print(f"Task done: {event.payload['task_id']}")
    return HandlerResult.ACK

bus.subscribe("task.completed", on_task_completed)

# ── Subscribe by pattern ──
bus.subscribe_pattern("task.*", lambda e: print(f"Task event: {e.event_type}"))

# ── Subscribe by correlation ──
bus.subscribe_correlation("goal-g-001", lambda e: print(f"Goal event: {e}"))
# All events with correlation_id="goal-g-001" are delivered here

# ── Replay ──
count = bus.replay_from(0, lambda e: print(e.event_type))
# count: number of events replayed

Event Schema

Handler Results

Event Taxonomy

goal.submitted  goal.planned  goal.started  goal.completed  goal.failed  goal.cancelled
task.created  task.ready  task.assigned  task.started  task.completed  task.failed  task.timed_out
agent.registered  agent.connected  agent.heartbeating  agent.disconnected  agent.removed
artifact.created  artifact.verified
verification.passed  verification.failed
plugin.loaded  plugin.started  plugin.stopped  plugin.error

5. Capability Registry

Module: zelos.capability_registry | Classes: CapabilityRegistry, CapabilityEntry

The WHAT registry. Agents register their capabilities. The Runtime discovers providers automatically. Never dispatch by agent name.

Usage

from zelos.capability_registry import CapabilityRegistry

reg = CapabilityRegistry()

# ── Register capabilities ──
reg.register(
    agent_id="agent-1",
    agent_name="CodeAgent",
    capabilities=[
        {
            "name": "code-generation.python",
            "version": "1.0.0",
            "description": "Generate Python code from specifications",
            "input_schema": {"type": "object", "properties": {"spec": {"type": "string"}}},
            "output_schema": {"type": "object", "properties": {"code": {"type": "string"}}},
            "tags": ["production", "python", "soc2-compliant"],
        }
    ]
)

reg.mark_available("agent-1")

# ── Query by capability ──
providers = reg.find_providers_for("code-generation.python")
# → ["agent-1"]

# ── Query by name pattern ──
providers = reg.find_by_name_pattern("code-generation.*")
# → ["agent-1"]

# ── Query by tags (AND logic) ──
providers = reg.find_by_tags(["production", "python"])
# → ["agent-1"]

# ── Query by version constraint ──
providers = reg.find_by_version("code-generation.python", ">=1.0,<2.0")
# → providers with matching version

# ── Lifecycle ──
reg.mark_deprecated("agent-1")  # Stop routing new tasks
reg.mark_unavailable("agent-1") # Agent is down

6. Task Graph

Module: zelos.task_graph | Classes: TaskGraphEngine, Task, TaskStatus

Manages the Task state machine and DAG dependency resolution. Every Goal becomes a Plan, every Plan is a DAG of Tasks.

Task State Machine

CREATED → READY → ASSIGNED → STARTED → COMPLETED  (terminal)
  ↓        ↓         ↓           ↓
  ↓        ↓         ↓           ├── FAILED → READY (retry)
  ↓        ↓         ↓           └── TIMED_OUT → READY (retry)
  ↓        ↓         └── READY (reject)
  ↓        └── FAILED (escalation tier 3)
  └── CANCELLED (terminal)

Usage

from zelos.task_graph import Task, TaskStatus, TaskGraphEngine

tg = TaskGraphEngine()

# ── Create tasks ──
t1 = Task(task_id="t1", plan_id="plan-1",
          description="Design the architecture",
          required_capability="design.architecture",
          priority="high", timeout_ms=60000)
t2 = Task(task_id="t2", plan_id="plan-1",
          description="Code the implementation",
          required_capability="code-generation.python",
          dependencies=["t1"])  # t2 depends on t1 completing
t3 = Task(task_id="t3", plan_id="plan-1",
          description="Review the code",
          required_capability="code-review.security",
          dependencies=["t2"])

tg.add_task(t1)
tg.add_task(t2)
tg.add_task(t3)

# ── Evaluate dependencies ──
ready = tg.evaluate_all()
# t1 has no deps → READY immediately
# t2 waits for t1
# t3 waits for t2

# ── State transitions ──
tg.transition("t1", TaskStatus.ASSIGNED, agent_id="agent-1")
tg.transition("t1", TaskStatus.STARTED)
tg.transition("t1", TaskStatus.COMPLETED)
# Now t2 becomes READY automatically

ready = tg.evaluate_all()  # → ["t2"]

# ── DAG validation ──
tg.add_dependency("t3", "t1")  # ⚠️ Would create cycle t1→t2→t3→t1
# → ValueError: Adding edge t3→t1 would create a cycle

# ── Dynamic modification (Phase 3) ──
tg.add_task_dynamic(Task(task_id="t4", plan_id="plan-1",
                         description="Extra step",
                         required_capability="testing.e2e",
                         dependencies=["t3"]))
tg.remove_task("t4")  # Only unstarted tasks

Task Fields

7. Scheduler

Module: zelos.scheduler | Classes: Scheduler, ScoringStrategy, DefaultScoringStrategy

The 5-phase scheduling pipeline that matches READY tasks to available Agents.

The Pipeline

Phase 1: ORDER   → Sort READY tasks by priority > fan-out > age
Phase 2: FILTER  → Find agents matching capability + tags + constraints
Phase 3: SCORE   → Rank agents (delegated to ScoringStrategy plugin)
Phase 4: POLICY  → Apply policy decisions (Allow/Reject/Delay/Retry)
Phase 5: SELECT  → Pick the best agent and mark task ASSIGNED

Custom Scoring Strategy

from zelos.scheduler import ScoringStrategy, AgentCandidate, ScoredCandidate

class CostFirstScoring(ScoringStrategy):
    """Prioritize cheapest agents. Fail-safe with success_rate floor."""

    def score(self, task, candidates):
        results = []
        for c in candidates:
            # Quality floor
            if c.success_rate < 0.80:
                results.append(ScoredCandidate(c, score=0,
                    reason=f"Success rate {c.success_rate:.0%} below 80% threshold"))
                continue

            # Cost-first formula
            score = (
                (1.0 / (c.cost_per_call + 0.01)) * 0.6 +  # Cheaper = better
                c.success_rate * 0.3 +                      # Some quality
                (1.0 - c.current_load) * 0.1                # Less loaded = better
            )
            results.append(ScoredCandidate(c, score=score,
                reason=f"Cost=${c.cost_per_call:.3f} Success={c.success_rate:.0%}"))
        return sorted(results, key=lambda r: r.score, reverse=True)

Default Scoring Weights

DefaultScoringStrategy(weights={
    "success_rate": 0.30,       # Proven reliability
    "cost_efficiency": 0.20,    # Lower cost preferred
    "load_distribution": 0.15,  # Prefer less busy agents
    "latency": 0.15,            # Faster is better
    "availability": 0.10,       # Higher uptime
    "affinity": 0.05,           # Same agent for related tasks
    "recency": 0.05,            # Recently used (context warmth)
})

Retry Logic

Task FAILED or TIMED_OUT
  → attempt += 1
  → if attempt <= max_retries:
      backoff = base * 2^(attempt-1) + random(0, 500) ms
      → READY again
  → else:
      → FAILED (terminal)

8. Execution Engine

Module: zelos.execution_engine | Classes: ExecutionEngine, AgentState

Dispatches tasks to agents, monitors their lifecycle, enforces timeouts, and tracks heartbeats.

Agent State Machine

registered → connected → heartbeating → disconnected → shutdown
                │                                      │
                └── (after heartbeat recovers) ────────┘

Dispatch Flow

# ── Agent Side ──
engine.register_agent(agent_id="agent-1", agent_name="CodeAgent",
                      max_concurrent_tasks=5)

# Heartbeat every 30s
engine.heartbeat("agent-1")  # → "heartbeating"

# ── Runtime Side ──
engine.dispatch("t1", "agent-1")
# → Task status: ASSIGNED → STARTED
# → Agent.current_tasks: ["t1"]
# → InFlight: {task_id, agent_id, started_at, timeout_at}

# ── Agent returns result ──
engine.submit_result("t1", "agent-1", {
    "status": "completed",
    "artifact": {"content_type": "application/json", "content": {...}}
})
# → Task status: COMPLETED
# → Agent.current_tasks: [] → "idle"

# ── Error handling ──
engine.submit_result("t2", "agent-1", {
    "status": "failed",
    "error": {"code": "internal_error", "message": "Out of memory"}
})
# → Task status: FAILED → Scheduler retries if attempts remain

# ── Cancellation ──
engine.cancel_task("t3")
# → Task status: CANCELLED

Timeout Monitor

Background thread checks every second: - Tasks exceeding timeout_at → TIMED_OUT - Agents without heartbeat for 3× heartbeat_interval → disconnected

9. Planner

Module: zelos.planner | Class: LLMPlanner

Decomposes natural-language Goals into structured Execution Plans (Task DAGs) using an LLM.

Supported Providers

from zelos.planner import LLMPlanner

# OpenAI / DeepSeek / Ollama / vLLM
planner = LLMPlanner({
    "provider": "openai",
    "model": "deepseek-v4-flash",
    "api_key": "sk-...",
    "base_url": "https://api.deepseek.com/v1",
    "temperature": 0.3,
    "max_tokens": 4000,
})

# Anthropic
planner = LLMPlanner({
    "provider": "anthropic",
    "model": "claude-sonnet-5",
    "api_key": "sk-ant-...",
})

# Google Gemini
planner = LLMPlanner({
    "provider": "google",
    "model": "gemini-2.0-flash",
    "api_key": "...",
})

# Mock (for testing)
planner = LLMPlanner({"provider": "mock"})

Plan → Replan

# Initial plan
plan = planner.plan("Build a user authentication system")
# → ExecutionPlan with Task DAG:
#   t1: Design auth architecture [design.architecture]
#   t2: Implement backend [code-generation.python] ← depends on t1
#   t3: Implement frontend [code-generation.typescript] ← depends on t1
#   t4: Security review [code-review.security] ← depends on t2,t3
#   t5: Integration tests [testing.e2e] ← depends on t4

# Replan on failure
new_plan = planner.replan(
    goal_description="Build a user authentication system",
    current_plan=plan,
    failed_tasks=[
        {"task_id": "t2", "reason": "no agent provides capability: code-generation.python"}
    ]
)
# → New alternative tasks to replace failed ones

10. Verifier

Every Agent artifact must pass through a verification gate before flowing downstream.

Built-in Verifiers

from zelos.verifier import SchemaVerifier
from zelos.verifier_v2 import CodeReviewer, SecurityScanner, FactChecker

# SchemaVerifier — validates against expected_output_schema
from zelos.verifier import VerificationCriteria
schema_v = SchemaVerifier()
verdict = schema_v.verify(
    {"name": "hello", "version": 1},
    VerificationCriteria(expected_output_schema={"type": "object", "required": ["name"]}),
)
print(verdict.verdict)  # → "passed"

# CodeReviewer — finds dangerous patterns (eval, hardcoded secrets)
cr = CodeReviewer()
verdict2 = cr.verify(
    'eval(user_input)',
    VerificationCriteria(options={"language": "python"}),
)
print(verdict2.verdict)   # → "failed" (use of eval)
print(verdict2.issues[0])  # → details about the finding

# SecurityScanner — SQL injection, XSS, command injection, unsafe deserialization
ss = SecurityScanner()
verdict3 = ss.verify(
    '"SELECT * FROM users WHERE id=" + user_id',
    VerificationCriteria(),
)
print(verdict3.verdict)  # → "failed" (SQL injection)

VerificationGate — Chain Verifiers

from zelos.verifier import VerificationGate

gate = VerificationGate()
gate.add_verifier(SchemaVerifier())
gate.add_verifier(CodeReviewer())
gate.add_verifier(SecurityScanner())

result = gate.verify(artifact, criteria)
# All pass → accepted. First failure → short-circuit and reject.

11. Policy

Policy enforces Allow / Reject / Delay / Retry — it never changes business logic.

from zelos.policy import CostLimitPolicy, RateLimitPolicy, AllowlistPolicy, CompositePolicy

# Cost limit: reject if goal cumulative cost exceeds budget
cost = CostLimitPolicy({"max_cost_per_goal": 100})
print(cost.evaluate({"goal_id": "g1", "task_cost": 50}))   # → "allow"  (50 < 100)
cost.record_cost("g1", 50)
print(cost.evaluate({"goal_id": "g1", "task_cost": 60}))   # → "reject" (50+60 > 100)

# Rate limit: sliding window of task submissions per minute
rate = RateLimitPolicy({"max_tasks_per_minute": 60, "window_seconds": 60})
# rate.evaluate({"task_count": N}) → "allow" or "reject"

# Allowlist: only allow specified agents
allow = AllowlistPolicy({"allowlist_agents": ["agent-coder", "agent-reviewer"]})
print(allow.evaluate({"agent_id": "agent-coder"}))     # → "allow"
print(allow.evaluate({"agent_id": "unknown-agent"}))    # → "reject"

# Composite: chain them — first reject short-circuits
policy = CompositePolicy({"policies": [cost, rate]})
print(policy.evaluate({"goal_id": "g2", "task_cost": 10}))
# → "allow"

12. Memory

Runtime owns all memory. 6 isolated layers, each with independent lifecycle.

from zelos.memory import InMemoryMemoryProvider

mem = InMemoryMemoryProvider(max_entries_per_layer=5000, ttl_seconds=3600)

# 6 layers: session | project | user | knowledge | execution | skill
mem.store("session", "goal-1-context", {"goal": "Build landing page", "deadline": "2026-08-01"})
mem.store("project", "coding-style", {"language": "python", "formatter": "ruff"})

ctx = mem.retrieve("session", "goal-1-context")
print(ctx["goal"])  # → "Build landing page"

# TTL expiry — entries auto-expire after ttl_seconds
mem.store("execution", "temp-key", "cache-me", ttl=1)
# ...after 1 second...
result = mem.retrieve("execution", "temp-key")  # → None

# Context Assembly — automatically merge relevant layers before Task dispatch
from zelos.memory import ContextAssembler
assembler = ContextAssembler(mem)
context = assembler.assemble(task_id="task-1", goal_id="goal-1")
# → MemoryContext with session/project/user/knowledge/execution/skill entries

13. Observability

Structured logging, Prometheus metrics, and OpenTelemetry tracing.

from zelos.observability import StructuredLogger, MetricsCollector, Tracer

# ── Structured JSON logging ──
logger = StructuredLogger(level="info")
logger.info("task_dispatched", task_id="t1", agent="coder-1")
logger.warn("heartbeat_late", agent="reviewer-2", drift_ms=1500)

# ── Metrics (Counter / Gauge / Histogram) ──
metrics = MetricsCollector()
metrics.inc_counter("task_completed_total", labels={"capability": "code"})
metrics.set_gauge("agents_connected", 5)
metrics.observe_histogram("task_duration_ms", 340)

# Export Prometheus text format
print(metrics.to_prometheus_text())
# # HELP task_completed_total ... \n # TYPE task_completed_total counter\n ...

# ── Tracing (OpenTelemetry span tree) ──
tracer = Tracer()
span_id = tracer.start_span("goal.execute", attributes={"goal_id": "g1"})
tracer.add_event(span_id, "task_dispatched", {"task_id": "t1"})
tracer.end_span(span_id)

14. Protocol Adapters

All adapters translate external protocols to the same Runtime API. No business logic.

from zelos.protocol_adapters import (
    GRPCAdapter, WebSocketAdapter, MCPAdapter, A2AAdapter,
)

# gRPC — SubmitGoal, RegisterAgent, GetHealth mapped to protobuf service
grpc = GRPCAdapter(runtime, host="0.0.0.0", port=50051)
grpc.start()

# WebSocket — real-time event streaming
ws = WebSocketAdapter(runtime, host="0.0.0.0", port=9877)
ws.start()  # Clients connect and subscribe to event types

# MCP — Tool Registry for Agent tool discovery
mcp = MCPAdapter(runtime)
tool_list = mcp.list_tools()  # Agents discover available tools via JSON-RPC

# A2A — Agent Card generation for cross-Runtime interop
a2a = A2AAdapter(runtime)
card = a2a.generate_agent_card("code-agent")  # capabilities → A2A skills

15. Plugin Isolation

Run plugins in separate processes for fault isolation.

from zelos.plugin_isolation import SubProcessPlugin

plugin = SubProcessPlugin(
    plugin_id="isolated-verifier",
    entrypoint="my_verifier.py",  # Runs in its own Python process
)
plugin.start()      # Spawn subprocess, communicate via stdin/stdout JSON
plugin.health()     # → True if subprocess responds
plugin.execute({"task": "verify", "artifact": {...}})
plugin.shutdown()   # Graceful stop, force-kill after timeout

16. Storage

Pluggable storage backends — one config line to switch.

from zelos.storage import create_storage_backend

# InMemory (default, dev)
backend = create_storage_backend({"type": "memory", "max_events": 10000})

# PostgreSQL (production)
backend = create_storage_backend({
    "type": "postgresql",
    "url": "postgresql://user:pass@localhost:5432/zelos",
})

# Redis (high throughput)
backend = create_storage_backend({
    "type": "redis",
    "url": "redis://localhost:6379/0",
    "prefix": "zelos",
})

backend.connect()
backend.append("task-events", [{"event_id": "e1", "type": "task.created"}])
events = backend.read("task-events", 0, 100)
backend.set_state("goal-g1", {"status": "executing", "progress": 0.6})
state = backend.get_state("goal-g1")
backend.create_snapshot("goal-g1", events_position=5, state={"status": "done"})
backend.disconnect()

17. Security (Phase 3)

Module: zelos.security | Classes: AccessControl, AuditLogger, APIKeyManager, TLSConfig

Production-grade security infrastructure. Integrated into ZelosRuntime — all API methods enforce RBAC and log audit events.

Role-Based Access Control

from zelos.security import AccessControl

ac = AccessControl()

# ── Default Roles ──
# admin:    * (wildcard — everything)
# operator: goal.*, task.*, agent.read, plugin.*
# agent:    task.execute, agent.heartbeat, artifact.create
# viewer:   goal.read, task.read, agent.read, metrics.read

# ── Permission checks ──
ac.check("admin", "goal.submit")     # → True (wildcard)
ac.check("operator", "task.create")  # → True (task.* pattern)
ac.check("agent", "goal.submit")     # → False (no permission)
ac.check("viewer", "agent.delete")   # → False

# ── Custom roles ──
ac.add_role("sre", [
    "goal.read", "metrics.read", "agent.read",
    "plugin.configure", "audit.read",
])
ac.update_role("sre", add_permissions=["goal.cancel"])
ac.remove_role("sre")

Audit Logging

from zelos.security import AuditLogger

al = AuditLogger(max_events=100000)

# Every operation logs automatically through ZelosRuntime
# Manual logging:
al.log("admin", "goal.submit", "g-001",
       detail="Deploy v2.0 to production", result="success")

# ── Multi-field query ──
events = al.query(
    actor="admin",           # Who did it
    action="goal.submit",    # What they did
    resource="g-001",        # On what
    result="success",        # Outcome
    start_time=last_hour,    # When
    limit=100,
)

# ── Export ──
json_str = al.export_json()

API Key Management

from zelos.security import APIKeyManager

akm = APIKeyManager()

# Generate (plaintext shown once — hash stored)
key = akm.generate_key("agent", "CI/CD pipeline", ttl_seconds=86400)
# → "zelos_a1b2c3d4e5f6..."

# Validate → returns role info or None
info = akm.validate(key)
# → {"role": "agent", "description": "CI/CD pipeline", "expires_at": ...}

# Revoke
akm.revoke(key)
akm.validate(key)  # → None

# List (hashes only, never plaintext)
keys = akm.list_keys()

mTLS Configuration

from zelos.security import TLSConfig

tls = TLSConfig(
    cert_file="/etc/zelos/certs/server.crt",
    key_file="/etc/zelos/certs/server.key",
    ca_file="/etc/zelos/certs/ca.crt",
    require_client_cert=True,
    min_tls_version="TLSv1.3",
)

if tls.is_configured():
    # Apply to HTTP adapter / gRPC server
    pass

Integration with Runtime

rt = ZelosRuntime(...)
rt.start()

# All API calls enforce RBAC:
rt.submit_goal("...", auth_context={"role": "agent"})
# → {"status": "rejected", "reason": "Permission denied: role 'agent' cannot 'goal.submit'"}

# All state changes auto-logged:
audit = rt.get_audit_log(actor="alice")
# → [{event_id, timestamp, actor, action, resource, detail, result}, ...]

# API key auth:
key = rt.generate_api_key("admin", "my key", auth_context={"role": "admin"})
rt.submit_goal("...", auth_context={"api_key": key})

18. Multi-tenancy (Phase 3)

Module: zelos.multi_tenancy | Classes: Namespace, ResourceQuota, TenantManager

Hard isolation between tenants sharing a single Runtime. Integrated into ZelosRuntime.

Namespace & Quotas

from zelos.multi_tenancy import Namespace, ResourceQuota

# ── Quota configuration ──
quotas = ResourceQuota(
    max_goals=100,              # Active goals
    max_tasks=500,              # Active tasks
    max_agents=20,              # Registered agents
    budget_per_goal=5000.0,     # Max cost per goal
    max_concurrent_tasks=50,    # Parallel execution limit
    max_storage_mb=2048,        # Storage allocation
)

# ── Namespace isolation ──
ns = Namespace("fintech", "Fintech Division", quotas=quotas)

ns.add_goal("g-1")
ns.add_agent("agent-1")
ns.add_task("t-1")

ns.check_quota("goals")      # → True (1/100)
ns.check_quota("agents")     # → True (1/20)
ns.quotas.check_budget(3000) # → True (within $5000 limit)

Tenant Management

from zelos.multi_tenancy import TenantManager

tm = TenantManager()

# ── Register tenants ──
tm.register_tenant("acme-corp", "ACME Corporation",
    quotas=ResourceQuota(max_goals=500, budget_per_goal=10000),
    metadata={"org": "Engineering", "tier": "enterprise"})

tm.register_tenant("dev-team", "Development Team",
    quotas=ResourceQuota(max_goals=20, budget_per_goal=200),
    metadata={"org": "R&D", "tier": "standard"})

# ── Lifecycle ──
tm.deactivate_tenant("dev-team")
tm.is_active("dev-team")    # → False
tm.activate_tenant("dev-team")

# ── Cross-tenant isolation ──
ns_a = tm.get_namespace("acme-corp")
ns_b = tm.get_namespace("dev-team")
ns_a.add_goal("g-1")
# g-1 is invisible to dev-team namespace

# ── Usage report ──
report = tm.get_usage_report()
# → {"total_tenants": 2, "active_tenants": 2,
#    "tenants": {"acme-corp": {"goals": 1, ...}, ...}}

Runtime Integration

rt = ZelosRuntime({
    "multi_tenancy": {
        "enabled": True,
        "tenants": [
            {"id": "eng", "name": "Engineering",
             "quotas": {"max_goals": 100, "budget_per_goal": 5000}},
            {"id": "research", "name": "Research Lab",
             "quotas": {"max_goals": 10, "budget_per_goal": 2000}},
        ]
    }
})

# Tenant-scoped operations
rt.submit_goal("...", auth_context={"role": "admin", "tenant_id": "eng"})
rt.add_agent("...", capabilities=[...],
             auth_context={"role": "admin", "tenant_id": "eng"})

# Cross-tenant isolation enforced automatically
rt.get_goal_status("g-from-eng", auth_context={"tenant_id": "research"})
# → None (not in research tenant)

19. Advanced Execution (Phase 3)

Module: zelos.advanced_execution | Classes: DynamicPlanModifier, SubGoalManager, HumanInTheLoop

Production execution features beyond basic task dispatch.

Dynamic Plan Modification

rt.modify_plan(plan_id, "add_task",
               task_id="hotfix-1",
               description="Emergency security patch",
               required_capability="code-generation.python",
               priority="critical",
               dependencies=["existing-task-3"])

rt.modify_plan(plan_id, "remove_task", task_id="obsolete-task")
rt.modify_plan(plan_id, "modify_task", task_id="t-2",
               priority="critical", timeout_ms=120000)
rt.modify_plan(plan_id, "add_dependency",
               from_task_id="t-1", to_task_id="t-4")
rt.modify_plan(plan_id, "remove_dependency",
               from_task_id="t-1", to_task_id="t-4")

Sub-Goal Spawning

# A task discovers it needs additional work:
sub = rt.spawn_sub_goal(
    parent_task_id="task-123",
    description="Research competitor pricing strategy",
    budget=25.0,
    required_capability="research.analysis",
    num_tasks=3,  # Create 3 sequential sub-tasks
)
# → {sub_goal_id: "sub-a1b2c3d4", parent_task_id: "task-123",
#    plan_id: "sub-plan-sub-a1b2c3d4", task_ids: ["sub-...-t1", "sub-...-t2", "sub-...-t3"],
#    budget: 25.0, status: "running"}

# Check sub-goal status
status = rt.get_sub_goal_status(sub["sub_goal_id"])

# All sub-goals for a parent auto-propagate:
# - All sub-goals completed → parent can complete
# - Any sub-goal failed → parent fails

Human-in-the-Loop Approval

# ── Require approval before goal executes ──
goal = rt.submit_goal(
    "Deploy v3.0 to production",
    require_approval=True,
    approvers=["alice", "bob"],  # Both must approve
    auth_context={"role": "admin"},
)
# → Goal is "planned" but tasks won't dispatch until approved

# ── Check pending approvals ──
pending = rt.list_pending_approvals()
# → [{request_id, task_id, description, approvers, require_all, status}, ...]

# ── First approver ──
rt.approve_task(pending[0]["task_id"], "alice", "Code review passed ✅")
# → Still pending — needs bob too

# ── Second approver ──
rt.approve_task(pending[0]["task_id"], "bob", "Monitoring dashboards clean ✅")
# → Approved! Tasks start dispatching

# ── Alternatively: Reject ──
rt.reject_task(task_id, "alice", "Security scan found 3 critical issues")
# → Goal marked "failed"

# ── Request changes ──
rt._hitl.request_changes(request_id, "bob", "Please add integration tests")
# → Status: "changes_requested" — developer sees feedback

20. Container Isolation

Run plugins in isolated containers for maximum security and resource control.

from zelos.container_isolation import ContainerPluginConfig, ContainerRunner, IsolationFactory

# Docker container config
config = ContainerPluginConfig(
    image="python:3.12-slim",
    command=["python", "-u", "agent.py"],
    env={"ZELOS_API_KEY": "zk-agent-xxx"},
    mounts={"/host/code": "/app/code"},
    cpu_limit=2,
    memory_mb=512,
    network="zelos-net",
)
print(config.to_docker_command())
# → docker run --cpus 2 --memory 512m --network zelos-net ...

# Remote plugin (HTTP)
from zelos.container_isolation import RemotePlugin
remote = RemotePlugin(
    task_endpoint="https://agent.example.com/execute",
    health_endpoint="https://agent.example.com/health",
)
# Health check via GET, tasks dispatched via POST, results via callback URL

# Factory — 5 isolation modes
runner = IsolationFactory.create("docker", config)
runner = IsolationFactory.create("podman", config)
runner = IsolationFactory.create("remote", endpoint_cfg)
runner = IsolationFactory.create("subprocess", {"entrypoint": "agent.py"})
runner = IsolationFactory.create("in-process", {})

21. Hot Reload (Phase 3)

Module: zelos.hot_reload | Classes: HotReloadManager, FileWatcher, UpgradeStrategy

Zero-downtime plugin upgrades. Supports 4 strategies with distinct behaviors.

Upgrade Strategies

Usage

from zelos.hot_reload import HotReloadManager, UpgradeStrategy

hrm = HotReloadManager(upgrade_strategy=UpgradeStrategy.ROLLING)

# ── Version lifecycle ──
hrm.register_version("my-plugin", "1.0.0", "plugins:v1", checksum="sha256:abc")
hrm.register_version("my-plugin", "1.1.0", "plugins:v2", checksum="sha256:def")
hrm.register_version("my-plugin", "2.0.0", "plugins:v3", checksum="sha256:ghi")

active = hrm.get_active_version("my-plugin")
# → PluginVersion(version="2.0.0", status="active")

# ── Drain old ──
hrm.drain_version("my-plugin", "1.0.0")
# → status: "drained" — no new tasks routed here

# ── Rollback ──
hrm.rollback("my-plugin", "1.1.0")
# → Version 1.1.0 active again

# ── Canary deployment ──
hrm.set_upgrade_strategy(UpgradeStrategy.CANARY)
hrm.register_version("api-gateway", "3.0.0", "gateway:v3",
                     canary_percent=5)
# → 5% traffic → v3.0.0, 95% → v2.x

# ── Version history ──
history = hrm.get_version_history("my-plugin")
# → [v1.0.0 (drained), v1.1.0 (active), v2.0.0 (rolled_back)]

# ── File watcher ──
fw = FileWatcher("/etc/zelos/plugins", patterns=["*.py"])
fw.on_change(lambda event: print(f"Plugin changed: {event['filename']}"))
fw.start()

Runtime Integration

# In zelos.yaml:
# hot_reload:
#   plugin_dir: "/etc/zelos/plugins"
#   upgrade_strategy: "rolling"

rt.reload_plugin("verifier", "zelos.verifier_v3:FullVerifier", "3.0.0")
rt.rollback_plugin("verifier", "2.0.0")
rt.set_upgrade_strategy("canary")
rt.get_plugin_versions("verifier")

22. Distributed Runtime (Phase 3)

Module: zelos.distributed | Classes: LeaderElection, WorkStealing, NodeRegistry

Multi-node cluster coordination. Reference implementation using in-memory coordination (production: etcd/Raft).

Leader Election

from zelos.distributed import LeaderElection

le = LeaderElection(node_id="zelos-01", heartbeat_interval_ms=500)
le.register_peer("zelos-02")
le.register_peer("zelos-03")
le.on_leader_change(lambda new_leader: print(f"New leader: {new_leader}"))
le.start()

le.is_leader()       # → True/False
le.get_leader_id()   # → "zelos-01" or None
le.resign()          # Voluntary step-down
le.stop()

Work Stealing

from zelos.distributed import WorkStealing

ws = WorkStealing(node_id="worker-1", max_concurrent_tasks=10)

ws.enqueue_task("t-1", "code-generation.python", priority="high")
ws.enqueue_task("t-2", "code-review.security", priority="medium")

ws.queue_size()       # → 2
ws.get_load_percent() # → 20.0%

# Steal from overloaded peer
overloaded = WorkStealing("worker-2", max_concurrent_tasks=5)
for i in range(8):
    overloaded.enqueue_task(f"heavy-{i}", "code-generation.python")

stolen = ws.steal_from(overloaded, max_count=3)
# → [{task_id, capability, priority, stolen_from, stolen_at}, ...]

Node Registry

from zelos.distributed import NodeRegistry, ClusterNode

nr = NodeRegistry(heartbeat_timeout_seconds=30.0)

nr.register(ClusterNode(
    node_id="zelos-primary", host="10.0.1.10", port=9876,
    capabilities=["code-generation.python", "code-review.security"],
    capacity=20,
))

nr.heartbeat("zelos-primary")
nr.find_by_capability("code-generation.python")  # → [ClusterNode, ...]
nr.detect_dead_nodes()                            # → ["zelos-worker-2"]
nr.cluster_status()
# → {total_nodes: 4, healthy_nodes: 3, dead_nodes: 1, cluster_capabilities: [...]}

Runtime Integration

# zelos.yaml
distributed:
  enabled: true
  node_id: "zelos-master-1"
  host: "10.0.1.10"
  port: 9876
  capacity: 20
  peers:
    - node_id: "zelos-worker-1"
      host: "10.0.1.11"
      capabilities: ["automation.browser"]
    - node_id: "zelos-worker-2"
      host: "10.0.1.12"
      capabilities: ["code-review.security"]

rt.is_leader()            # → True/False
rt.get_cluster_status()   # → {total_nodes, healthy_nodes, is_leader, ...}
rt.get_work_queue_depth() # → current task queue depth

23. v0.9.0 — Change Evidence Package

v0.9.0 upgrades Zelos from execution engine to governance platform. Humans review Intent + Evidence + Confidence, not code.

23.1 Execution Trace

Query the complete execution timeline of any Goal:

trace = runtime.get_goal_trace(goal_id)
# Each task has: timeline events, status, agent, input/output
trace = runtime.get_goal_trace(goal_id, include_artifacts=True)
trace = runtime.get_goal_trace(goal_id, limit=50, offset=0)  # pagination

23.2 Evidence Collection

Agents produce typed Evidence alongside Artifacts:

from zelos.evidence import Evidence

result.evidence = [
    Evidence("test_result", "pytest", "PASS", {"passed": 47}),
    Evidence("security_scan", "bandit", "PASS", {"issues": 0}),
    Evidence("benchmark", "wrk", "PASS", {"tps": 1120}),
]

Runtime auto-aggregates into EvidenceBag with per-type summaries.

23.3 Confidence Scoring

report = runtime.get_execution_report(goal_id)
print(report.confidence.score)          # 0.97
print(report.confidence.recommendation)  # "approve"

6-factor weighted scoring, configurable via zelos.yaml.

23.4 Execution Report

Unified Change Evidence Package — one API call:

report = runtime.get_execution_report(goal_id)
# Contains: intent, architecture_delta, trace, evidence, confidence, rollback

23.5 Policy Gate v2

Evidence-based auto-approve/reject/human-review:

policy_gate:
  rules:
    - if: "confidence >= 0.95 AND risk == 'low'"
      then: "auto_approve"
    - if: "confidence < 0.40"
      then: "auto_reject"
    - default: "require_human"

23.6 Intent Specification

from zelos.execution_report import IntentSpec
intent = IntentSpec(
    description="Implement OAuth2 login",
    success_criteria=["Google OAuth", "Token 24h expiry"],
    constraints=["Don't break password login"],
)
runtime.submit_goal("OAuth2 login", intent=intent)

24. CLI Tool (Phase 3)

Module: zelos.cli | Class: ZelosCLI

Complete command-line interface.

Available Commands

zelos --version                              # 0.9.0
zelos --help

# Runtime
zelos start --config zelos.yaml --port 9876
zelos stop

# Goals
zelos goal submit --description "Build a landing page" --priority high --budget 100
zelos goal status --goal-id g-abc123
zelos goal list
zelos goal cancel --goal-id g-abc123

# Agents
zelos agent list
zelos agent info --agent-id agent-codex-1

# Admin
zelos health
zelos metrics
zelos plugin list
zelos namespace list
zelos config show
zelos config validate

Programmatic Usage

from zelos.cli import ZelosCLI

cli = ZelosCLI(runtime=my_runtime)  # Connect to real runtime
# OR
cli = ZelosCLI()                     # Simulation mode

output = cli.run(["goal", "submit", "--description", "Test"])
print(output)

24. Configuration

Complete zelos.yaml reference:

runtime:
  instance_id: "zelos-prod-01"
  api:
    host: "0.0.0.0"
    port: 9876
    auth:
      enabled: true
      keys:
        - key: "zelos_admin_***"
          role: admin
  limits:
    max_goals: 1000
    max_tasks_per_goal: 100

plugins:
  - id: "llm-planner"
    type: "planner"
    entrypoint: "zelos.planner.LLMPlanner"
    config:
      provider: "openai"
      model: "gpt-4o"
      api_key: "${OPENAI_API_KEY}"
      temperature: 0.3

  - id: "schema-verifier"
    type: "verifier"
    entrypoint: "zelos.verifier.SchemaVerifier"

  - id: "default-policy"
    type: "policy"
    entrypoint: "zelos.policy.CompositePolicy"
    config:
      max_cost_per_goal: 100.0
      max_tasks_per_minute: 60

  - id: "default-scoring"
    type: "scoring_strategy"
    entrypoint: "zelos.scheduler.DefaultScoringStrategy"
    config:
      weights:
        success_rate: 0.30
        cost_efficiency: 0.20

  - id: "inmemory-storage"
    type: "storage"
    entrypoint: "zelos.storage.InMemoryStorageBackend"

storage:
  type: "postgresql"
  url: "postgresql://user:pass@host:5432/zelos"

security:                     # Phase 3
  cert_file: "/etc/zelos/certs/server.crt"
  key_file: "/etc/zelos/certs/server.key"
  ca_file: "/etc/zelos/certs/ca.crt"
  require_client_cert: true
  audit_max_events: 100000

multi_tenancy:                # Phase 3
  enabled: true
  tenants:
    - id: "eng"
      name: "Engineering"
      quotas:
        max_goals: 100
        max_agents: 20
        budget_per_goal: 5000

hot_reload:                   # Phase 3
  plugin_dir: "/etc/zelos/plugins"
  upgrade_strategy: "rolling"
  watch_patterns: ["*.py", "*.yaml"]

distributed:                  # Phase 3
  enabled: true
  node_id: "zelos-01"
  host: "10.0.1.10"
  port: 9876
  capacity: 20
  peers:
    - node_id: "zelos-02"
      host: "10.0.1.11"

25. API Reference

Auth Context

Every API method accepts an optional auth_context dict:

# Full auth
{"role": "admin", "tenant_id": "eng", "actor": "alice@corp.com"}

# API key auth
{"api_key": "zelos_a1b2c3..."}

# No auth → defaults to admin (backward compatible)
None

Return Value Patterns

Success:

{"goal_id": "g-...", "status": "planned", "plan_id": "...", "task_count": 4}
{"status": "ok", "operation": "add_task"}
{"status": "approved"}

Rejection:

{"goal_id": "...", "status": "rejected", "reason": "Permission denied: role 'agent' cannot 'goal.submit'"}
{"status": "rejected", "reason": "Goal quota exceeded for tenant 'eng'"}
{"goal_id": "...", "status": "rejected", "reason": "Invalid or expired API key"}

Error:

{"status": "failed", "reason": "Task 'x' not found"}
{"status": "not_found"}
None

Role Permission Matrix

26. Deployment Guide

Development (Single Node)

python3 -c "
from zelos.runtime import ZelosRuntime
rt = ZelosRuntime.from_yaml('zelos.yaml')
rt.start()
# ... ctrl-c to stop
"

Production (Multi-node)

# Node 1 (leader candidate)
distributed:
  enabled: true
  node_id: "zelos-primary"
  host: "10.0.1.10"
  peers:
    - {node_id: "zelos-worker-1", host: "10.0.1.11"}
    - {node_id: "zelos-worker-2", host: "10.0.1.12"}

security:
  cert_file: "/etc/zelos/certs/server.crt"
  key_file: "/etc/zelos/certs/server.key"
  ca_file: "/etc/zelos/certs/ca.crt"
  require_client_cert: true

storage:
  type: "postgresql"
  url: "postgresql://zelos:${DB_PASS}@pg-cluster:5432/zelos"

Container Deployment

FROM python:3.12-slim
COPY zelos/ /app/zelos/
COPY zelos.yaml /app/
CMD ["python3", "-c", "from zelos.runtime import ZelosRuntime; \
     rt = ZelosRuntime.from_yaml('zelos.yaml'); rt.start()"]

28. Coordination (etcd + InMemory)

Pluggable distributed coordination — leader election, node registry, watch.

from zelos.coordination import (
    InMemoryCoordinationBackend, EtcdCoordinationBackend,
    CoordinationNode, create_coordination_backend,
)

# InMemory (default, single-node)
coord = create_coordination_backend({"type": "memory"})
coord.connect()
coord.register_node(CoordinationNode(node_id="node-1", capabilities=["code"]))

# Bully algorithm — smallest node_id wins
coord.elect_leader("node-1", ttl_seconds=30)
print(coord.get_leader())  # → "node-1"

# Watch for leader changes
coord.watch_leader(lambda new_leader: print(f"Leader changed: {new_leader}"))

# etcd (production, multi-node)
coord_etcd = create_coordination_backend({
    "type": "etcd",
    "endpoints": "localhost:2379",
    "prefix": "/zelos/",
})
coord_etcd.connect()
coord_etcd.register_node(CoordinationNode(node_id="prod-node-1", host="10.0.0.1", port=9001))
coord_etcd.elect_leader("prod-node-1", ttl_seconds=30)
coord_etcd.heartbeat("prod-node-1")  # Renew lease

29. Messaging (NATS + InMemory)

Pluggable message bus for cross-node EventBus transport.

from zelos.messaging_nats import (
    InMemoryMessageBus, NatsMessageBus, create_message_bus,
)

# InMemory (default, single-node)
bus = create_message_bus({"type": "memory"})
bus.connect()

# Publish-Subscribe with pattern matching
bus.subscribe("task.*", lambda data, headers: print(f"Task event: {data}"))
bus.publish("task.created", {"task_id": "t1", "capability": "code"})
bus.publish("task.completed", {"task_id": "t1"})
bus.publish("goal.submitted", {"goal_id": "g1"})  # Not matched by "task.*"

# Request-Reply pattern
def handler(data, headers):
    reply = headers.get("reply")
    if reply:
        bus.publish(reply, {"answer": f"Processed: {data['question']}"})

bus.subscribe("query", handler)
result = bus.request("query", {"question": "status?"}, timeout=1.0)
print(result["answer"])  # → "Processed: status?"

# NATS (production, multi-node)
bus_nats = create_message_bus({
    "type": "nats",
    "servers": ["nats://localhost:4222"],
})
bus_nats.connect()
bus_nats.subscribe("zelos.events", process_fn)
bus_nats.publish("zelos.events", {"event_type": "task.completed"})

30. Go SDK

Build Agents and submit Goals from Go — the cloud-native language.

package main

import (
    "fmt"
    "github.com/AI-Zelos/zelos-go/client"
    "github.com/AI-Zelos/zelos-go/agent"
    "github.com/AI-Zelos/zelos-go/schema"
)

func main() {
    // ── Remote client ──
    c := client.New("http://localhost:9876", "zk-client-dev")
    health, _ := c.Health()
    fmt.Println(health["status"])  // → "healthy"

    goal, _ := c.SubmitGoal("Build a landing page", "high")
    fmt.Println(goal["goal_id"])

    // ── Build a custom Agent ──
    type MyCoder struct {
        agent.BaseAgent
    }

    func (m *MyCoder) DeclareCapabilities() []schema.CapabilityDeclaration {
        return []schema.CapabilityDeclaration{{
            Name: "code-generation.python",
            Version: "1.0.0",
            Tags: []string{"python"},
        }}
    }

    func (m *MyCoder) Execute(task schema.Task) (schema.TaskResult, error) {
        return schema.TaskResult{
            Status: "completed",
            Artifact: &schema.ArtifactResult{
                ContentType: "text/plain",
                Content: "def hello(): return 'Hello from Zelos!'",
            },
        }, nil
    }
}

Available packages: | Package | Description | |---------|-------------| | schema | CapabilityDeclaration, Task, Artifact, MemoryContext types | | agent | Agent interface, BaseAgent, DemoAgent | | client | HTTP client for remote Runtime (Goal/Agent/Admin APIs) |

31. TypeScript SDK

Build Agents and submit Goals from TypeScript/JavaScript.

import { ZelosClient, BaseAgent, DemoAgent } from "@zelos/sdk";
import type { CapabilityDeclaration, Task, TaskResult } from "@zelos/sdk";

// ── Remote client ──
const client = new ZelosClient("http://localhost:9876", "zk-client-dev");
const health = await client.health();
console.log(health.status);  // → "healthy"

// ── Build a custom Agent ──
class MyCoder extends BaseAgent {
  declareCapabilities(): CapabilityDeclaration[] {
    return [{ name: "code-generation.python", version: "1.0.0" }];
  }

  async execute(task: Task): Promise<TaskResult> {
    return {
      status: "completed",
      artifact: { contentType: "text/plain", content: "done" },
    };
  }
}

Install: npm install @zelos/sdk

32. OpenTelemetry Integration

Export spans to Jaeger/Zipkin for distributed tracing.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# Configure OTLP exporter → Jaeger
exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
provider = TracerProvider(resource=Resource.create({"service.name": "zelos-runtime"}))
provider.add_span_processor(BatchSpanProcessor(exporter))

tracer = provider.get_tracer("zelos")

# Create spans (goal → plan → dispatch → execute)
with tracer.start_as_current_span("goal.submit") as span:
    span.set_attribute("goal.description", "Build a REST API")
    span.add_event("goal.accepted", {"goal_id": "g-001"})

    with tracer.start_as_current_span("planner.plan") as plan:
        plan.set_attribute("tasks.count", 3)
        plan.add_event("plan.created", {"plan_id": "plan-001"})

provider.force_flush(timeout_millis=5000)  # Push to Jaeger
# → View traces at http://localhost:16686

27. Troubleshooting

34. v1.1.0 — Agent Credential Management

AI Agents need identity credentials (JWT, API keys, OAuth tokens) to access external services. v1.1.0 makes the Runtime responsible for credential lifecycle — storage, injection, validation, and isolation.

34.1 Core Principles

• Credentials belong to the Runtime, never to Agents
• Zero-leak between Agents: Agent A never sees Agent B's credentials
• Credentials are injected in-memory only — never appear in EventBus events

34.2 Agent Registration

runtime.add_agent(
    "GitHubCoder", "agents.coder:CodingAgent",
    capabilities=[...],
    required_credentials=["github-token"],  # v1.1.0
)

34.3 Using Credentials in Agents

class CodingAgent(BaseAgent):
    def execute(self, task):
        creds = task.constraints.get("credential_refs", {})
        token = creds["github-token"]["token"]
        headers = {"Authorization": f"Bearer {token}"}
        response = requests.get("https://api.github.com/user", headers=headers)

34.4 Four Backends

34.5 Configuration (zelos.yaml)

credentials:
  store: "env"              # env | file | vault | k8s
  policy:
    validate_on_dispatch: true
    refresh_before_expiry_seconds: 300

34.6 Custom CredentialStore

from zelos.credential_store import CredentialStore, Credential

class MyStore(CredentialStore):
    def get(self, credential_name, agent_id):
        # Custom logic: DB, cloud KMS, encrypted file, etc.
        return Credential(name=credential_name, token="...")