Swarm Orchestrator Skill
Scope: The bundled Python scripts (
scripts/*.py
) make no network calls, use only the Python standard library, and have zero third-party dependencies. Tokens are UUID-based (grant_{uuid4().hex}
) stored in data/active_grants.json
. Audit logging is plain JSONL (data/audit_log.jsonl
).
Data-flow notice: This skill does NOT implement, invoke, or control
sessions_send
. That is a host-platform built-in (OpenClaw runtime). The orchestration instructions below describe when to call the platform's sessions_send
after budget checks pass — but the actual network call, model endpoint, and data transmission are entirely the host platform's responsibility. If you need to prevent external network calls, disable or reroute sessions_send
in your platform settings before installing this skill.
PII / sensitive-data warning: The
justification
field in permission requests and the audit log (data/audit_log.jsonl
) store free-text strings provided by agents. Do not include PII, secrets, or credentials in justification text. Consider restricting file permissions on data/
or running this skill in an isolated workspace.
Setup
No pip install required. All 6 scripts use Python standard library only — zero third-party packages.
Note on requirements.txt
: The file exists for documentation purposes only — it lists the stdlib modules used and has no required packages. All listed deps are commented out as optional. You do not need to run
pip install -r requirements.txt
.
# Prerequisite: python3 (any version ≥ 3.8)
python3 --version
That's it. Run any script directly:
python3 scripts/blackboard.py list
python3 scripts/swarm_guard.py budget-init --task-id "task_001" --budget 10000
Optional: for cross-platform file locking on Windows production hosts
pip install filelock # only needed if you see locking issues on Windows
The
data/
directory is created automatically on first run. No configuration files, environment variables, or credentials are required.
Multi-agent coordination system for complex workflows requiring task delegation, parallel execution, and permission-controlled access to sensitive APIs.
🎯 Orchestrator System Instructions
You are the Orchestrator Agent responsible for decomposing complex tasks, delegating to specialized agents, and synthesizing results. Follow this protocol:
Core Responsibilities
- DECOMPOSE complex prompts into 3 specialized sub-tasks
- DELEGATE using the budget-aware handoff protocol
- VERIFY results on the blackboard before committing
- SYNTHESIZE final output only after all validations pass
Task Decomposition Protocol
When you receive a complex request, decompose it into exactly 3 sub-tasks:
┌─────────────────────────────────────────────────────────────────┐
│ COMPLEX USER REQUEST │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────┼─────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ SUB-TASK 1 │ │ SUB-TASK 2 │ │ SUB-TASK 3 │
│ data_analyst │ │ risk_assessor │ │strategy_advisor│
│ (DATA) │ │ (VERIFY) │ │ (RECOMMEND) │
└───────────────┘ └───────────────┘ └───────────────┘
│ │ │
└─────────────────────┼─────────────────────┘
▼
┌───────────────┐
│ SYNTHESIZE │
│ orchestrator │
└───────────────┘
Decomposition Template:
TASK DECOMPOSITION for: "{user_request}"
Sub-Task 1 (DATA): [data_analyst]
- Objective: Extract/process raw data
- Output: Structured JSON with metrics
Sub-Task 2 (VERIFY): [risk_assessor]
- Objective: Validate data quality & compliance
- Output: Validation report with confidence score
Sub-Task 3 (RECOMMEND): [strategy_advisor]
Objective: Generate actionable insights
Output: Recommendations with rationale
Budget-Aware Handoff Protocol
CRITICAL: Before EVERY
sessions_send
, call the handoff interceptor:
# ALWAYS run this BEFORE sessions_send
python {baseDir}/scripts/swarm_guard.py intercept-handoff \
--task-id "task_001" \
--from orchestrator \
--to data_analyst \
--message "Analyze Q4 revenue data"
Decision Logic:
IF result.allowed == true:
→ Proceed with sessions_send
→ Note tokens_spent and remaining_budget
ELSE:
→ STOP - Do NOT call sessions_send
→ Report blocked reason to user
→ Consider: reduce scope or abort task
Pre-Commit Verification Workflow
Before returning final results to the user:
# Step 1: Check all sub-task results on blackboard
python {baseDir}/scripts/blackboard.py read "task:001:data_analyst"
python {baseDir}/scripts/blackboard.py read "task:001:risk_assessor"
python {baseDir}/scripts/blackboard.py read "task:001:strategy_advisor"
Step 2: Validate each result
python {baseDir}/scripts/swarm_guard.py validate-result
--task-id "task_001"
--agent data_analyst
--result '{"status":"success","output":{...},"confidence":0.85}'
Step 3: Supervisor review (checks all issues)
python {baseDir}/scripts/swarm_guard.py supervisor-review --task-id "task_001"
Step 4: Only if APPROVED, commit final state
python {baseDir}/scripts/blackboard.py write "task:001:final"
'{"status":"SUCCESS","output":{...}}'
Verdict Handling:
| Verdict | Action |
|---|
APPROVED
| Commit and return results to user |
WARNING
| Review issues, fix if possible, then commit |
BLOCKED
| Do NOT return results. Report failure. |
The 3-Layer Memory Model
Every agent in the swarm operates with three memory layers, each with a different scope and lifetime:
| Layer | Name | Lifetime | Managed by |
|---|
| 1 | Agent context | Ephemeral — current task only | Platform (per-session) |
| 2 | Blackboard | TTL-scoped — shared across agents | scripts/blackboard.py
|
| 3 | Project context | Persistent — survives all sessions | scripts/context_manager.py
|
Layer 1 — Agent Context
Each agent's own context window: the current task instructions, conversation history, and immediate working memory. Managed automatically by the OpenClaw/LLM platform. Nothing to configure.
Layer 2 — Blackboard (Shared Coordination State)
A shared markdown file (
swarm-blackboard.md
) for real-time cross-agent coordination: task results, grant tokens, status flags, and TTL-scoped cache entries. Agents read and write via
scripts/blackboard.py
. Entries expire automatically.
Layer 3 — Project Context (Persistent Long-Term Memory)
A JSON file (
data/project-context.json
) that holds information every agent should know, regardless of what session or task is running:
- Goals — long-term objectives of the project
- Tech stack — languages, frameworks, infrastructure
- Milestones — completed, in-progress, and planned work
- Architecture decisions — design choices and their rationales
- Banned approaches — approaches that have been ruled out
Initialising Project Context
python {baseDir}/scripts/context_manager.py init \
--name "MyProject" \
--description "Multi-agent workflow automation" \
--version "1.0.0"
Injecting Context into an Agent System Prompt
python {baseDir}/scripts/context_manager.py inject
Copy the output block to the top of your agent's system prompt. Every agent that receives this block shares the same long-term project awareness.
Recording a Decision
python {baseDir}/scripts/context_manager.py update \
--section decisions \
--add '{"decision": "Use atomic blackboard commits", "rationale": "Prevent race conditions in parallel agents"}'
Updating Milestones
# Mark a milestone complete
python {baseDir}/scripts/context_manager.py update \
--section milestones --complete "Ship v2.0"
Add a planned milestone
python {baseDir}/scripts/context_manager.py update
--section milestones --add '{"planned": "Integrate vector memory"}'
Setting the Tech Stack
python {baseDir}/scripts/context_manager.py update \
--section stack \
--set '{"language": "Python", "runtime": "Python 3.11", "framework": "SwarmOrchestrator"}'
Banning an Approach
python {baseDir}/scripts/context_manager.py update \
--section banned \
--add "Direct database writes from agent scripts (use permission gating)"
When to Use This Skill
- Task Delegation: Route work to specialized agents (data_analyst, strategy_advisor, risk_assessor)
- Parallel Execution: Run multiple agents simultaneously and synthesize results
- Permission Wall: Gate access to DATABASE, PAYMENTS, EMAIL, or FILE_EXPORT operations (abstract local resource types — no external credentials required)
- Shared Blackboard: Coordinate agent state via persistent markdown file
Quick Start
1. Initialize Budget (FIRST!)
Always initialize a budget before any multi-agent task:
python {baseDir}/scripts/swarm_guard.py budget-init \
--task-id "task_001" \
--budget 10000 \
--description "Q4 Financial Analysis"
2. Delegate a Task to Another Session
Platform note:
sessions_list
, sessions_send
, and sessions_history
are OpenClaw host platform built-ins — they are part of the OpenClaw runtime, not provided or invoked by this skill's Python scripts. This skill only runs local python scripts/*.py
commands. The guidance below describes how to combine the platform's session tools with this skill's budget guard.
First check budget, then use the OpenClaw platform operation:
# 1. Check budget (this skill's Python script)
python {baseDir}/scripts/swarm_guard.py intercept-handoff \
--task-id "task_001" --from orchestrator --to data_analyst \
--message "Analyze Q4 revenue data"
2. If allowed, delegate using the OpenClaw platform tool (not this skill):
sessions_list → see available sessions/agents
sessions_send → send task to another session
sessions_history → check results from delegated work
Example delegation prompt:
After running swarm_guard.py intercept-handoff and getting result.allowed == true,
use the OpenClaw sessions_send platform tool to ask the data_analyst session:
"Analyze Q4 revenue trends from the SAP export data and summarize key insights"
3. Check Permission Before API Access
Before accessing SAP or Financial APIs, evaluate the request:
# Run the permission checker script
python {baseDir}/scripts/check_permission.py \
--agent "data_analyst" \
--resource "DATABASE" \
--justification "Need Q4 invoice data for quarterly report" \
--scope "read:invoices"
The script will output a grant token if approved, or denial reason if rejected.
4. Use the Shared Blackboard
Read/write coordination state:
# Write to blackboard
python {baseDir}/scripts/blackboard.py write "task:q4_analysis" '{"status": "in_progress", "agent": "data_analyst"}'
Read from blackboard
python {baseDir}/scripts/blackboard.py read "task:q4_analysis"
List all entries
python {baseDir}/scripts/blackboard.py list
Agent-to-Agent Handoff Protocol
When delegating tasks between agents/sessions:
Step 1: Initialize Budget & Check Capacity
# Initialize budget (if not already done)
python {baseDir}/scripts/swarm_guard.py budget-init --task-id "task_001" --budget 10000
Check current status
python {baseDir}/scripts/swarm_guard.py budget-check --task-id "task_001"
Step 2: Identify Target Agent
Platform note:
sessions_list
is an OpenClaw host platform built-in, not provided by this skill.
sessions_list # OpenClaw platform operation — find available agents
Common agent types:
| Agent | Specialty |
|---|
data_analyst
| Data processing, SQL, analytics |
strategy_advisor
| Business strategy, recommendations |
risk_assessor
| Risk analysis, compliance checks |
orchestrator
| Coordination, task decomposition |
Step 3: Intercept Before Handoff (REQUIRED)
# This checks budget AND handoff limits before allowing the call
python {baseDir}/scripts/swarm_guard.py intercept-handoff \
--task-id "task_001" \
--from orchestrator \
--to data_analyst \
--message "Analyze Q4 data" \
--artifact # Include if expecting output
If ALLOWED: Proceed to Step 4
If BLOCKED: Stop - do not call sessions_send
Step 4: Construct Handoff Message
Include these fields in your delegation:
- instruction: Clear task description
- context: Relevant background information
- constraints: Any limitations or requirements
- expectedOutput: What format/content you need back
Step 5: Send via OpenClaw Platform Session Tool
Platform note:
sessions_send
is an OpenClaw host platform built-in — it is NOT implemented by this skill. This skill only provides the budget guard (swarm_guard.py
) that must be run first.
# OpenClaw platform operation (not this skill):
sessions_send to data_analyst:
"[HANDOFF]
Instruction: Analyze Q4 revenue by product category
Context: Using SAP export from ./data/q4_export.csv
Constraints: Focus on top 5 categories only
Expected Output: JSON summary with category, revenue, growth_pct
[/HANDOFF]"
Step 6: Check Results
Platform note:
sessions_history
is an OpenClaw host platform built-in, not provided by this skill.
sessions_history data_analyst # OpenClaw platform operation — get the response
Permission Wall
CRITICAL: Always check permissions before accessing:
DATABASE
- Internal database / data store accessPAYMENTS
- Financial/payment data servicesEMAIL
- Email sending capabilityFILE_EXPORT
- Exporting data to local files
Note: These are abstract local resource type names used by
check_permission.py
. No external API credentials are required or used — all permission evaluation runs locally.
Permission Evaluation Criteria
| Factor | Weight | Criteria |
|---|
| Justification | 40% | Must explain specific task need |
| Trust Level | 30% | Agent's established trust score |
| Risk Assessment | 30% | Resource sensitivity + scope breadth |
Using the Permission Script
# Request permission
python {baseDir}/scripts/check_permission.py \
--agent "your_agent_id" \
--resource "PAYMENTS" \
--justification "Generating quarterly financial summary for board presentation" \
--scope "read:revenue,read:expenses"
Output if approved:
✅ GRANTED
Token: grant_a1b2c3d4e5f6
Expires: 2026-02-04T15:30:00Z
Restrictions: read_only, no_pii_fields, audit_required
Output if denied:
❌ DENIED
Reason: Justification is insufficient. Please provide specific task context.
Restriction Types
| Resource | Default Restrictions |
|---|
| DATABASE | read_only
, max_records:100
|
| PAYMENTS | read_only
, no_pii_fields
, audit_required
|
| EMAIL | rate_limit:10_per_minute
|
| FILE_EXPORT | anonymize_pii
, local_only
|
Shared Blackboard Pattern
The blackboard (
swarm-blackboard.md
) is a markdown file for agent coordination:
# Swarm Blackboard
Last Updated: 2026-02-04T10:30:00Z
Knowledge Cache
task:q4_analysis
{"status": "completed", "result": {...}, "agent": "data_analyst"}
cache:revenue_summary
{"q4_total": 1250000, "growth": 0.15}
Blackboard Operations
# Write with TTL (expires after 1 hour)
python {baseDir}/scripts/blackboard.py write "cache:temp_data" '{"value": 123}' --ttl 3600
Read (returns null if expired)
python {baseDir}/scripts/blackboard.py read "cache:temp_data"
Delete
python {baseDir}/scripts/blackboard.py delete "cache:temp_data"
Get full snapshot
python {baseDir}/scripts/blackboard.py snapshot
Parallel Execution
For tasks requiring multiple agent perspectives:
Strategy 1: Merge (Default)
Combine all agent outputs into unified result.
Ask data_analyst AND strategy_advisor to both analyze the dataset.
Merge their insights into a comprehensive report.
Strategy 2: Vote
Use when you need consensus - pick the result with highest confidence.
Strategy 3: First-Success
Use for redundancy - take first successful result.
Strategy 4: Chain
Sequential processing - output of one feeds into next.
TypeScript engine (v4.15.0): These strategies map directly to the
FanOutFanIn
module (lib/fan-out.ts
) which provides merge
, vote
, firstSuccess
, and consensus
fan-in strategies with concurrency control. For multi-phase workflows with approval gates, see PhasePipeline
(lib/phase-pipeline.ts
). For result scoring and threshold filtering, see ConfidenceFilter
(lib/confidence-filter.ts
). Matcher-based hooks (lib/adapter-hooks.ts
) can target specific agents or tools via glob patterns. For sandboxed agent execution, see AgentRuntime
(lib/agent-runtime.ts
). For large-scale agent coordination, see StrategyAgent
(lib/strategy-agent.ts
).
Example Parallel Workflow
Platform note:
sessions_send
and sessions_history
are OpenClaw host platform built-ins, not provided by this skill. This skill provides only the swarm_guard.py
budget/handoff check that runs before each delegation.
# For each delegation below, first run:
# python {baseDir}/scripts/swarm_guard.py intercept-handoff --task-id "task_001" --from orchestrator --to <agent> --message "<task>"
# Then, if allowed, use the OpenClaw platform tool:
1. sessions_send to data_analyst: "Extract key metrics from Q4 data"
2. sessions_send to risk_assessor: "Identify compliance risks in Q4 data"
3. sessions_send to strategy_advisor: "Recommend actions based on Q4 trends"
4. Wait for all responses via sessions_history
5. Synthesize: Combine metrics + risks + recommendations into executive summary
Security Considerations
- Never bypass the permission wall for gated resources
- Always include justification explaining the business need
- Use minimal scope - request only what you need
- Check token expiry - tokens are valid for 5 minutes
- Validate tokens - use
python {baseDir}/scripts/validate_token.py TOKEN
to verify grant tokens before use
- Audit trail - all permission requests are logged
📝 Audit Trail Requirements (MANDATORY)
Every sensitive action MUST be logged to data/audit_log.jsonl
to maintain compliance and enable forensic analysis.
Privacy note: Audit log entries contain agent-provided free-text fields (justifications, descriptions). These are stored locally in
data/audit_log.jsonl
and never transmitted over the network by this skill. However, do not put PII, passwords, or API keys in justification strings — they persist on disk. Consider periodic log rotation and restricting OS file permissions on the data/
directory.
What Gets Logged Automatically
The scripts automatically log these events:
permission_granted
- When access is approvedpermission_denied
- When access is rejectedpermission_revoked
- When a token is manually revokedttl_cleanup
- When expired tokens are purgedresult_validated
/ result_rejected
- Swarm Guard validations
Log Entry Format
{
"timestamp": "2026-02-04T10:30:00+00:00",
"action": "permission_granted",
"details": {
"agent_id": "data_analyst",
"resource_type": "DATABASE",
"justification": "Q4 revenue analysis",
"token": "grant_abc123...",
"restrictions": ["read_only", "max_records:100"]
}
}
Reading the Audit Log
# View recent entries (last 10)
tail -10 {baseDir}/data/audit_log.jsonl
Search for specific agent
grep "data_analyst" {baseDir}/data/audit_log.jsonl
Count actions by type
cat {baseDir}/data/audit_log.jsonl | jq -r '.action' | sort | uniq -c
Custom Audit Entries
If you perform a sensitive action manually, log it:
import json
from datetime import datetime, timezone
from pathlib import Path
audit_file = Path("{baseDir}/data/audit_log.jsonl")
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"action": "manual_data_access",
"details": {
"agent": "orchestrator",
"description": "Direct database query for debugging",
"justification": "Investigating data sync issue #1234"
}
}
with open(audit_file, "a") as f:
f.write(json.dumps(entry) + "\n")
🧹 TTL Enforcement (Token Lifecycle)
Expired permission tokens are automatically tracked. Run periodic cleanup:
# Validate a grant token
python {baseDir}/scripts/validate_token.py grant_a1b2c3d4e5f6
List expired tokens (without removing)
python {baseDir}/scripts/revoke_token.py --list-expired
Remove all expired tokens
python {baseDir}/scripts/revoke_token.py --cleanup
Output:
🧹 TTL Cleanup Complete
Removed: 3 expired token(s)
Remaining active grants: 2
Best Practice: Run
--cleanup
at the start of each multi-agent task to ensure a clean permission state.
⚠️ Swarm Guard: Preventing Common Failures
Two critical issues can derail multi-agent swarms:
1. The Handoff Tax 💸
Problem: Agents waste tokens "talking about" work instead of doing it.
Prevention:
# Before each handoff, check your budget:
python {baseDir}/scripts/swarm_guard.py check-handoff --task-id "task_001"
Output:
🟢 Task: task_001
Handoffs: 1/3
Remaining: 2
Action Ratio: 100%
Rules enforced:
- Max 3 handoffs per task - After 3, produce output or abort
- Max 500 chars per message - Be concise: instruction + constraints + expected output
- 60% action ratio - At least 60% of handoffs must produce artifacts
- 2-minute planning limit - No output after 2min = timeout
# Record a handoff (with tax checking):
python {baseDir}/scripts/swarm_guard.py record-handoff \
--task-id "task_001" \
--from orchestrator \
--to data_analyst \
--message "Analyze sales data, output JSON summary" \
--artifact # Include if this handoff produces output
2. Silent Failure Detection 👻
Problem: One agent fails silently, others keep working on bad data.
Prevention - Heartbeats:
# Agents must send heartbeats while working:
python {baseDir}/scripts/swarm_guard.py heartbeat --agent data_analyst --task-id "task_001"
Check if an agent is healthy:
python {baseDir}/scripts/swarm_guard.py health-check --agent data_analyst
Output if healthy:
💚 Agent 'data_analyst' is HEALTHY
Last seen: 15s ago
Output if failed:
💔 Agent 'data_analyst' is UNHEALTHY
Reason: STALE_HEARTBEAT
→ Do NOT use any pending results from this agent.
Prevention - Result Validation:
# Before using another agent's result, validate it:
python {baseDir}/scripts/swarm_guard.py validate-result \
--task-id "task_001" \
--agent data_analyst \
--result '{"status": "success", "output": {"revenue": 125000}, "confidence": 0.85}'
Output:
✅ RESULT VALID
→ APPROVED - Result can be used by other agents
Required result fields:
status
,
output
,
confidence
Supervisor Review
Before finalizing any task, run supervisor review:
python {baseDir}/scripts/swarm_guard.py supervisor-review --task-id "task_001"
Output:
✅ SUPERVISOR VERDICT: APPROVED
Task: task_001
Age: 1.5 minutes
Handoffs: 2
Artifacts: 2
Verdicts:
APPROVED
- Task healthy, results usableWARNING
- Issues detected, review recommendedBLOCKED
- Critical failures, do NOT use results
Troubleshooting
Permission Denied
- Provide more specific justification (mention task, purpose, expected outcome)
- Narrow the requested scope
- Check agent trust level
Blackboard Read Returns Null
- Entry may have expired (check TTL)
- Key may be misspelled
- Entry was never written
Session Not Found
- Run
sessions_list
(OpenClaw platform built-in) to see available sessions
- Session may need to be started first
References
This skill is part of the larger Network-AI project. See the repository for full documentation on the permission system, blackboard schema, and trust-level calculations.
