Restructure: move src to core/ crate, move cmd/supervisor.rs to core/src/bin/supervisor.rs, create workspace

This commit is contained in:
Timur Gordon
2025-11-06 23:30:18 +01:00
parent 0a617ad359
commit 6d518599b8
39 changed files with 824 additions and 140 deletions

71
core/Cargo.toml Normal file
View File

@@ -0,0 +1,71 @@
[package]
name = "hero-supervisor"
version = "0.1.0"
edition = "2021"
[lib]
name = "hero_supervisor"
path = "src/lib.rs"
[[bin]]
name = "supervisor"
path = "src/bin/supervisor.rs"
[dependencies]
# Job types
hero-job = { git = "https://git.ourworld.tf/herocode/job.git" }
hero-job-client = { git = "https://git.ourworld.tf/herocode/job.git" }
# Async runtime
tokio = { version = "1.0", features = ["full"] }
# Async trait support
async-trait = "0.1"
# Redis client
redis = { version = "0.25", features = ["aio", "tokio-comp"] }
# Job module dependencies (now integrated)
uuid = { version = "1.0", features = ["v4"] }
# Logging
log = "0.4"
thiserror = "1.0"
chrono = "0.4"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
env_logger = "0.10"
# CLI argument parsing
clap = { version = "4.0", features = ["derive"] }
toml = "0.8"
# OpenRPC dependencies (now always included)
jsonrpsee = { version = "0.24", features = ["server", "macros"] }
anyhow = "1.0"
# CORS support for OpenRPC server
tower-http = { version = "0.5", features = ["cors"] }
tower = "0.4"
hyper = { version = "1.0", features = ["full"] }
hyper-util = { version = "0.1", features = ["tokio"] }
# Mycelium integration (optional)
base64 = { version = "0.22", optional = true }
rand = { version = "0.8", optional = true }
reqwest = { version = "0.12", features = ["json"], optional = true }
[dev-dependencies]
tokio-test = "0.4"
hero-supervisor-openrpc-client = { path = "../client" }
escargot = "0.5"
[features]
default = ["cli"]
cli = []
mycelium = ["base64", "rand", "reqwest"]
# Examples
[[example]]
name = "osiris_openrpc"
path = "examples/osiris_openrpc/main.rs"

74
core/examples/README.md Normal file
View File

@@ -0,0 +1,74 @@
# Hero Supervisor Examples
This directory contains examples demonstrating Hero Supervisor functionality.
## Available Examples
### osiris_openrpc
Comprehensive example showing the complete workflow of using Hero Supervisor with OSIRIS runners via OpenRPC.
**Features:**
- Automatic supervisor and runner startup
- OpenRPC client communication
- Runner registration and management
- Job dispatching with multiple scripts
- Context-based access control
- Graceful shutdown
**Run:**
```bash
cargo run --example osiris_openrpc
```
See [osiris_openrpc/README.md](osiris_openrpc/README.md) for details.
## Prerequisites
All examples require:
- Redis server running on `localhost:6379`
- Rust toolchain installed
## Example Structure
```
examples/
├── README.md # This file
├── osiris_openrpc/ # OSIRIS + OpenRPC example
│ ├── main.rs # Main example code
│ ├── README.md # Detailed documentation
│ ├── note.rhai # Note creation script
│ ├── event.rhai # Event creation script
│ ├── query.rhai # Query script
│ └── access_denied.rhai # Access control test script
└── _archive/ # Archived old examples
```
## Architecture Overview
The examples demonstrate the Hero Supervisor architecture:
```
Client (OpenRPC)
Supervisor (OpenRPC Server)
Redis Queue
Runners (OSIRIS, SAL, etc.)
```
## Development
To add a new example:
1. Create a new directory under `examples/`
2. Add `main.rs` with your example code
3. Add any required script files (`.rhai`)
4. Add a `README.md` documenting the example
5. Update `Cargo.toml` to register the example
6. Update this README with a link
## Archived Examples
Previous examples have been moved to `_archive/` for reference. These may be outdated but can provide useful patterns for specific use cases.

View File

@@ -0,0 +1,364 @@
# End-to-End Examples
Complete examples demonstrating the full Supervisor + Runner + Client workflow.
## Overview
These examples show how to:
1. Start a Hero Supervisor
2. Start an OSIS Runner
3. Register the runner with the supervisor
4. Execute jobs using both blocking (`job.run`) and non-blocking (`job.start`) modes
## Prerequisites
### Required Services
1. **Redis** - Must be running on `localhost:6379`
```bash
redis-server
```
2. **Supervisor** - Hero Supervisor with Mycelium integration
```bash
cargo run --bin hero-supervisor -- --redis-url redis://localhost:6379
```
3. **Runner** - OSIS Runner to execute jobs
```bash
cargo run --bin runner_osis -- test_runner --redis-url redis://localhost:6379
```
## Examples
### 1. Simple End-to-End (`simple_e2e.rs`)
**Recommended for beginners** - A minimal example with clear step-by-step execution.
#### What it does:
- Registers a runner with the supervisor
- Runs 2 blocking jobs (with immediate results)
- Starts 1 non-blocking job (fire and forget)
- Shows clear output at each step
#### How to run:
**Terminal 1 - Redis:**
```bash
redis-server
```
**Terminal 2 - Supervisor:**
```bash
cd /Users/timurgordon/code/git.ourworld.tf/herocode/supervisor
RUST_LOG=info cargo run --bin hero-supervisor -- --redis-url redis://localhost:6379
```
**Terminal 3 - Runner:**
```bash
cd /Users/timurgordon/code/git.ourworld.tf/herocode/runner_rust
RUST_LOG=info cargo run --bin runner_osis -- test_runner \
--redis-url redis://localhost:6379 \
--db-path /tmp/test_runner.db
```
**Terminal 4 - Demo:**
```bash
cd /Users/timurgordon/code/git.ourworld.tf/herocode/supervisor
RUST_LOG=info cargo run --example simple_e2e
```
#### Expected Output:
```
╔════════════════════════════════════════╗
║ Simple End-to-End Demo ║
╚════════════════════════════════════════╝
📋 Step 1: Registering Runner
─────────────────────────────────────────
✅ Runner registered successfully
📋 Step 2: Running a Simple Job (Blocking)
─────────────────────────────────────────
✅ Job completed!
Result: {"message":"Hello from the runner!","number":42,"timestamp":1234567890}
📋 Step 3: Running a Calculation Job
─────────────────────────────────────────
✅ Calculation completed!
Result: {"sum":55,"product":3628800,"count":10,"average":5}
📋 Step 4: Starting a Non-Blocking Job
─────────────────────────────────────────
✅ Job started!
Job ID: abc-123 (running in background)
🎉 Demo completed successfully!
```
### 2. Full End-to-End (`end_to_end_demo.rs`)
**Advanced** - Automatically spawns supervisor and runner processes.
#### What it does:
- Automatically starts supervisor and runner
- Runs multiple test jobs
- Demonstrates both execution modes
- Handles cleanup automatically
#### How to run:
**Terminal 1 - Redis:**
```bash
redis-server
```
**Terminal 2 - Demo:**
```bash
cd /Users/timurgordon/code/git.ourworld.tf/herocode/supervisor
RUST_LOG=info cargo run --example end_to_end_demo
```
#### Features:
- ✅ Automatic process management
- ✅ Multiple job examples
- ✅ Graceful shutdown
- ✅ Comprehensive logging
## Job Execution Modes
### job.run (Blocking)
Executes a job and waits for the result.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "job.run",
"params": [{
"secret": "admin_secret",
"job": { /* job object */ },
"timeout": 30
}],
"id": 1
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"job_id": "uuid",
"status": "completed",
"result": "{ /* actual result */ }"
},
"id": 1
}
```
**Use when:**
- You need immediate results
- Job completes quickly (< 60 seconds)
- Synchronous workflow
### job.start (Non-Blocking)
Starts a job and returns immediately.
**Request:**
```json
{
"jsonrpc": "2.0",
"method": "job.start",
"params": [{
"secret": "admin_secret",
"job": { /* job object */ }
}],
"id": 1
}
```
**Response:**
```json
{
"jsonrpc": "2.0",
"result": {
"job_id": "uuid",
"status": "queued"
},
"id": 1
}
```
**Use when:**
- Long-running operations
- Background processing
- Async workflows
- Don't need immediate results
## Job Structure
Jobs are created using the `JobBuilder`:
```rust
use runner_rust::job::JobBuilder;
let job = JobBuilder::new()
.caller_id("my_client")
.context_id("my_context")
.payload(r#"
// Rhai script to execute
let result = 2 + 2;
to_json(result)
"#)
.runner("runner_name")
.executor("rhai")
.timeout(30)
.build()?;
```
### Job Fields
- **caller_id**: Identifier for the client making the request
- **context_id**: Context for the job execution
- **payload**: Rhai script to execute
- **runner**: Name of the runner to execute on
- **executor**: Type of executor (always "rhai" for OSIS)
- **timeout**: Maximum execution time in seconds
## Rhai Script Examples
### Simple Calculation
```rhai
let result = 2 + 2;
to_json(result)
```
### String Manipulation
```rhai
let message = "Hello, World!";
let upper = message.to_upper();
to_json(upper)
```
### Array Operations
```rhai
let numbers = [1, 2, 3, 4, 5];
let sum = 0;
for n in numbers {
sum += n;
}
to_json(#{sum: sum, count: numbers.len()})
```
### Object Creation
```rhai
let person = #{
name: "Alice",
age: 30,
email: "alice@example.com"
};
to_json(person)
```
## Troubleshooting
### "Failed to connect to supervisor"
**Problem:** Supervisor is not running or wrong port.
**Solution:**
```bash
# Check if supervisor is running
curl http://localhost:3030
# Start supervisor
cargo run --bin hero-supervisor -- --redis-url redis://localhost:6379
```
### "Runner not found"
**Problem:** Runner is not registered or not running.
**Solution:**
```bash
# Start the runner
cargo run --bin runner_osis -- test_runner --redis-url redis://localhost:6379
# Check runner logs for connection issues
```
### "Job execution timeout"
**Problem:** Job took longer than timeout value.
**Solution:**
- Increase timeout in job builder: `.timeout(60)`
- Or in job.run request: `"timeout": 60`
### "Redis connection failed"
**Problem:** Redis is not running.
**Solution:**
```bash
# Start Redis
redis-server
# Or specify custom Redis URL
cargo run --bin hero-supervisor -- --redis-url redis://localhost:6379
```
## Architecture
```
┌─────────────┐
│ Client │
│ (Example) │
└──────┬──────┘
│ HTTP/JSON-RPC
┌─────────────┐
│ Supervisor │
│ (Mycelium) │
└──────┬──────┘
│ Redis Queue
┌─────────────┐
│ Runner │
│ (OSIS) │
└─────────────┘
```
### Flow
1. **Client** creates a job with Rhai script
2. **Client** sends job to supervisor via JSON-RPC
3. **Supervisor** verifies signatures (if present)
4. **Supervisor** queues job to runner's Redis queue
5. **Runner** picks up job from queue
6. **Runner** executes Rhai script
7. **Runner** stores result in Redis
8. **Supervisor** retrieves result (for job.run)
9. **Client** receives result
## Next Steps
- Add signature verification to jobs (see `JOB_SIGNATURES.md`)
- Implement job status polling for non-blocking jobs
- Create custom Rhai functions for your use case
- Scale with multiple runners
## Related Documentation
- `JOB_EXECUTION.md` - Detailed job execution modes
- `JOB_SIGNATURES.md` - Cryptographic job signing
- `README.md` - Supervisor overview
---
**Status:** ✅ Production Ready
**Last Updated:** 2025-10-24

View File

@@ -0,0 +1,192 @@
# Supervisor Examples - Summary
## ✅ **Complete End-to-End Examples with OpenRPC Client**
All examples now use the official `hero-supervisor-openrpc-client` library for type-safe, async communication with the supervisor.
### **What Was Updated:**
1. **OpenRPC Client Library** (`client/src/lib.rs`)
- Added `JobRunResponse` - Response from blocking `job.run`
- Added `JobStartResponse` - Response from non-blocking `job.start`
- Updated `job_run()` method - Now accepts timeout parameter
- Updated `job_start()` method - Now accepts Job instead of job_id
- Re-exports `Job` and `JobBuilder` from `runner_rust`
2. **Simple E2E Example** (`examples/simple_e2e.rs`)
- Uses `SupervisorClient` from OpenRPC library
- Clean, type-safe API calls
- No manual JSON-RPC construction
- Perfect for learning and testing
3. **Full E2E Demo** (`examples/end_to_end_demo.rs`)
- Automated supervisor and runner spawning
- Uses OpenRPC client throughout
- Helper functions for common operations
- Comprehensive test scenarios
### **Key Changes:**
**Before (Manual JSON-RPC):**
```rust
let request = json!({
"jsonrpc": "2.0",
"method": "job.run",
"params": [{
"secret": secret,
"job": job,
"timeout": 30
}],
"id": 1
});
let response = http_client.post(url).json(&request).send().await?;
```
**After (OpenRPC Client):**
```rust
let response = client.job_run(secret, job, Some(30)).await?;
println!("Result: {:?}", response.result);
```
### **Client API:**
#### **Job Execution**
```rust
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder};
// Create client
let client = SupervisorClient::new("http://localhost:3030")?;
// Register runner
client.register_runner("admin_secret", "runner_name", "queue_name").await?;
// Run job (blocking - waits for result)
let response = client.job_run("admin_secret", job, Some(60)).await?;
// response.result contains the actual result
// Start job (non-blocking - returns immediately)
let response = client.job_start("admin_secret", job).await?;
// response.job_id for later polling
```
#### **Response Types**
```rust
// JobRunResponse (from job.run)
pub struct JobRunResponse {
pub job_id: String,
pub status: String, // "completed"
pub result: Option<String>, // Actual result from runner
}
// JobStartResponse (from job.start)
pub struct JobStartResponse {
pub job_id: String,
pub status: String, // "queued"
}
```
### **Examples Overview:**
| Example | Description | Use Case |
|---------|-------------|----------|
| `simple_e2e.rs` | Manual setup, step-by-step | Learning, testing |
| `end_to_end_demo.rs` | Automated, comprehensive | CI/CD, integration tests |
### **Running the Examples:**
**Prerequisites:**
```bash
# Terminal 1: Redis
redis-server
# Terminal 2: Supervisor
cargo run --bin hero-supervisor -- --redis-url redis://localhost:6379
# Terminal 3: Runner
cargo run --bin runner_osis -- test_runner --redis-url redis://localhost:6379
```
**Run Simple Example:**
```bash
# Terminal 4
RUST_LOG=info cargo run --example simple_e2e
```
**Run Full Demo:**
```bash
# Only needs Redis running (spawns supervisor and runner automatically)
RUST_LOG=info cargo run --example end_to_end_demo
```
### **Benefits of OpenRPC Client:**
**Type Safety** - Compile-time checking of requests/responses
**Async/Await** - Native Rust async support
**Error Handling** - Proper Result types with detailed errors
**Auto Serialization** - No manual JSON construction
**Documentation** - IntelliSense and type hints
**Maintainability** - Single source of truth for API
### **Architecture:**
```
┌─────────────────┐
│ Example Code │
│ (simple_e2e) │
└────────┬────────┘
┌─────────────────┐
│ OpenRPC Client │
│ (typed API) │
└────────┬────────┘
│ JSON-RPC over HTTP
┌─────────────────┐
│ Supervisor │
│ (Mycelium) │
└────────┬────────┘
│ Redis Queue
┌─────────────────┐
│ OSIS Runner │
│ (Rhai Engine) │
└─────────────────┘
```
### **Job Execution Modes:**
**Blocking (`job.run`):**
- Client waits for result
- Uses `queue_and_wait` internally
- Returns actual result
- Best for: CRUD, queries, short jobs
**Non-Blocking (`job.start`):**
- Client returns immediately
- Job runs in background
- Returns job_id for polling
- Best for: Long jobs, batch processing
### **Files Modified:**
-`client/src/lib.rs` - Updated client methods and response types
-`examples/simple_e2e.rs` - Refactored to use OpenRPC client
-`examples/end_to_end_demo.rs` - Refactored to use OpenRPC client
-`examples/E2E_EXAMPLES.md` - Updated documentation
-`examples/EXAMPLES_SUMMARY.md` - This file
### **Next Steps:**
1. **Add more examples** - Specific use cases (batch jobs, error handling)
2. **Job polling** - Implement `wait_for_job()` helper
3. **WASM support** - Browser-based examples
4. **Signature examples** - Jobs with cryptographic signatures
---
**Status:** ✅ Complete and Production Ready
**Last Updated:** 2025-10-24
**Client Version:** hero-supervisor-openrpc-client 0.1.0

View File

@@ -0,0 +1,182 @@
# Hero Supervisor Examples
This directory contains examples demonstrating the new job API functionality and workflows.
## Examples Overview
### 1. `job_api_examples.rs` - Comprehensive API Demo
Complete demonstration of all new job API methods:
- **Fire-and-forget execution** using `job.run`
- **Asynchronous processing** with `jobs.create`, `job.start`, `job.status`, `job.result`
- **Batch job processing** for multiple jobs
- **Job listing** with `jobs.list`
**Run with:**
```bash
cargo run --example job_api_examples
```
### 2. `simple_job_workflow.rs` - Basic Workflow
Simple example showing the basic job lifecycle:
1. Create job with `jobs.create`
2. Start job with `job.start`
3. Monitor with `job.status`
4. Get result with `job.result`
**Run with:**
```bash
cargo run --example simple_job_workflow
```
### 3. `integration_test.rs` - Integration Tests
Comprehensive integration tests validating:
- Complete job lifecycle
- Immediate job execution
- Job listing functionality
- Authentication error handling
- Nonexistent job operations
**Run with:**
```bash
cargo test --test integration_test
```
## Prerequisites
Before running the examples, ensure:
1. **Redis is running:**
```bash
docker run -d -p 6379:6379 redis:alpine
```
2. **Supervisor is running:**
```bash
./target/debug/supervisor --config examples/supervisor/config.toml
```
3. **Runners are configured** in your config.toml:
```toml
[[actors]]
id = "osis_runner_1"
name = "osis_runner_1"
binary_path = "/path/to/osis_runner"
db_path = "/tmp/osis_db"
redis_url = "redis://localhost:6379"
process_manager = "simple"
```
## API Convention Summary
The examples demonstrate the new job API convention:
### General Operations (`jobs.`)
- `jobs.create` - Create a job without queuing it
- `jobs.list` - List all job IDs in the system
### Specific Operations (`job.`)
- `job.run` - Run a job immediately and return result
- `job.start` - Start a previously created job
- `job.status` - Get current job status (non-blocking)
- `job.result` - Get job result (blocking until complete)
## Workflow Patterns
### Pattern 1: Fire-and-Forget
```rust
let result = client.job_run(secret, job).await?;
match result {
JobResult::Success { success } => println!("Output: {}", success),
JobResult::Error { error } => println!("Error: {}", error),
}
```
### Pattern 2: Asynchronous Processing
```rust
// Create and start
let job_id = client.jobs_create(secret, job).await?;
client.job_start(secret, &job_id).await?;
// Monitor (non-blocking)
loop {
let status = client.job_status(&job_id).await?;
if status.status == "completed" { break; }
sleep(Duration::from_secs(1)).await;
}
// Get result
let result = client.job_result(&job_id).await?;
```
### Pattern 3: Batch Processing
```rust
// Create all jobs
let mut job_ids = Vec::new();
for job_spec in job_specs {
let job_id = client.jobs_create(secret, job_spec).await?;
job_ids.push(job_id);
}
// Start all jobs
for job_id in &job_ids {
client.job_start(secret, job_id).await?;
}
// Collect results
for job_id in &job_ids {
let result = client.job_result(job_id).await?;
// Process result...
}
```
## Error Handling
The examples demonstrate proper error handling for:
- **Authentication errors** - Invalid secrets
- **Job not found errors** - Nonexistent job IDs
- **Connection errors** - Supervisor not available
- **Execution errors** - Job failures
## Authentication
Examples use different secret types:
- **Admin secrets**: Full system access
- **User secrets**: Job operations only (used in examples)
- **Register secrets**: Runner registration only
Configure secrets in your supervisor config:
```toml
admin_secrets = ["admin-secret-123"]
user_secrets = ["user-secret-456"]
register_secrets = ["register-secret-789"]
```
## Troubleshooting
### Common Issues
1. **Connection refused**
- Ensure supervisor is running on localhost:3030
- Check supervisor logs for errors
2. **Authentication failed**
- Verify secret is configured in supervisor
- Check secret type matches operation requirements
3. **Job execution failed**
- Ensure runners are properly configured and running
- Check runner logs for execution errors
- Verify job payload is valid for the target runner
4. **Redis connection failed**
- Ensure Redis is running on localhost:6379
- Check Redis connectivity from supervisor
### Debug Mode
Run examples with debug logging:
```bash
RUST_LOG=debug cargo run --example job_api_examples
```
This will show detailed API calls and responses for troubleshooting.

View File

@@ -0,0 +1,290 @@
//! Comprehensive OpenRPC Example for Hero Supervisor
//!
//! This example demonstrates the complete OpenRPC workflow:
//! 1. Automatically starting a Hero Supervisor with OpenRPC server using escargot
//! 2. Building and using a mock runner binary
//! 3. Connecting with the OpenRPC client
//! 4. Managing runners (add, start, stop, remove)
//! 5. Creating and queuing jobs
//! 6. Monitoring job execution and verifying results
//! 7. Bulk operations and status monitoring
//! 8. Gracefully shutting down the supervisor
//!
//! To run this example:
//! `cargo run --example basic_openrpc_client`
//!
//! This example is completely self-contained and will start/stop the supervisor automatically.
use hero_supervisor_openrpc_client::{
SupervisorClient, RunnerConfig, RunnerType, ProcessManagerType,
JobBuilder
};
use std::time::Duration;
use escargot::CargoBuild;
use std::process::Stdio;
use tokio::time::sleep;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// env_logger::init(); // Commented out to avoid version conflicts
println!("🚀 Comprehensive OpenRPC Example for Hero Supervisor");
println!("====================================================");
// Build the supervisor with OpenRPC feature (force rebuild to avoid escargot caching)
println!("\n🔨 Force rebuilding supervisor with OpenRPC feature...");
// Clear target directory to force fresh build
let _ = std::process::Command::new("cargo")
.arg("clean")
.output();
let supervisor_binary = CargoBuild::new()
.bin("supervisor")
.features("openrpc")
.current_release()
.run()?;
println!("✅ Supervisor binary built successfully");
// Build the mock runner binary
println!("\n🔨 Building mock runner binary...");
let mock_runner_binary = CargoBuild::new()
.example("mock_runner")
.current_release()
.run()?;
println!("✅ Mock runner binary built successfully");
// Start the supervisor process
println!("\n🚀 Starting supervisor with OpenRPC server...");
let mut supervisor_process = supervisor_binary
.command()
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
println!("✅ Supervisor process started (PID: {})", supervisor_process.id());
// Wait for the server to start up
println!("\n⏳ Waiting for OpenRPC server to start...");
sleep(Duration::from_secs(5)).await;
// Create client
let client = SupervisorClient::new("http://127.0.0.1:3030")?;
println!("✅ Client created for: {}", client.server_url());
// Test connectivity with retries
println!("\n🔍 Testing server connectivity...");
let mut connection_attempts = 0;
let max_attempts = 10;
loop {
connection_attempts += 1;
match client.list_runners().await {
Ok(runners) => {
println!("✅ Server is responsive");
println!("📋 Current runners: {:?}", runners);
break;
}
Err(e) if connection_attempts < max_attempts => {
println!("⏳ Attempt {}/{}: Server not ready yet, retrying...", connection_attempts, max_attempts);
sleep(Duration::from_secs(1)).await;
continue;
}
Err(e) => {
eprintln!("❌ Failed to connect to server after {} attempts: {}", max_attempts, e);
// Clean up the supervisor process before returning
let _ = supervisor_process.kill();
return Err(e.into());
}
}
}
// Add a simple runner using the mock runner binary
let config = RunnerConfig {
actor_id: "basic_example_actor".to_string(),
runner_type: RunnerType::OSISRunner,
binary_path: mock_runner_binary.path().to_path_buf(),
db_path: "/tmp/example_db".to_string(),
redis_url: "redis://localhost:6379".to_string(),
};
println!(" Adding runner: {}", config.actor_id);
client.add_runner(config, ProcessManagerType::Simple).await?;
// Start the runner
println!("▶️ Starting runner...");
client.start_runner("basic_example_actor").await?;
// Check status
let status = client.get_runner_status("basic_example_actor").await?;
println!("📊 Runner status: {:?}", status);
// Create and queue multiple jobs to demonstrate functionality
let jobs = vec![
("Hello World", "print('Hello from comprehensive OpenRPC example!');"),
("Math Calculation", "let result = 42 * 2; print(`The answer is: ${result}`);"),
("Current Time", "print('Job executed at: ' + new Date().toISOString());"),
];
let mut job_ids = Vec::new();
for (description, payload) in jobs {
let job = JobBuilder::new()
.caller_id("comprehensive_client")
.context_id("demo")
.payload(payload)
.runner("basic_example_actor")
.executor("rhai")
.timeout(30)
.build()?;
println!("📤 Queuing job '{}': {}", description, job.id);
client.queue_job_to_runner("basic_example_actor", job.clone()).await?;
job_ids.push((job.id, description.to_string()));
// Small delay between jobs
sleep(Duration::from_millis(500)).await;
}
// Demonstrate synchronous job execution using polling approach
// (Note: queue_and_wait OpenRPC method registration needs debugging)
println!("\n🎯 Demonstrating synchronous job execution with result verification...");
let sync_jobs = vec![
("Synchronous Hello", "print('Hello from synchronous execution!');"),
("Synchronous Math", "let result = 123 + 456; print(`Calculation result: ${result}`);"),
("Synchronous Status", "print('Job processed with result verification');"),
];
for (description, payload) in sync_jobs {
let job = JobBuilder::new()
.caller_id("sync_client")
.context_id("sync_demo")
.payload(payload)
.runner("basic_example_actor")
.executor("rhai")
.timeout(30)
.build()?;
println!("🚀 Executing '{}' with result verification...", description);
let job_id = job.id.clone();
// Queue the job
client.queue_job_to_runner("basic_example_actor", job).await?;
// Poll for completion with timeout
let mut attempts = 0;
let max_attempts = 20; // 10 seconds with 500ms intervals
let mut result = None;
while attempts < max_attempts {
match client.get_job_result(&job_id).await {
Ok(Some(job_result)) => {
result = Some(job_result);
break;
}
Ok(None) => {
// Job not finished yet, wait and retry
sleep(Duration::from_millis(500)).await;
attempts += 1;
}
Err(e) => {
println!("⚠️ Error getting result for job {}: {}", job_id, e);
break;
}
}
}
match result {
Some(job_result) => {
println!("✅ Job '{}' completed successfully!", description);
println!(" 📋 Job ID: {}", job_id);
println!(" 📤 Result: {}", job_result);
}
None => {
println!("⏰ Job '{}' did not complete within timeout", description);
}
}
// Small delay between jobs
sleep(Duration::from_millis(500)).await;
}
// Demonstrate bulk operations and status monitoring
println!("\n📊 Demonstrating bulk operations and status monitoring...");
// Get all runner statuses
println!("📋 Getting all runner statuses...");
match client.get_all_runner_status().await {
Ok(statuses) => {
println!("✅ Runner statuses:");
for (runner_id, status) in statuses {
println!(" - {}: {:?}", runner_id, status);
}
}
Err(e) => println!("❌ Failed to get runner statuses: {}", e),
}
// List all runners one more time
println!("\n📋 Final runner list:");
match client.list_runners().await {
Ok(runners) => {
println!("✅ Active runners: {:?}", runners);
}
Err(e) => println!("❌ Failed to list runners: {}", e),
}
// Stop and remove runner
println!("\n⏹️ Stopping runner...");
client.stop_runner("basic_example_actor", false).await?;
println!("🗑️ Removing runner...");
client.remove_runner("basic_example_actor").await?;
// Final verification
println!("\n🔍 Final verification - listing remaining runners...");
match client.list_runners().await {
Ok(runners) => {
if runners.contains(&"basic_example_actor".to_string()) {
println!("⚠️ Runner still present: {:?}", runners);
} else {
println!("✅ Runner successfully removed. Remaining runners: {:?}", runners);
}
}
Err(e) => println!("❌ Failed to verify runner removal: {}", e),
}
// Gracefully shutdown the supervisor process
println!("\n🛑 Shutting down supervisor process...");
match supervisor_process.kill() {
Ok(()) => {
println!("✅ Supervisor process terminated successfully");
// Wait for the process to fully exit
match supervisor_process.wait() {
Ok(status) => println!("✅ Supervisor exited with status: {}", status),
Err(e) => println!("⚠️ Error waiting for supervisor exit: {}", e),
}
}
Err(e) => println!("⚠️ Error terminating supervisor: {}", e),
}
println!("\n🎉 Comprehensive OpenRPC Example Complete!");
println!("==========================================");
println!("✅ Successfully demonstrated:");
println!(" - Automatic supervisor startup with escargot");
println!(" - Mock runner binary integration");
println!(" - OpenRPC client connectivity with retry logic");
println!(" - Runner management (add, start, stop, remove)");
println!(" - Asynchronous job creation and queuing");
println!(" - Synchronous job execution with result polling");
println!(" - Job result verification from Redis job hash");
println!(" - Bulk operations and status monitoring");
println!(" - Graceful cleanup and supervisor shutdown");
println!("\n🎯 The Hero Supervisor OpenRPC integration is fully functional!");
println!("📝 Note: queue_and_wait method implemented but OpenRPC registration needs debugging");
println!("🚀 Both async job queuing and sync result polling patterns work perfectly!");
Ok(())
}

View File

@@ -0,0 +1,278 @@
//! End-to-End Demo: Supervisor + Runner + Client
//!
//! This example demonstrates the complete workflow:
//! 1. Starts a supervisor with Mycelium integration
//! 2. Starts an OSIS runner
//! 3. Uses the supervisor client to run jobs
//! 4. Shows both job.run (blocking) and job.start (non-blocking) modes
//!
//! Prerequisites:
//! - Redis running on localhost:6379
//!
//! Usage:
//! ```bash
//! RUST_LOG=info cargo run --example end_to_end_demo
//! ```
use anyhow::{Result, Context};
use log::{info, error};
use std::process::{Command, Child, Stdio};
use std::time::Duration;
use tokio::time::sleep;
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder};
/// Configuration for the demo
struct DemoConfig {
redis_url: String,
supervisor_port: u16,
runner_id: String,
db_path: String,
}
impl Default for DemoConfig {
fn default() -> Self {
Self {
redis_url: "redis://localhost:6379".to_string(),
supervisor_port: 3030,
runner_id: "example_runner".to_string(),
db_path: "/tmp/example_runner.db".to_string(),
}
}
}
/// Supervisor process wrapper
struct SupervisorProcess {
child: Child,
}
impl SupervisorProcess {
fn start(config: &DemoConfig) -> Result<Self> {
info!("🚀 Starting supervisor on port {}...", config.supervisor_port);
let child = Command::new("cargo")
.args(&[
"run",
"--bin",
"hero-supervisor",
"--",
"--redis-url",
&config.redis_url,
"--port",
&config.supervisor_port.to_string(),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.context("Failed to start supervisor")?;
Ok(Self { child })
}
}
impl Drop for SupervisorProcess {
fn drop(&mut self) {
info!("🛑 Stopping supervisor...");
let _ = self.child.kill();
let _ = self.child.wait();
}
}
/// Runner process wrapper
struct RunnerProcess {
child: Child,
}
impl RunnerProcess {
fn start(config: &DemoConfig) -> Result<Self> {
info!("🤖 Starting OSIS runner '{}'...", config.runner_id);
let child = Command::new("cargo")
.args(&[
"run",
"--bin",
"runner_osis",
"--",
&config.runner_id,
"--db-path",
&config.db_path,
"--redis-url",
&config.redis_url,
])
.env("RUST_LOG", "info")
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.context("Failed to start runner")?;
Ok(Self { child })
}
}
impl Drop for RunnerProcess {
fn drop(&mut self) {
info!("🛑 Stopping runner...");
let _ = self.child.kill();
let _ = self.child.wait();
}
}
/// Helper functions for the demo
async fn register_runner_helper(client: &SupervisorClient, runner_id: &str, secret: &str) -> Result<()> {
info!("📝 Registering runner '{}'...", runner_id);
let queue = format!("hero:q:work:type:osis:group:default:inst:{}", runner_id);
client.register_runner(secret, runner_id, &queue).await?;
info!("✅ Runner registered successfully");
Ok(())
}
async fn run_job_helper(client: &SupervisorClient, job: runner_rust::job::Job, secret: &str, timeout: u64) -> Result<String> {
info!("🚀 Running job {} (blocking)...", job.id);
let response = client.job_run(secret, job, Some(timeout)).await?;
let result = response.result
.ok_or_else(|| anyhow::anyhow!("No result in response"))?;
info!("✅ Job completed with result: {}", result);
Ok(result)
}
async fn start_job_helper(client: &SupervisorClient, job: runner_rust::job::Job, secret: &str) -> Result<String> {
info!("🚀 Starting job {} (non-blocking)...", job.id);
let response = client.job_start(secret, job).await?;
info!("✅ Job queued with ID: {}", response.job_id);
Ok(response.job_id)
}
#[tokio::main]
async fn main() -> Result<()> {
// Initialize logging
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
println!("\n╔════════════════════════════════════════════════════════════╗");
println!("║ End-to-End Demo: Supervisor + Runner + Client ║");
println!("╚════════════════════════════════════════════════════════════╝\n");
let config = DemoConfig::default();
// Step 1: Start supervisor
println!("📋 Step 1: Starting Supervisor");
println!("─────────────────────────────────────────────────────────────");
let _supervisor = SupervisorProcess::start(&config)?;
sleep(Duration::from_secs(3)).await;
println!("✅ Supervisor started on port {}\n", config.supervisor_port);
// Step 2: Start runner
println!("📋 Step 2: Starting OSIS Runner");
println!("─────────────────────────────────────────────────────────────");
let _runner = RunnerProcess::start(&config)?;
sleep(Duration::from_secs(3)).await;
println!("✅ Runner '{}' started\n", config.runner_id);
// Step 3: Create client and register runner
println!("📋 Step 3: Registering Runner with Supervisor");
println!("─────────────────────────────────────────────────────────────");
let client = SupervisorClient::new(&format!("http://localhost:{}", config.supervisor_port))?;
register_runner_helper(&client, &config.runner_id, "admin_secret").await?;
println!("✅ Runner registered\n");
sleep(Duration::from_secs(2)).await;
// Step 4: Run blocking jobs (job.run)
println!("📋 Step 4: Running Blocking Jobs (job.run)");
println!("─────────────────────────────────────────────────────────────");
// Job 1: Simple calculation
println!("\n🔹 Job 1: Simple Calculation");
let job1 = JobBuilder::new()
.caller_id("demo_client")
.context_id("demo_context")
.payload("let result = 2 + 2; to_json(result)")
.runner(&config.runner_id)
.executor("rhai")
.timeout(30)
.build()?;
let result1 = run_job_helper(&client, job1, "admin_secret", 30).await?;
println!(" Result: {}", result1);
// Job 2: String manipulation
println!("\n🔹 Job 2: String Manipulation");
let job2 = JobBuilder::new()
.caller_id("demo_client")
.context_id("demo_context")
.payload(r#"let msg = "Hello from OSIS Runner!"; to_json(msg)"#)
.runner(&config.runner_id)
.executor("rhai")
.timeout(30)
.build()?;
let result2 = run_job_helper(&client, job2, "admin_secret", 30).await?;
println!(" Result: {}", result2);
// Job 3: Array operations
println!("\n🔹 Job 3: Array Operations");
let job3 = JobBuilder::new()
.caller_id("demo_client")
.context_id("demo_context")
.payload(r#"
let numbers = [1, 2, 3, 4, 5];
let sum = 0;
for n in numbers {
sum += n;
}
to_json(#{sum: sum, count: numbers.len()})
"#)
.runner(&config.runner_id)
.executor("rhai")
.timeout(30)
.build()?;
let result3 = run_job_helper(&client, job3, "admin_secret", 30).await?;
println!(" Result: {}", result3);
println!("\n✅ All blocking jobs completed successfully\n");
// Step 5: Start non-blocking jobs (job.start)
println!("📋 Step 5: Starting Non-Blocking Jobs (job.start)");
println!("─────────────────────────────────────────────────────────────");
println!("\n🔹 Job 4: Background Task");
let job4 = JobBuilder::new()
.caller_id("demo_client")
.context_id("demo_context")
.payload(r#"
let result = "Background task completed";
to_json(result)
"#)
.runner(&config.runner_id)
.executor("rhai")
.timeout(30)
.build()?;
let job4_id = start_job_helper(&client, job4, "admin_secret").await?;
println!(" Job ID: {} (running in background)", job4_id);
println!("\n✅ Non-blocking job started\n");
// Step 6: Summary
println!("📋 Step 6: Demo Summary");
println!("─────────────────────────────────────────────────────────────");
println!("✅ Supervisor: Running on port {}", config.supervisor_port);
println!("✅ Runner: '{}' registered and processing jobs", config.runner_id);
println!("✅ Blocking jobs: 3 completed successfully");
println!("✅ Non-blocking jobs: 1 started");
println!("\n🎉 Demo completed successfully!");
// Keep processes running for a bit to see logs
println!("\n⏳ Keeping processes running for 5 seconds...");
sleep(Duration::from_secs(5)).await;
println!("\n🛑 Shutting down...");
Ok(())
}

View File

@@ -0,0 +1,196 @@
//! Integration test for the new job API
//!
//! This test demonstrates the complete job lifecycle and validates
//! that all new API methods work correctly together.
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder, JobResult};
use std::time::Duration;
use tokio::time::sleep;
#[tokio::test]
async fn test_complete_job_lifecycle() -> Result<(), Box<dyn std::error::Error>> {
// Skip test if supervisor is not running
let client = match SupervisorClient::new("http://localhost:3030") {
Ok(c) => c,
Err(_) => {
println!("Skipping integration test - supervisor not available");
return Ok(());
}
};
// Test connection
if client.discover().await.is_err() {
println!("Skipping integration test - supervisor not responding");
return Ok(());
}
let secret = "user-secret-456";
// Test 1: Create job
let job = JobBuilder::new()
.caller_id("integration_test")
.context_id("test_lifecycle")
.payload("echo 'Integration test job'")
.executor("osis")
.runner("osis_runner_1")
.timeout(30)
.build()?;
let job_id = client.jobs_create(secret, job).await?;
assert!(!job_id.is_empty());
// Test 2: Start job
client.job_start(secret, &job_id).await?;
// Test 3: Monitor status
let mut attempts = 0;
let max_attempts = 15; // 15 seconds max
let mut final_status = String::new();
while attempts < max_attempts {
let status = client.job_status(&job_id).await?;
final_status = status.status.clone();
if final_status == "completed" || final_status == "failed" || final_status == "timeout" {
break;
}
attempts += 1;
sleep(Duration::from_secs(1)).await;
}
// Test 4: Get result
let result = client.job_result(&job_id).await?;
match result {
JobResult::Success { success: _ } => {
assert_eq!(final_status, "completed");
},
JobResult::Error { error: _ } => {
assert!(final_status == "failed" || final_status == "timeout");
}
}
Ok(())
}
#[tokio::test]
async fn test_job_run_immediate() -> Result<(), Box<dyn std::error::Error>> {
let client = match SupervisorClient::new("http://localhost:3030") {
Ok(c) => c,
Err(_) => return Ok(()), // Skip if not available
};
if client.discover().await.is_err() {
return Ok(()); // Skip if not responding
}
let secret = "user-secret-456";
let job = JobBuilder::new()
.caller_id("integration_test")
.context_id("test_immediate")
.payload("echo 'Immediate job test'")
.executor("osis")
.runner("osis_runner_1")
.timeout(30)
.build()?;
// Test immediate execution
let result = client.job_run(secret, job).await?;
// Should get either success or error, but not panic
match result {
JobResult::Success { success } => {
assert!(!success.is_empty());
},
JobResult::Error { error } => {
assert!(!error.is_empty());
}
}
Ok(())
}
#[tokio::test]
async fn test_jobs_list() -> Result<(), Box<dyn std::error::Error>> {
let client = match SupervisorClient::new("http://localhost:3030") {
Ok(c) => c,
Err(_) => return Ok(()), // Skip if not available
};
if client.discover().await.is_err() {
return Ok(()); // Skip if not responding
}
// Test listing jobs
let job_ids = client.jobs_list().await?;
// Should return a vector (might be empty)
assert!(job_ids.len() >= 0);
Ok(())
}
#[tokio::test]
async fn test_authentication_errors() -> Result<(), Box<dyn std::error::Error>> {
let client = match SupervisorClient::new("http://localhost:3030") {
Ok(c) => c,
Err(_) => return Ok(()), // Skip if not available
};
if client.discover().await.is_err() {
return Ok(()); // Skip if not responding
}
let invalid_secret = "invalid-secret";
let job = JobBuilder::new()
.caller_id("integration_test")
.context_id("test_auth")
.payload("echo 'Auth test'")
.executor("osis")
.runner("osis_runner_1")
.timeout(30)
.build()?;
// Test that invalid secret fails
let result = client.jobs_create(invalid_secret, job.clone()).await;
assert!(result.is_err());
let result = client.job_run(invalid_secret, job.clone()).await;
assert!(result.is_err());
let result = client.job_start(invalid_secret, "fake-job-id").await;
assert!(result.is_err());
Ok(())
}
#[tokio::test]
async fn test_nonexistent_job_operations() -> Result<(), Box<dyn std::error::Error>> {
let client = match SupervisorClient::new("http://localhost:3030") {
Ok(c) => c,
Err(_) => return Ok(()), // Skip if not available
};
if client.discover().await.is_err() {
return Ok(()); // Skip if not responding
}
let fake_job_id = "nonexistent-job-id";
// Test operations on nonexistent job
let result = client.job_status(fake_job_id).await;
assert!(result.is_err());
let result = client.job_result(fake_job_id).await;
assert!(result.is_err());
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("Integration test example - this would contain test logic");
Ok(())
}

View File

@@ -0,0 +1,269 @@
//! Examples demonstrating the new job API workflows
//!
//! This example shows how to use the new job API methods:
//! - jobs.create: Create a job without queuing
//! - jobs.list: List all jobs
//! - job.run: Run a job and get result immediately
//! - job.start: Start a created job
//! - job.status: Get job status (non-blocking)
//! - job.result: Get job result (blocking)
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder, JobResult};
use std::time::Duration;
use tokio::time::sleep;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize logging
env_logger::init();
println!("🚀 Hero Supervisor Job API Examples");
println!("===================================\n");
// Create client
let client = SupervisorClient::new("http://localhost:3030")?;
let secret = "user-secret-456"; // Use a user secret for job operations
// Test connection
println!("📡 Testing connection...");
match client.discover().await {
Ok(_) => println!("✅ Connected to supervisor\n"),
Err(e) => {
println!("❌ Failed to connect: {}", e);
println!("Make sure the supervisor is running with: ./supervisor --config examples/supervisor/config.toml\n");
return Ok(());
}
}
// Example 1: Fire-and-forget job execution
println!("🔥 Example 1: Fire-and-forget job execution");
println!("--------------------------------------------");
let job = JobBuilder::new()
.caller_id("example_client")
.context_id("fire_and_forget")
.payload("echo 'Hello from fire-and-forget job!'")
.executor("osis")
.runner("osis_runner_1")
.timeout(30)
.build()?;
println!("Running job immediately...");
match client.job_run(secret, job).await {
Ok(JobResult::Success { success }) => {
println!("✅ Job completed successfully:");
println!(" Output: {}", success);
},
Ok(JobResult::Error { error }) => {
println!("❌ Job failed:");
println!(" Error: {}", error);
},
Err(e) => {
println!("❌ API call failed: {}", e);
}
}
println!();
// Example 2: Asynchronous job processing
println!("⏰ Example 2: Asynchronous job processing");
println!("------------------------------------------");
let job = JobBuilder::new()
.caller_id("example_client")
.context_id("async_processing")
.payload("sleep 2 && echo 'Hello from async job!'")
.executor("osis")
.runner("osis_runner_1")
.timeout(60)
.build()?;
// Step 1: Create the job
println!("1. Creating job...");
let job_id = match client.jobs_create(secret, job).await {
Ok(id) => {
println!("✅ Job created with ID: {}", id);
id
},
Err(e) => {
println!("❌ Failed to create job: {}", e);
return Ok(());
}
};
// Step 2: Start the job
println!("2. Starting job...");
match client.job_start(secret, &job_id).await {
Ok(_) => println!("✅ Job started"),
Err(e) => {
println!("❌ Failed to start job: {}", e);
return Ok(());
}
}
// Step 3: Poll for completion (non-blocking)
println!("3. Monitoring job progress...");
let mut attempts = 0;
let max_attempts = 30; // 30 seconds max
loop {
attempts += 1;
match client.job_status(&job_id).await {
Ok(status) => {
println!(" Status: {} (attempt {})", status.status, attempts);
if status.status == "completed" || status.status == "failed" || status.status == "timeout" {
break;
}
if attempts >= max_attempts {
println!(" ⏰ Timeout waiting for job completion");
break;
}
sleep(Duration::from_secs(1)).await;
},
Err(e) => {
println!(" ❌ Failed to get job status: {}", e);
break;
}
}
}
// Step 4: Get the result
println!("4. Getting job result...");
match client.job_result(&job_id).await {
Ok(JobResult::Success { success }) => {
println!("✅ Job completed successfully:");
println!(" Output: {}", success);
},
Ok(JobResult::Error { error }) => {
println!("❌ Job failed:");
println!(" Error: {}", error);
},
Err(e) => {
println!("❌ Failed to get job result: {}", e);
}
}
println!();
// Example 3: Batch job processing
println!("📦 Example 3: Batch job processing");
println!("-----------------------------------");
let job_specs = vec![
("echo 'Batch job 1'", "batch_1"),
("echo 'Batch job 2'", "batch_2"),
("echo 'Batch job 3'", "batch_3"),
];
let mut job_ids = Vec::new();
// Create all jobs
println!("Creating batch jobs...");
for (i, (payload, context)) in job_specs.iter().enumerate() {
let job = JobBuilder::new()
.caller_id("example_client")
.context_id(context)
.payload(payload)
.executor("osis")
.runner("osis_runner_1")
.timeout(30)
.build()?;
match client.jobs_create(secret, job).await {
Ok(job_id) => {
println!("✅ Created job {}: {}", i + 1, job_id);
job_ids.push(job_id);
},
Err(e) => {
println!("❌ Failed to create job {}: {}", i + 1, e);
}
}
}
// Start all jobs
println!("Starting all batch jobs...");
for (i, job_id) in job_ids.iter().enumerate() {
match client.job_start(secret, job_id).await {
Ok(_) => println!("✅ Started job {}", i + 1),
Err(e) => println!("❌ Failed to start job {}: {}", i + 1, e),
}
}
// Collect results
println!("Collecting results...");
for (i, job_id) in job_ids.iter().enumerate() {
match client.job_result(job_id).await {
Ok(JobResult::Success { success }) => {
println!("✅ Job {} result: {}", i + 1, success);
},
Ok(JobResult::Error { error }) => {
println!("❌ Job {} failed: {}", i + 1, error);
},
Err(e) => {
println!("❌ Failed to get result for job {}: {}", i + 1, e);
}
}
}
println!();
// Example 4: List all jobs
println!("📋 Example 4: Listing all jobs");
println!("-------------------------------");
match client.jobs_list().await {
Ok(jobs) => {
println!("✅ Found {} jobs in the system:", jobs.len());
for (i, job) in jobs.iter().take(10).enumerate() {
println!(" {}. {}", i + 1, job.id);
}
if jobs.len() > 10 {
println!(" ... and {} more", jobs.len() - 10);
}
},
Err(e) => {
println!("❌ Failed to list jobs: {}", e);
}
}
println!();
println!("🎉 All examples completed!");
println!("\nAPI Convention Summary:");
println!("- jobs.create: Create job without queuing");
println!("- jobs.list: List all job IDs");
println!("- job.run: Run job and return result immediately");
println!("- job.start: Start a created job");
println!("- job.status: Get job status (non-blocking)");
println!("- job.result: Get job result (blocking)");
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_job_builder() {
let job = JobBuilder::new()
.caller_id("test")
.context_id("test")
.payload("echo 'test'")
.executor("osis")
.runner("test_runner")
.build();
assert!(job.is_ok());
let job = job.unwrap();
assert_eq!(job.caller_id, "test");
assert_eq!(job.context_id, "test");
assert_eq!(job.payload, "echo 'test'");
}
#[tokio::test]
async fn test_client_creation() {
let client = SupervisorClient::new("http://localhost:3030");
assert!(client.is_ok());
}
}

View File

@@ -0,0 +1,171 @@
//! Mock Runner Binary for Testing OpenRPC Examples
//!
//! This is a simple mock runner that simulates an actor binary for testing
//! the Hero Supervisor OpenRPC integration. It connects to Redis, listens for
//! jobs using the proper Hero job queue system, and echoes the job payload.
//!
//! Usage:
//! ```bash
//! cargo run --example mock_runner -- --actor-id test_actor --db-path /tmp/test_db --redis-url redis://localhost:6379
//! ```
use std::env;
use std::time::Duration;
use tokio::time::sleep;
use redis::AsyncCommands;
use hero_supervisor::{
Job, JobStatus, JobError, Client, ClientBuilder
};
#[derive(Debug, Clone)]
pub struct MockRunnerConfig {
pub actor_id: String,
pub db_path: String,
pub redis_url: String,
}
impl MockRunnerConfig {
pub fn from_args() -> Result<Self, Box<dyn std::error::Error>> {
let args: Vec<String> = env::args().collect();
let mut actor_id = None;
let mut db_path = None;
let mut redis_url = None;
let mut i = 1;
while i < args.len() {
match args[i].as_str() {
"--actor-id" => {
if i + 1 < args.len() {
actor_id = Some(args[i + 1].clone());
i += 2;
} else {
return Err("Missing value for --actor-id".into());
}
}
"--db-path" => {
if i + 1 < args.len() {
db_path = Some(args[i + 1].clone());
i += 2;
} else {
return Err("Missing value for --db-path".into());
}
}
"--redis-url" => {
if i + 1 < args.len() {
redis_url = Some(args[i + 1].clone());
i += 2;
} else {
return Err("Missing value for --redis-url".into());
}
}
_ => i += 1,
}
}
Ok(MockRunnerConfig {
actor_id: actor_id.ok_or("Missing required --actor-id argument")?,
db_path: db_path.ok_or("Missing required --db-path argument")?,
redis_url: redis_url.unwrap_or_else(|| "redis://localhost:6379".to_string()),
})
}
}
pub struct MockRunner {
config: MockRunnerConfig,
client: Client,
}
impl MockRunner {
pub async fn new(config: MockRunnerConfig) -> Result<Self, Box<dyn std::error::Error>> {
let client = ClientBuilder::new()
.redis_url(&config.redis_url)
.build()
.await?;
Ok(MockRunner {
config,
client,
})
}
pub async fn run(&self) -> Result<(), Box<dyn std::error::Error>> {
println!("🤖 Mock Runner '{}' starting...", self.config.actor_id);
println!("📂 DB Path: {}", self.config.db_path);
println!("🔗 Redis URL: {}", self.config.redis_url);
// Use the proper Hero job queue key for this actor instance
// Format: hero:q:work:type:{job_type}:group:{group}:inst:{instance}
let work_queue_key = format!("hero:q:work:type:osis:group:default:inst:{}", self.config.actor_id);
println!("👂 Listening for jobs on queue: {}", work_queue_key);
loop {
// Try to pop a job ID from the work queue using the Hero protocol
let job_id = self.client.get_job_id(&work_queue_key).await?;
match job_id {
Some(job_id) => {
println!("📨 Received job ID: {}", job_id);
if let Err(e) = self.process_job(&job_id).await {
eprintln!("❌ Error processing job {}: {}", job_id, e);
// Mark job as error
if let Err(e2) = self.client.set_job_status(&job_id, JobStatus::Error).await {
eprintln!("❌ Failed to set job error status: {}", e2);
}
}
}
None => {
// No jobs available, wait a bit
sleep(Duration::from_millis(100)).await;
}
}
}
}
async fn process_job(&self, job_id: &str) -> Result<(), JobError> {
// Load the job from Redis using the Hero job system
let job = self.client.get_job(job_id).await?;
self.process_job_internal(&self.client, job_id, &job).await
}
async fn process_job_internal(
&self,
client: &Client,
job_id: &str,
job: &Job,
) -> Result<(), JobError> {
println!("🔄 Processing job {} with payload: {}", job_id, job.payload);
// Mark job as started
client.set_job_status(job_id, JobStatus::Started).await?;
println!("🚀 Job {} marked as Started", job_id);
// Simulate processing time
sleep(Duration::from_millis(500)).await;
// Echo the payload (simulate job execution)
let output = format!("echo: {}", job.payload);
println!("📤 Output: {}", output);
// Set the job result
client.set_result(job_id, &output).await?;
println!("✅ Job {} completed successfully", job_id);
Ok(())
}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Parse command line arguments
let config = MockRunnerConfig::from_args()?;
// Create and run the mock runner
let runner = MockRunner::new(config).await?;
runner.run().await?;
Ok(())
}

View File

@@ -0,0 +1,203 @@
//! Simple End-to-End Example
//!
//! A minimal example showing supervisor + runner + client workflow.
//!
//! Prerequisites:
//! - Redis running on localhost:6379
//!
//! Usage:
//! ```bash
//! # Terminal 1: Start Redis
//! redis-server
//!
//! # Terminal 2: Run this example
//! RUST_LOG=info cargo run --example simple_e2e
//! ```
use anyhow::Result;
use log::info;
use std::time::Duration;
use tokio::time::sleep;
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder};
#[tokio::main]
async fn main() -> Result<()> {
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
println!("\n╔════════════════════════════════════════╗");
println!("║ Simple End-to-End Demo ║");
println!("╚════════════════════════════════════════╝\n");
let supervisor_url = "http://localhost:3030";
let runner_id = "test_runner";
let secret = "admin_secret";
// Create supervisor client
let client = SupervisorClient::new(supervisor_url)?;
println!("📝 Prerequisites:");
println!(" 1. Redis running on localhost:6379");
println!(" 2. Supervisor running on {}", supervisor_url);
println!(" 3. Runner '{}' registered and running\n", runner_id);
println!("💡 To start the supervisor:");
println!(" cargo run --bin hero-supervisor -- --redis-url redis://localhost:6379\n");
println!("💡 To start a runner:");
println!(" cd /Users/timurgordon/code/git.ourworld.tf/herocode/runner_rust");
println!(" cargo run --bin runner_osis -- {} --redis-url redis://localhost:6379\n", runner_id);
println!("⏳ Waiting 3 seconds for you to start the prerequisites...\n");
sleep(Duration::from_secs(3)).await;
// Register runner
println!("📋 Step 1: Registering Runner");
println!("─────────────────────────────────────────");
let queue = format!("hero:q:work:type:osis:group:default:inst:{}", runner_id);
match client.register_runner(secret, runner_id, &queue).await {
Ok(_) => {
println!("✅ Runner registered successfully");
}
Err(e) => {
println!("⚠️ Registration error: {} (runner might already be registered)", e);
}
}
sleep(Duration::from_secs(1)).await;
// Run a simple job
println!("\n📋 Step 2: Running a Simple Job (Blocking)");
println!("─────────────────────────────────────────");
let job = JobBuilder::new()
.caller_id("simple_demo")
.context_id("demo_context")
.payload(r#"
let message = "Hello from the runner!";
let number = 42;
to_json(#{
message: message,
number: number,
timestamp: timestamp()
})
"#)
.runner(runner_id)
.executor("rhai")
.timeout(30)
.build()?;
let job_id = job.id.clone();
info!("Sending job with ID: {}", job_id);
match client.job_run(secret, job, Some(30)).await {
Ok(response) => {
println!("✅ Job completed!");
if let Some(result) = response.result {
println!(" Result: {}", result);
}
}
Err(e) => {
println!("❌ Job failed: {}", e);
return Ok(());
}
}
// Run another job (calculation)
println!("\n📋 Step 3: Running a Calculation Job");
println!("─────────────────────────────────────────");
let calc_job = JobBuilder::new()
.caller_id("simple_demo")
.context_id("demo_context")
.payload(r#"
let numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
let sum = 0;
let product = 1;
for n in numbers {
sum += n;
product *= n;
}
to_json(#{
sum: sum,
product: product,
count: numbers.len(),
average: sum / numbers.len()
})
"#)
.runner(runner_id)
.executor("rhai")
.timeout(30)
.build()?;
let calc_job_id = calc_job.id.clone();
info!("Sending calculation job with ID: {}", calc_job_id);
match client.job_run(secret, calc_job, Some(30)).await {
Ok(response) => {
println!("✅ Calculation completed!");
if let Some(result) = response.result {
println!(" Result: {}", result);
}
}
Err(e) => {
println!("❌ Calculation failed: {}", e);
}
}
// Start a non-blocking job
println!("\n📋 Step 4: Starting a Non-Blocking Job");
println!("─────────────────────────────────────────");
let async_job = JobBuilder::new()
.caller_id("simple_demo")
.context_id("demo_context")
.payload(r#"
let result = "This job was started asynchronously";
to_json(result)
"#)
.runner(runner_id)
.executor("rhai")
.timeout(30)
.build()?;
let async_job_id = async_job.id.clone();
info!("Starting async job with ID: {}", async_job_id);
match client.job_start(secret, async_job).await {
Ok(response) => {
println!("✅ Job started!");
println!(" Job ID: {} (running in background)", response.job_id);
println!(" Status: {}", response.status);
}
Err(e) => {
println!("❌ Failed to start job: {}", e);
}
}
// Summary
println!("\n╔════════════════════════════════════════╗");
println!("║ Demo Summary ║");
println!("╚════════════════════════════════════════╝");
println!("✅ Runner registered: {}", runner_id);
println!("✅ Blocking jobs completed: 2");
println!("✅ Non-blocking jobs started: 1");
println!("\n🎉 Demo completed successfully!\n");
println!("📚 What happened:");
println!(" 1. Registered a runner with the supervisor");
println!(" 2. Sent jobs with Rhai scripts to execute");
println!(" 3. Supervisor queued jobs to the runner");
println!(" 4. Runner executed the scripts and returned results");
println!(" 5. Client received results (for blocking jobs)\n");
println!("🔍 Key Concepts:");
println!(" • job.run = Execute and wait for result (blocking)");
println!(" • job.start = Start and return immediately (non-blocking)");
println!(" • Jobs contain Rhai scripts that run on the runner");
println!(" • Supervisor coordinates job distribution via Redis\n");
Ok(())
}

View File

@@ -0,0 +1,64 @@
//! Simple job workflow example
//!
//! This example demonstrates the basic job lifecycle using the new API:
//! 1. Create a job
//! 2. Start the job
//! 3. Monitor its progress
//! 4. Get the result
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder, JobResult};
use std::time::Duration;
use tokio::time::sleep;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("Simple Job Workflow Example");
println!("============================\n");
// Create client
let client = SupervisorClient::new("http://localhost:3030")?;
let secret = "user-secret-456";
// Create a simple job
let job = JobBuilder::new()
.caller_id("simple_example")
.context_id("demo")
.payload("echo 'Hello from Hero Supervisor!' && sleep 3 && echo 'Job completed!'")
.executor("osis")
.runner("osis_runner_1")
.timeout(60)
.env_var("EXAMPLE_VAR", "example_value")
.build()?;
println!("📝 Creating job...");
let job_id = client.jobs_create(secret, job).await?;
println!("✅ Job created: {}\n", job_id);
println!("🚀 Starting job...");
client.job_start(secret, &job_id).await?;
println!("✅ Job started\n");
println!("👀 Monitoring job progress...");
loop {
let status = client.job_status(&job_id).await?;
println!(" Status: {}", status.status);
if status.status == "completed" || status.status == "failed" {
break;
}
sleep(Duration::from_secs(2)).await;
}
println!("\n📋 Getting job result...");
match client.job_result(&job_id).await? {
JobResult::Success { success } => {
println!("✅ Success: {}", success);
},
JobResult::Error { error } => {
println!("❌ Error: {}", error);
}
}
Ok(())
}

View File

@@ -0,0 +1,108 @@
# Hero Supervisor Example
This example demonstrates how to configure and run the Hero Supervisor with multiple actors using a TOML configuration file.
## Files
- `config.toml` - Example supervisor configuration with multiple actors
- `run_supervisor.sh` - Shell script to build and run the supervisor with the example config
- `run_supervisor.rs` - Rust script using escargot to build and run the supervisor
- `README.md` - This documentation file
## Configuration
The `config.toml` file defines:
- **Redis connection**: URL for the Redis server used for job queuing
- **Database path**: Local path for supervisor state storage
- **Job queue key**: Redis key for the supervisor job queue
- **Actors**: List of actor configurations with:
- `name`: Unique identifier for the actor
- `runner_type`: Type of runner ("SAL", "OSIS", "V", "Python")
- `binary_path`: Path to the actor binary
- `process_manager`: Process management type ("simple" or "tmux")
## Prerequisites
1. **Redis Server**: Ensure Redis is running on `localhost:6379` (or update the config)
2. **Actor Binaries**: Build the required actor binaries referenced in the config:
```bash
# Build SAL worker
cd ../../sal
cargo build --bin sal_worker
# Build OSIS and system workers
cd ../../worker
cargo build --bin osis
cargo build --bin system
```
## Running the Example
### Option 1: Shell Script (Recommended)
```bash
./run_supervisor.sh
```
### Option 2: Rust Script with Escargot
```bash
cargo +nightly -Zscript run_supervisor.rs
```
### Option 3: Manual Build and Run
```bash
# Build the supervisor
cd ../../../supervisor
cargo build --bin supervisor --features cli
# Run with config
./target/debug/supervisor --config ../baobab/examples/supervisor/config.toml
```
## Usage
Once running, the supervisor will:
1. Load the configuration from `config.toml`
2. Initialize and start all configured actors
3. Listen for jobs on the Redis queue (`hero:supervisor:jobs`)
4. Dispatch jobs to appropriate actors based on the `runner` field
5. Monitor actor health and status
## Testing
You can test the supervisor by dispatching jobs to the Redis queue:
```bash
# Using redis-cli to add a test job
redis-cli LPUSH "hero:supervisor:jobs" '{"id":"test-123","runner":"sal_actor_1","script":"print(\"Hello from SAL actor!\")"}'
```
## Stopping
Use `Ctrl+C` to gracefully shutdown the supervisor. It will:
1. Stop accepting new jobs
2. Wait for running jobs to complete
3. Shutdown all managed actors
4. Clean up resources
## Customization
Modify `config.toml` to:
- Add more actors
- Change binary paths to match your build locations
- Update Redis connection settings
- Configure different process managers per actor
- Adjust database and queue settings
## Troubleshooting
- **Redis Connection**: Ensure Redis is running and accessible
- **Binary Paths**: Verify all actor binary paths exist and are executable
- **Permissions**: Ensure the supervisor has permission to create the database directory
- **Ports**: Check that Redis port (6379) is not blocked by firewall

View File

@@ -0,0 +1,18 @@
# Hero Supervisor Configuration
# This configuration defines the Redis connection, database path, and actors to manage
# Redis connection URL
redis_url = "redis://localhost:6379"
# Database path for supervisor state
db_path = "/tmp/supervisor_example_db"
# Job queue key for supervisor jobs
job_queue_key = "hero:supervisor:jobs"
# Actor configurations
[[actors]]
name = "sal_actor_1"
runner_type = "SAL"
binary_path = "cargo run /Users/timurgordon/code/git.ourworld.tf/herocode/supervisor/examples/mock_runner.rs"
process_manager = "tmux"

View File

@@ -0,0 +1,70 @@
#!/usr/bin/env cargo +nightly -Zscript
//! ```cargo
//! [dependencies]
//! escargot = "0.5"
//! tokio = { version = "1.0", features = ["full"] }
//! log = "0.4"
//! env_logger = "0.10"
//! ```
use escargot::CargoBuild;
use std::process::Command;
use log::{info, error};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize logging
env_logger::init();
info!("Building and running Hero Supervisor with example configuration");
// Get the current directory (when running as cargo example, this is the crate root)
let current_dir = std::env::current_dir()?;
info!("Current directory: {}", current_dir.display());
// Path to the supervisor crate (current directory when running as example)
let supervisor_crate_path = current_dir.clone();
// Path to the config file (in examples/supervisor subdirectory)
let config_path = current_dir.join("examples/supervisor/config.toml");
if !config_path.exists() {
error!("Config file not found: {}", config_path.display());
return Err("Config file not found".into());
}
info!("Using config file: {}", config_path.display());
// Build the supervisor binary using escargot
info!("Building supervisor binary...");
let supervisor_bin = CargoBuild::new()
.bin("supervisor")
.manifest_path(supervisor_crate_path.join("Cargo.toml"))
.features("cli")
.run()?;
info!("Supervisor binary built successfully");
// Run the supervisor with the config file
info!("Starting supervisor with config: {}", config_path.display());
let mut cmd = Command::new(supervisor_bin.path());
cmd.arg("--config")
.arg(&config_path);
// Add environment variables for better logging
cmd.env("RUST_LOG", "info");
info!("Executing: {:?}", cmd);
// Execute the supervisor
let status = cmd.status()?;
if status.success() {
info!("Supervisor completed successfully");
} else {
error!("Supervisor exited with status: {}", status);
}
Ok(())
}

View File

@@ -0,0 +1,52 @@
#!/bin/bash
# Hero Supervisor Example Runner
# This script builds and runs the supervisor binary with the example configuration
set -e
# Get the directory of this script
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SUPERVISOR_DIR="$SCRIPT_DIR/../../../supervisor"
CONFIG_FILE="$SCRIPT_DIR/config.toml"
echo "🚀 Building and running Hero Supervisor with example configuration"
echo "📁 Script directory: $SCRIPT_DIR"
echo "🔧 Supervisor crate: $SUPERVISOR_DIR"
echo "⚙️ Config file: $CONFIG_FILE"
# Check if config file exists
if [ ! -f "$CONFIG_FILE" ]; then
echo "❌ Config file not found: $CONFIG_FILE"
exit 1
fi
# Check if supervisor directory exists
if [ ! -d "$SUPERVISOR_DIR" ]; then
echo "❌ Supervisor directory not found: $SUPERVISOR_DIR"
exit 1
fi
# Build the supervisor binary
echo "🔨 Building supervisor binary..."
cd "$SUPERVISOR_DIR"
cargo build --bin supervisor --features cli
# Check if build was successful
if [ $? -ne 0 ]; then
echo "❌ Failed to build supervisor binary"
exit 1
fi
echo "✅ Supervisor binary built successfully"
# Run the supervisor with the config file
echo "🎯 Starting supervisor with config: $CONFIG_FILE"
echo "📝 Use Ctrl+C to stop the supervisor"
echo ""
# Set environment variables for better logging
export RUST_LOG=info
# Execute the supervisor
exec "$SUPERVISOR_DIR/target/debug/supervisor" --config "$CONFIG_FILE"

View File

@@ -0,0 +1,65 @@
/// Generate test secp256k1 keypairs for supervisor authentication testing
///
/// Run with: cargo run --example generate_keypairs
use secp256k1::{Secp256k1, SecretKey, PublicKey};
use hex;
fn main() {
let secp = Secp256k1::new();
println!("# Test Keypairs for Supervisor Auth\n");
println!("These are secp256k1 keypairs for testing the supervisor authentication system.\n");
println!("⚠️ WARNING: These are TEST keypairs only! Never use these in production!\n");
// Generate 5 keypairs with simple private keys for testing
let test_keys = vec![
("Alice (Admin)", "1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"),
("Bob (User)", "fedcba0987654321fedcba0987654321fedcba0987654321fedcba0987654321"),
("Charlie (Register)", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
("Dave (Test)", "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
("Eve (Test)", "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"),
];
for (i, (name, privkey_hex)) in test_keys.iter().enumerate() {
println!("## Keypair {} ({})", i + 1, name);
println!("```");
// Parse private key
let privkey_bytes = hex::decode(privkey_hex).expect("Invalid hex");
let secret_key = SecretKey::from_slice(&privkey_bytes).expect("Invalid private key");
// Derive public key
let public_key = PublicKey::from_secret_key(&secp, &secret_key);
// Serialize keys
let pubkey_hex = hex::encode(public_key.serialize_uncompressed());
println!("Private Key: 0x{}", privkey_hex);
println!("Public Key: 0x{}", pubkey_hex);
println!("```\n");
}
println!("\n## Usage Examples\n");
println!("### Using with OpenRPC Client\n");
println!("```rust");
println!("use secp256k1::{{Secp256k1, SecretKey}};");
println!("use hex;");
println!();
println!("// Alice's private key");
println!("let alice_privkey = SecretKey::from_slice(");
println!(" &hex::decode(\"1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef\").unwrap()");
println!(").unwrap();");
println!();
println!("// Create client with signature");
println!("let client = WasmSupervisorClient::new_with_keypair(");
println!(" \"http://127.0.0.1:3030\",");
println!(" alice_privkey");
println!(");");
println!("```\n");
println!("### Testing Different Scopes\n");
println!("1. **Admin Scope** - Use Alice's keypair for full admin access");
println!("2. **User Scope** - Use Bob's keypair for limited user access");
println!("3. **Register Scope** - Use Charlie's keypair for runner registration only\n");
}

View File

@@ -0,0 +1,102 @@
# OSIRIS + OpenRPC Comprehensive Example
This example demonstrates the complete workflow of using Hero Supervisor with OSIRIS runners via OpenRPC.
## What This Example Does
1. **Builds and starts** Hero Supervisor with OpenRPC server enabled
2. **Builds** the OSIRIS runner binary
3. **Connects** an OpenRPC client to the supervisor
4. **Registers and starts** an OSIRIS runner
5. **Dispatches multiple jobs** via OpenRPC:
- Create a Note
- Create an Event
- Query stored data
- Test access control (expected to fail)
6. **Monitors** job execution and results
7. **Gracefully shuts down** all components
## Prerequisites
**IMPORTANT: Redis must be running before starting this example!**
```bash
# Start Redis (if not already running)
redis-server
```
Other requirements:
- Redis server running on `localhost:6379`
- Rust toolchain installed
- Both `supervisor` and `runner_rust` crates available
## Running the Example
```bash
cargo run --example osiris_openrpc
```
## Job Scripts
The example uses separate Rhai script files for each job:
- `note.rhai` - Creates and stores a Note object
- `event.rhai` - Creates and stores an Event object
- `query.rhai` - Queries and retrieves stored objects
- `access_denied.rhai` - Tests access control (should fail)
## Architecture
```
┌─────────────────┐
│ This Example │
│ (OpenRPC │
│ Client) │
└────────┬────────┘
│ JSON-RPC
┌─────────────────┐
│ Supervisor │
│ (OpenRPC │
│ Server) │
└────────┬────────┘
│ Redis Queue
┌─────────────────┐
│ OSIRIS Runner │
│ (Rhai Engine │
│ + HeroDB) │
└─────────────────┘
```
## Key Features Demonstrated
- **Automatic binary building** using escargot
- **OpenRPC communication** between client and supervisor
- **Runner registration** with configuration
- **Job dispatching** with signatories
- **Context-based access control** in OSIRIS
- **Typed object storage** (Note, Event)
- **Graceful shutdown** and cleanup
## Expected Output
The example will:
1. ✅ Create a Note successfully
2. ✅ Create an Event successfully
3. ✅ Query and retrieve stored objects
4. ✅ Deny access for unauthorized participants
5. ✅ Clean up all resources
## Troubleshooting
**Redis Connection Error:**
- Ensure Redis is running: `redis-server`
**Build Errors:**
- Ensure both supervisor and runner_rust crates are available
- Check that all dependencies are up to date
**OpenRPC Connection Error:**
- Port 3030 might be in use
- Check supervisor logs for startup issues

View File

@@ -0,0 +1,8 @@
print("Attempting to access context with non-signatories...");
print("Participants: [dave, eve]");
print("Signatories: [alice, bob, charlie]");
// This should fail because neither dave nor eve are signatories
let ctx = get_context(["dave", "eve"]);
"This should not succeed!"

View File

@@ -0,0 +1,18 @@
print("Creating context for [alice, bob]...");
let ctx = get_context(["alice", "bob"]);
print("✓ Context ID: " + ctx.context_id());
print("\nCreating event...");
let event = event("events")
.title("Team Retrospective")
.description("Review what went well and areas for improvement")
.location("Virtual - Zoom Room A")
.category("retrospective");
print("✓ Event created");
print("\nStoring event in context...");
ctx.save(event);
print("✓ Event stored");
"Event 'Team Retrospective' created and stored successfully"

View File

@@ -0,0 +1,293 @@
///! Comprehensive OSIRIS + OpenRPC + Admin UI Example
///!
/// This example demonstrates using the Hero Supervisor OpenRPC client
/// to run OSIRIS scripts through the supervisor.
///
/// The client library is located at: client/
///!
///! 1. Starting a Hero Supervisor with OpenRPC server
///! 2. Building and serving the Admin UI (Yew WASM)
///! 3. Building and starting an OSIRIS runner
///! 4. Registering the runner with the supervisor
///! 5. Dispatching multiple OSIRIS jobs via OpenRPC
///! 6. Monitoring job execution via CLI and Web UI
///! 7. Graceful shutdown
///!
///! Services:
///! - Supervisor OpenRPC API: http://127.0.0.1:3030
///! - Admin UI: http://127.0.0.1:8080
///!
///! Usage:
///! ```bash
///! cargo run --example osiris_openrpc
///! ```
///!
///! Requirements:
///! - Redis running on localhost:6379
///! - Trunk installed (cargo install trunk)
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder};
use std::time::Duration;
use escargot::CargoBuild;
use std::process::{Stdio, Command};
use tokio::time::sleep;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("🚀 OSIRIS + OpenRPC Comprehensive Example");
println!("=========================================\n");
// ========================================================================
// STEP 1: Build and start supervisor with OpenRPC
// ========================================================================
println!("Step 1: Building and starting supervisor");
println!("─────────────────────────────────────────────────────────────\n");
let supervisor_binary = CargoBuild::new()
.bin("supervisor")
.current_release()
.manifest_path("../supervisor/Cargo.toml")
.run()?;
println!("✅ Supervisor binary built");
let mut supervisor = supervisor_binary.command()
.arg("--redis-url")
.arg("redis://localhost:6379")
.arg("--port")
.arg("3030")
.arg("--admin-secret")
.arg("admin_secret")
.arg("--user-secret")
.arg("user_secret")
.stdout(Stdio::inherit())
.stderr(Stdio::inherit())
.spawn()?;
println!("✅ Supervisor started on port 3030");
println!("⏳ Waiting for supervisor to initialize...");
sleep(Duration::from_secs(5)).await;
// Check if supervisor is still running
match supervisor.try_wait()? {
Some(status) => {
return Err(format!("Supervisor exited early with status: {}", status).into());
}
None => {
println!("✅ Supervisor is running");
}
}
// ========================================================================
// STEP 2: Build and serve Admin UI
// ========================================================================
println!("\nStep 2: Building and serving Admin UI");
println!("─────────────────────────────────────────────────────────────\n");
let mut admin_ui = Command::new("trunk")
.arg("serve")
.arg("--port")
.arg("8080")
.arg("--address")
.arg("127.0.0.1")
.current_dir("ui")
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()?;
println!("✅ Admin UI building...");
println!("🌐 Admin UI will be available at: http://127.0.0.1:8080");
sleep(Duration::from_secs(3)).await;
// ========================================================================
// STEP 3: Build OSIRIS runner
// ========================================================================
println!("\nStep 3: Building OSIRIS runner");
println!("─────────────────────────────────────────────────────────────\n");
let runner_binary = CargoBuild::new()
.bin("runner_osiris")
.current_release()
.manifest_path("../runner_rust/Cargo.toml")
.run()?;
println!("✅ OSIRIS runner binary built");
// ========================================================================
// STEP 4: Connect OpenRPC client
// ========================================================================
println!("\nStep 4: Connecting OpenRPC client");
println!("─────────────────────────────────────────────────────────────\n");
let client = SupervisorClient::new("http://127.0.0.1:3030")?;
println!("✅ Connected to supervisor\n");
// ========================================================================
// STEP 5: Register and start OSIRIS runner
// ========================================================================
println!("Step 5: Registering OSIRIS runner");
println!("─────────────────────────────────────────────────────────────\n");
let runner_path = runner_binary.path().to_string_lossy();
let db_path = "/tmp/osiris_openrpc.db";
// Register the runner with the supervisor
// Note: The current OpenRPC server uses register_runner, not add_runner
client.register_runner("admin_secret", "osiris_runner").await?;
println!("✅ Runner registered: osiris_runner");
client.start_runner("admin_secret", "osiris_runner").await?;
println!("✅ Runner started\n");
sleep(Duration::from_secs(2)).await;
// ========================================================================
// STEP 6: Load job scripts
// ========================================================================
println!("Step 6: Loading job scripts");
println!("─────────────────────────────────────────────────────────────\n");
let note_script = std::fs::read_to_string("examples/osiris_openrpc/note.rhai")?;
let event_script = std::fs::read_to_string("examples/osiris_openrpc/event.rhai")?;
let query_script = std::fs::read_to_string("examples/osiris_openrpc/query.rhai")?;
let access_denied_script = std::fs::read_to_string("examples/osiris_openrpc/access_denied.rhai")?;
println!("✅ Loaded 4 job scripts\n");
// ========================================================================
// STEP 7: Dispatch jobs via OpenRPC
// ========================================================================
println!("Step 7: Dispatching jobs");
println!("─────────────────────────────────────────────────────────────\n");
// Job 1: Create Note
println!("📝 Job 1: Creating Note...");
let job1 = JobBuilder::new()
.caller_id("openrpc_client")
.context_id("osiris_demo")
.payload(&note_script)
.runner("osiris_runner")
.executor("rhai")
.timeout(30)
.signature("alice", "")
.signature("bob", "")
.build()?;
let job1_result = client.run_job("user_secret", job1).await;
match job1_result {
Ok(result) => println!("{:?}\n", result),
Err(e) => println!("❌ Job failed: {}\n", e),
}
sleep(Duration::from_secs(1)).await;
// Job 2: Create Event
println!("📅 Job 2: Creating Event...");
let job2 = JobBuilder::new()
.caller_id("openrpc_client")
.context_id("osiris_demo")
.payload(&event_script)
.runner("osiris_runner")
.executor("rhai")
.timeout(30)
.signature("alice", "")
.signature("bob", "")
.build()?;
let job2_result = client.run_job("user_secret", job2).await;
match job2_result {
Ok(result) => println!("{:?}\n", result),
Err(e) => println!("❌ Job failed: {}\n", e),
}
sleep(Duration::from_secs(1)).await;
// Job 3: Query Data
println!("🔍 Job 3: Querying Data...");
let job3 = JobBuilder::new()
.caller_id("openrpc_client")
.context_id("osiris_demo")
.payload(&query_script)
.runner("osiris_runner")
.executor("rhai")
.timeout(30)
.signature("alice", "")
.signature("bob", "")
.signature("charlie", "")
.build()?;
let job3_result = client.run_job("user_secret", job3).await;
match job3_result {
Ok(result) => println!("{:?}\n", result),
Err(e) => println!("❌ Job failed: {}\n", e),
}
sleep(Duration::from_secs(1)).await;
// Job 4: Access Control Test (should fail)
println!("🔒 Job 4: Testing Access Control (expected to fail)...");
let job4 = JobBuilder::new()
.caller_id("openrpc_client")
.context_id("osiris_demo")
.payload(&access_denied_script)
.runner("osiris_runner")
.executor("rhai")
.timeout(30)
.signature("alice", "")
.signature("bob", "")
.signature("charlie", "")
.build()?;
let job4_result = client.run_job("user_secret", job4).await;
match job4_result {
Ok(result) => println!("❌ Unexpected success: {:?}\n", result),
Err(e) => println!("✅ Access denied as expected: {}\n", e),
}
// ========================================================================
// STEP 8: Check runner status
// ========================================================================
println!("\nStep 8: Checking runner status");
println!("─────────────────────────────────────────────────────────────\n");
let status = client.get_runner_status("admin_secret", "osiris_runner").await?;
println!("Runner status: {:?}\n", status);
// ========================================================================
// STEP 9: Keep services running for manual testing
// ========================================================================
println!("\nStep 9: Services Running");
println!("─────────────────────────────────────────────────────────────\n");
println!("🌐 Admin UI: http://127.0.0.1:8080");
println!("📡 OpenRPC API: http://127.0.0.1:3030");
println!("\n⏸️ Press Ctrl+C to stop all services...\n");
// Wait for Ctrl+C
tokio::signal::ctrl_c().await?;
// ========================================================================
// STEP 10: Cleanup
// ========================================================================
println!("\n\nStep 10: Cleanup");
println!("─────────────────────────────────────────────────────────────\n");
client.stop_runner("admin_secret", "osiris_runner", false).await?;
println!("✅ Runner stopped");
client.remove_runner("admin_secret", "osiris_runner").await?;
println!("✅ Runner removed");
admin_ui.kill()?;
println!("✅ Admin UI stopped");
supervisor.kill()?;
println!("✅ Supervisor stopped");
println!("\n✨ Example completed successfully!");
Ok(())
}

View File

@@ -0,0 +1,20 @@
print("Creating context for [alice, bob]...");
let ctx = get_context(["alice", "bob"]);
print("✓ Context ID: " + ctx.context_id());
print("\nCreating note...");
let note = note("notes")
.title("Sprint Planning Meeting")
.content("Discussed Q1 2025 roadmap and milestones")
.tag("sprint", "2025-Q1")
.tag("team", "engineering")
.tag("priority", "high")
.mime("text/markdown");
print("✓ Note created");
print("\nStoring note in context...");
ctx.save(note);
print("✓ Note stored");
"Note 'Sprint Planning Meeting' created and stored successfully"

View File

@@ -0,0 +1,21 @@
print("Querying context [alice, bob]...");
let ctx = get_context(["alice", "bob"]);
print("✓ Context ID: " + ctx.context_id());
print("\nListing all notes...");
let notes = ctx.list("notes");
print("✓ Found " + notes.len() + " note(s)");
print("\nRetrieving specific note...");
let note = ctx.get("notes", "sprint_planning_001");
print("✓ Retrieved note: sprint_planning_001");
print("\nQuerying context [alice, bob, charlie]...");
let ctx2 = get_context(["alice", "bob", "charlie"]);
print("✓ Context ID: " + ctx2.context_id());
print("\nListing all events...");
let events = ctx2.list("events");
print("✓ Found " + events.len() + " event(s)");
"Query complete: Found " + notes.len() + " notes and " + events.len() + " events"

190
core/src/app.rs Normal file
View File

@@ -0,0 +1,190 @@
//! # Hero Supervisor Application
//!
//! Simplified supervisor application that wraps a built Supervisor instance.
//! Use SupervisorBuilder to construct the supervisor with all configuration,
//! then pass it to SupervisorApp for runtime management.
use crate::Supervisor;
#[cfg(feature = "mycelium")]
use crate::mycelium::MyceliumIntegration;
use log::{info, error, debug};
#[cfg(feature = "mycelium")]
use std::sync::Arc;
#[cfg(feature = "mycelium")]
use tokio::sync::Mutex;
/// Main supervisor application
pub struct SupervisorApp {
pub supervisor: Supervisor,
pub mycelium_url: String,
pub topic: String,
}
impl SupervisorApp {
/// Create a new supervisor application with a built supervisor
pub fn new(supervisor: Supervisor, mycelium_url: String, topic: String) -> Self {
Self {
supervisor,
mycelium_url,
topic,
}
}
/// Start the complete supervisor application
/// This method handles the entire application lifecycle:
/// - Starts all configured runners
/// - Connects to Mycelium daemon for message transport
/// - Sets up graceful shutdown handling
/// - Keeps the application running
pub async fn start(&mut self) -> Result<(), Box<dyn std::error::Error>> {
info!("Starting Hero Supervisor Application");
// Start all configured runners
self.start_all().await?;
// Start Mycelium integration
self.start_mycelium_integration().await?;
// Set up graceful shutdown
self.setup_graceful_shutdown().await;
// Keep the application running
info!("Supervisor is running. Press Ctrl+C to shutdown.");
self.run_main_loop().await;
Ok(())
}
/// Start the Mycelium integration
async fn start_mycelium_integration(&self) -> Result<(), Box<dyn std::error::Error>> {
#[cfg(feature = "mycelium")]
{
// Skip Mycelium if URL is empty
if self.mycelium_url.is_empty() {
info!("Mycelium integration disabled (no URL provided)");
return Ok(());
}
info!("Starting Mycelium integration...");
let supervisor_for_mycelium = Arc::new(Mutex::new(self.supervisor.clone()));
let mycelium_url = self.mycelium_url.clone();
let topic = self.topic.clone();
let mycelium_integration = MyceliumIntegration::new(
supervisor_for_mycelium,
mycelium_url,
topic,
);
// Start the Mycelium integration in a background task
let integration_handle = tokio::spawn(async move {
if let Err(e) = mycelium_integration.start().await {
error!("Mycelium integration error: {}", e);
}
});
// Give the integration a moment to start
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
info!("Mycelium integration started successfully");
// Store the handle for potential cleanup
std::mem::forget(integration_handle); // For now, let it run in background
}
#[cfg(not(feature = "mycelium"))]
{
info!("Mycelium integration not enabled (compile with --features mycelium)");
}
Ok(())
}
/// Set up graceful shutdown handling
async fn setup_graceful_shutdown(&self) {
tokio::spawn(async move {
tokio::signal::ctrl_c().await.expect("Failed to listen for ctrl+c");
info!("Received shutdown signal");
std::process::exit(0);
});
}
/// Main application loop
async fn run_main_loop(&self) {
// Keep the main thread alive
loop {
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}
}
/// Start all configured runners
pub async fn start_all(&mut self) -> Result<(), Box<dyn std::error::Error>> {
info!("Starting all runners");
let results = self.supervisor.start_all().await;
let mut failed_count = 0;
for (runner_id, result) in results {
match result {
Ok(_) => info!("Runner {} started successfully", runner_id),
Err(e) => {
error!("Failed to start runner {}: {}", runner_id, e);
failed_count += 1;
}
}
}
if failed_count == 0 {
info!("All runners started successfully");
} else {
error!("Failed to start {} runners", failed_count);
}
Ok(())
}
/// Stop all configured runners
pub async fn stop_all(&mut self, force: bool) -> Result<(), Box<dyn std::error::Error>> {
info!("Stopping all runners (force: {})", force);
let results = self.supervisor.stop_all(force).await;
let mut failed_count = 0;
for (runner_id, result) in results {
match result {
Ok(_) => info!("Runner {} stopped successfully", runner_id),
Err(e) => {
error!("Failed to stop runner {}: {}", runner_id, e);
failed_count += 1;
}
}
}
if failed_count == 0 {
info!("All runners stopped successfully");
} else {
error!("Failed to stop {} runners", failed_count);
}
Ok(())
}
/// Get status of all runners
pub async fn get_status(&self) -> Result<Vec<(String, String)>, Box<dyn std::error::Error>> {
debug!("Getting status of all runners");
let statuses = self.supervisor.get_all_runner_status().await
.map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;
let status_strings: Vec<(String, String)> = statuses
.into_iter()
.map(|(runner_id, status)| {
let status_str = format!("{:?}", status);
(runner_id, status_str)
})
.collect();
Ok(status_strings)
}
}

134
core/src/auth.rs Normal file
View File

@@ -0,0 +1,134 @@
//! Authentication and API key management
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use uuid::Uuid;
/// API key scope/permission level
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ApiKeyScope {
/// Full access - can manage keys, runners, jobs
Admin,
/// Can register new runners
Registrar,
/// Can create and manage jobs
User,
}
impl ApiKeyScope {
pub fn as_str(&self) -> &'static str {
match self {
ApiKeyScope::Admin => "admin",
ApiKeyScope::Registrar => "registrar",
ApiKeyScope::User => "user",
}
}
}
/// An API key with metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ApiKey {
/// The actual key value (UUID or custom string)
pub key: String,
/// Human-readable name for the key
pub name: String,
/// Permission scope
pub scope: ApiKeyScope,
/// When the key was created
pub created_at: String,
/// Optional expiration timestamp
pub expires_at: Option<String>,
}
impl ApiKey {
/// Create a new API key with a generated UUID
pub fn new(name: String, scope: ApiKeyScope) -> Self {
Self {
key: Uuid::new_v4().to_string(),
name,
scope,
created_at: chrono::Utc::now().to_rfc3339(),
expires_at: None,
}
}
/// Create a new API key with a specific key value
pub fn with_key(key: String, name: String, scope: ApiKeyScope) -> Self {
Self {
key,
name,
scope,
created_at: chrono::Utc::now().to_rfc3339(),
expires_at: None,
}
}
}
/// API key store
#[derive(Debug, Clone, Default)]
pub struct ApiKeyStore {
/// Map of key -> ApiKey
keys: HashMap<String, ApiKey>,
}
impl ApiKeyStore {
pub fn new() -> Self {
Self {
keys: HashMap::new(),
}
}
/// Add a new API key
pub fn add_key(&mut self, key: ApiKey) {
self.keys.insert(key.key.clone(), key);
}
/// Remove an API key by its key value
pub fn remove_key(&mut self, key: &str) -> Option<ApiKey> {
self.keys.remove(key)
}
/// Get an API key by its key value
pub fn get_key(&self, key: &str) -> Option<&ApiKey> {
self.keys.get(key)
}
/// Verify a key and return its metadata if valid
pub fn verify_key(&self, key: &str) -> Option<&ApiKey> {
self.get_key(key)
}
/// List all keys with a specific scope
pub fn list_keys_by_scope(&self, scope: ApiKeyScope) -> Vec<&ApiKey> {
self.keys
.values()
.filter(|k| k.scope == scope)
.collect()
}
/// List all keys
pub fn list_all_keys(&self) -> Vec<&ApiKey> {
self.keys.values().collect()
}
/// Count keys by scope
pub fn count_by_scope(&self, scope: ApiKeyScope) -> usize {
self.keys.values().filter(|k| k.scope == scope).count()
}
/// Bootstrap with an initial admin key
pub fn bootstrap_admin_key(&mut self, name: String) -> ApiKey {
let key = ApiKey::new(name, ApiKeyScope::Admin);
self.add_key(key.clone());
key
}
}
/// Response for auth verification
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthVerifyResponse {
pub valid: bool,
pub name: String,
pub scope: String,
}

176
core/src/bin/supervisor.rs Normal file
View File

@@ -0,0 +1,176 @@
//! # Hero Supervisor Binary
//!
//! Main supervisor binary that manages multiple actors and listens to jobs over Redis.
//! The supervisor builds with actor configuration, starts actors, and dispatches jobs
//! to the appropriate runners based on the job's runner field.
use hero_supervisor::{SupervisorApp, SupervisorBuilder};
use clap::Parser;
use log::{info, error};
use std::path::PathBuf;
/// Command line arguments for the supervisor
#[derive(Parser, Debug)]
#[command(name = "supervisor")]
#[command(about = "Hero Supervisor - manages multiple actors and dispatches jobs")]
struct Args {
/// Path to the configuration TOML file
#[arg(short, long, value_name = "FILE")]
config: Option<PathBuf>,
/// Redis URL for job queue
#[arg(long, default_value = "redis://localhost:6379")]
redis_url: String,
/// Namespace for Redis keys
#[arg(long, default_value = "")]
namespace: String,
/// Admin secrets (can be specified multiple times)
#[arg(long = "admin-secret", value_name = "SECRET")]
admin_secrets: Vec<String>,
/// User secrets (can be specified multiple times)
#[arg(long = "user-secret", value_name = "SECRET")]
user_secrets: Vec<String>,
/// Register secrets (can be specified multiple times)
#[arg(long = "register-secret", value_name = "SECRET")]
register_secrets: Vec<String>,
/// Mycelium daemon URL
#[arg(long, default_value = "http://127.0.0.1:8990")]
mycelium_url: String,
/// Mycelium topic for supervisor RPC messages
#[arg(long, default_value = "supervisor.rpc")]
topic: String,
/// Port for OpenRPC HTTP server
#[arg(long, default_value = "3030")]
port: u16,
/// Bind address for OpenRPC HTTP server
#[arg(long, default_value = "127.0.0.1")]
bind_address: String,
/// Bootstrap an initial admin API key with the given name
#[arg(long = "bootstrap-admin-key", value_name = "NAME")]
bootstrap_admin_key: Option<String>,
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize logging
env_logger::init();
info!("Starting Hero Supervisor");
// Parse command line arguments
let args = Args::parse();
// Create and initialize supervisor using builder pattern
let mut builder = SupervisorBuilder::new()
.redis_url(&args.redis_url)
.namespace(&args.namespace);
// Add secrets from CLI arguments
if !args.admin_secrets.is_empty() {
info!("Adding {} admin secret(s)", args.admin_secrets.len());
builder = builder.admin_secrets(args.admin_secrets);
}
if !args.user_secrets.is_empty() {
info!("Adding {} user secret(s)", args.user_secrets.len());
builder = builder.user_secrets(args.user_secrets);
}
if !args.register_secrets.is_empty() {
info!("Adding {} register secret(s)", args.register_secrets.len());
builder = builder.register_secrets(args.register_secrets);
}
let supervisor = match args.config {
Some(_config_path) => {
info!("Loading configuration from config file not yet implemented");
// For now, use CLI configuration
builder.build().await?
}
None => {
info!("Using CLI configuration");
builder.build().await?
}
};
// Bootstrap admin key if requested
if let Some(admin_key_name) = args.bootstrap_admin_key {
info!("Bootstrapping admin API key: {}", admin_key_name);
let admin_key = supervisor.bootstrap_admin_key(admin_key_name).await;
println!("\n╔════════════════════════════════════════════════════════════╗");
println!("║ 🔑 Admin API Key Created ║");
println!("╚════════════════════════════════════════════════════════════╝");
println!(" Name: {}", admin_key.name);
println!(" Key: {}", admin_key.key);
println!(" Scope: {}", admin_key.scope.as_str());
println!(" ⚠️ SAVE THIS KEY - IT WILL NOT BE SHOWN AGAIN!");
println!("╚════════════════════════════════════════════════════════════╝\n");
}
// Print startup information
let server_url = format!("http://{}:{}", args.bind_address, args.port);
println!("\n╔════════════════════════════════════════════════════════════╗");
println!("║ Hero Supervisor Started ║");
println!("╚════════════════════════════════════════════════════════════╝");
println!(" 📡 OpenRPC Server: {}", server_url);
println!(" 🔗 Redis: {}", args.redis_url);
#[cfg(feature = "mycelium")]
if !args.mycelium_url.is_empty() {
println!(" 🌐 Mycelium: {}", args.mycelium_url);
} else {
println!(" 🌐 Mycelium: Disabled");
}
#[cfg(not(feature = "mycelium"))]
println!(" 🌐 Mycelium: Not compiled (use --features mycelium)");
println!("╚════════════════════════════════════════════════════════════╝\n");
// Start OpenRPC server in background
use std::sync::Arc;
use tokio::sync::Mutex;
use hero_supervisor::openrpc::start_http_openrpc_server;
let supervisor_arc = Arc::new(Mutex::new(supervisor.clone()));
let bind_addr = args.bind_address.clone();
let port = args.port;
tokio::spawn(async move {
info!("Starting OpenRPC server on {}:{}", bind_addr, port);
match start_http_openrpc_server(supervisor_arc, &bind_addr, port).await {
Ok(handle) => {
info!("OpenRPC server started successfully");
// Keep the server running by holding the handle
handle.stopped().await;
error!("OpenRPC server stopped unexpectedly");
}
Err(e) => {
error!("OpenRPC server error: {}", e);
}
}
});
// Give the server a moment to start
tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
let mut app = SupervisorApp::new(supervisor, args.mycelium_url, args.topic);
// Start the complete supervisor application
app.start().await?;
Ok(())
}

3
core/src/job.rs Normal file
View File

@@ -0,0 +1,3 @@
// Re-export job types from the hero-job crate
pub use hero_job::{Job, JobBuilder, JobStatus, JobError};
use hero_job_client::{Client, ClientBuilder};

25
core/src/lib.rs Normal file
View File

@@ -0,0 +1,25 @@
//! Hero Supervisor - Actor management for the Hero ecosystem.
//!
//! See README.md for detailed documentation and usage examples.
pub mod runner;
pub mod job;
pub mod supervisor;
pub mod app;
pub mod openrpc;
pub mod auth;
pub mod services;
#[cfg(feature = "mycelium")]
pub mod mycelium;
// Re-export main types for convenience
pub use runner::{Runner, RunnerConfig, RunnerResult, RunnerStatus};
// pub use sal_service_manager::{ProcessManager, SimpleProcessManager, TmuxProcessManager};
pub use supervisor::{Supervisor, SupervisorBuilder, ProcessManagerType};
pub use hero_job::{Job, JobBuilder, JobStatus, JobError};
use hero_job_client::{Client, ClientBuilder};
pub use app::SupervisorApp;
#[cfg(feature = "mycelium")]
pub use mycelium::{MyceliumIntegration, MyceliumServer};

519
core/src/mycelium.rs Normal file
View File

@@ -0,0 +1,519 @@
//! # Mycelium Integration for Hero Supervisor
//!
//! This module integrates the supervisor with Mycelium's message transport system.
//! Instead of running its own server, it connects to an existing Mycelium daemon
//! and listens for incoming supervisor RPC messages via HTTP REST API.
use std::sync::Arc;
use tokio::sync::Mutex;
use serde_json::{Value, json};
use log::{info, error, debug, trace};
use base64::Engine;
use reqwest::Client as HttpClient;
use crate::Supervisor;
use tokio::time::{sleep, Duration};
/// Mycelium integration that connects to a Mycelium daemon and handles supervisor RPC messages
pub struct MyceliumIntegration {
supervisor: Arc<Mutex<Supervisor>>,
mycelium_url: String,
http_client: HttpClient,
topic: String,
running: Arc<Mutex<bool>>,
}
impl MyceliumIntegration {
pub fn new(supervisor: Arc<Mutex<Supervisor>>, mycelium_url: String, topic: String) -> Self {
Self {
supervisor,
mycelium_url,
http_client: HttpClient::new(),
topic,
running: Arc::new(Mutex::new(false)),
}
}
/// Start listening for messages on the Mycelium network
pub async fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
info!("Starting Mycelium integration with daemon at {}", self.mycelium_url);
// Skip connection test for now due to API compatibility issues
// TODO: Fix Mycelium API compatibility
info!("Skipping connection test - assuming Mycelium daemon is running");
// Set running flag
{
let mut running = self.running.lock().await;
*running = true;
}
info!("Mycelium integration started successfully, listening on topic: {}", self.topic);
// Start message polling loop
let supervisor = Arc::clone(&self.supervisor);
let http_client = self.http_client.clone();
let mycelium_url = self.mycelium_url.clone();
let topic = self.topic.clone();
let running = Arc::clone(&self.running);
tokio::spawn(async move {
Self::message_loop(supervisor, http_client, mycelium_url, topic, running).await;
});
Ok(())
}
/// Test connection to Mycelium daemon using JSON-RPC
async fn test_connection(&self) -> Result<(), Box<dyn std::error::Error>> {
let test_request = json!({
"jsonrpc": "2.0",
"method": "getInfo",
"params": [],
"id": 1
});
let response = self.http_client
.post(&self.mycelium_url)
.json(&test_request)
.send()
.await?;
if response.status().is_success() {
let result: Value = response.json().await?;
if result.get("result").is_some() {
info!("Successfully connected to Mycelium daemon at {}", self.mycelium_url);
Ok(())
} else {
error!("Mycelium daemon returned error: {}", result);
Err("Mycelium daemon returned error".into())
}
} else {
let status = response.status();
let text = response.text().await.unwrap_or_default();
error!("Failed to connect to Mycelium daemon: {} - {}", status, text);
Err(format!("Mycelium connection failed: {}", status).into())
}
}
/// Handle incoming supervisor RPC message (called by Mycelium daemon via pushMessage)
pub async fn handle_supervisor_message(
&self,
payload_b64: &str,
reply_info: Option<(String, String)>,
) -> Result<Option<String>, Box<dyn std::error::Error + Send + Sync>> {
// Decode the base64 payload
let payload_bytes = base64::engine::general_purpose::STANDARD
.decode(payload_b64.as_bytes())?;
let payload_str = String::from_utf8(payload_bytes)?;
info!("Received supervisor message: {}", payload_str);
// Parse the JSON-RPC request
let request: Value = serde_json::from_str(&payload_str)?;
debug!("Decoded supervisor RPC: {}", request);
// Extract method and params from supervisor JSON-RPC
let method = request.get("method")
.and_then(|v| v.as_str())
.ok_or("missing method")?;
let rpc_params = request.get("params")
.cloned()
.unwrap_or(json!([]));
let rpc_id = request.get("id").cloned();
// Route to appropriate supervisor method
let result = self.route_supervisor_call(method, rpc_params).await?;
// If we have reply info, send the response back via Mycelium
if let Some((src_ip, _msg_id)) = reply_info {
let supervisor_response = json!({
"jsonrpc": "2.0",
"id": rpc_id,
"result": result
});
let response_b64 = base64::engine::general_purpose::STANDARD
.encode(serde_json::to_string(&supervisor_response)?.as_bytes());
info!("Sending response back to client at {}: {}", src_ip, supervisor_response);
// Send reply back to the client
match self.send_reply(&src_ip, &response_b64).await {
Ok(()) => info!("✅ Response sent successfully to {}", src_ip),
Err(e) => error!("❌ Failed to send response to {}: {}", src_ip, e),
}
}
Ok(Some("handled".to_string()))
}
/// Send a reply message back to a client using Mycelium JSON-RPC
async fn send_reply(
&self,
dst_ip: &str,
payload_b64: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// Send response to a dedicated response topic
let response_topic = "supervisor.response";
let topic_b64 = base64::engine::general_purpose::STANDARD.encode(response_topic.as_bytes());
let message_info = json!({
"dst": { "ip": dst_ip },
"topic": topic_b64,
"payload": payload_b64 // payload_b64 is already base64 encoded
});
let push_request = json!({
"jsonrpc": "2.0",
"method": "pushMessage",
"params": [message_info, null],
"id": 1
});
let response = self.http_client
.post(&self.mycelium_url)
.json(&push_request)
.send()
.await?;
if response.status().is_success() {
let result: Value = response.json().await?;
if result.get("result").is_some() {
debug!("Sent reply to {}", dst_ip);
Ok(())
} else {
error!("Failed to send reply, Mycelium error: {}", result);
Err("Mycelium pushMessage failed".into())
}
} else {
let status = response.status();
let text = response.text().await.unwrap_or_default();
error!("Failed to send reply: {} - {}", status, text);
Err(format!("Failed to send reply: {}", status).into())
}
}
/// Route supervisor method calls to the appropriate supervisor functions
async fn route_supervisor_call(
&self,
method: &str,
params: Value,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
let mut supervisor_guard = self.supervisor.lock().await;
match method {
"list_runners" => {
// list_runners doesn't require parameters
let runners = supervisor_guard.list_runners();
Ok(json!(runners))
}
"register_runner" => {
if let Some(param_obj) = params.as_array().and_then(|arr| arr.get(0)) {
let secret = param_obj.get("secret")
.and_then(|v| v.as_str())
.ok_or("missing secret")?;
let name = param_obj.get("name")
.and_then(|v| v.as_str())
.ok_or("missing name")?;
let queue = param_obj.get("queue")
.and_then(|v| v.as_str())
.ok_or("missing queue")?;
supervisor_guard.register_runner(secret, name, queue).await?;
Ok(json!("success"))
} else {
Err("invalid register_runner params".into())
}
}
"start_runner" => {
if let Some(actor_id) = params.as_array().and_then(|arr| arr.get(0)).and_then(|v| v.as_str()) {
supervisor_guard.start_runner(actor_id).await?;
Ok(json!("success"))
} else {
Err("invalid start_runner params".into())
}
}
"stop_runner" => {
if let Some(arr) = params.as_array() {
let actor_id = arr.get(0).and_then(|v| v.as_str()).ok_or("missing actor_id")?;
let force = arr.get(1).and_then(|v| v.as_bool()).unwrap_or(false);
supervisor_guard.stop_runner(actor_id, force).await?;
Ok(json!("success"))
} else {
Err("invalid stop_runner params".into())
}
}
"get_runner_status" => {
if let Some(actor_id) = params.as_array().and_then(|arr| arr.get(0)).and_then(|v| v.as_str()) {
let status = supervisor_guard.get_runner_status(actor_id).await?;
Ok(json!(format!("{:?}", status)))
} else {
Err("invalid get_runner_status params".into())
}
}
"get_all_runner_status" => {
let statuses = supervisor_guard.get_all_runner_status().await?;
let status_map: std::collections::HashMap<String, String> = statuses
.into_iter()
.map(|(id, status)| (id, format!("{:?}", status)))
.collect();
Ok(json!(status_map))
}
"start_all" => {
let results = supervisor_guard.start_all().await;
let status_results: Vec<(String, String)> = results
.into_iter()
.map(|(id, result)| {
let status = match result {
Ok(_) => "started".to_string(),
Err(e) => format!("error: {}", e),
};
(id, status)
})
.collect();
Ok(json!(status_results))
}
"stop_all" => {
let force = params.as_array()
.and_then(|arr| arr.get(0))
.and_then(|v| v.as_bool())
.unwrap_or(false);
let results = supervisor_guard.stop_all(force).await;
let status_results: Vec<(String, String)> = results
.into_iter()
.map(|(id, result)| {
let status = match result {
Ok(_) => "stopped".to_string(),
Err(e) => format!("error: {}", e),
};
(id, status)
})
.collect();
Ok(json!(status_results))
}
"job.run" => {
// Run job and wait for result (blocking)
if let Some(param_obj) = params.as_array().and_then(|arr| arr.get(0)) {
let _secret = param_obj.get("secret")
.and_then(|v| v.as_str())
.ok_or("missing secret")?;
let job_value = param_obj.get("job")
.ok_or("missing job")?;
let timeout = param_obj.get("timeout")
.and_then(|v| v.as_u64())
.unwrap_or(60);
// Deserialize the job
let job: hero_job::Job = serde_json::from_value(job_value.clone())
.map_err(|e| format!("invalid job format: {}", e))?;
let job_id = job.id.clone();
let runner_name = job.runner.clone();
// Verify signatures
job.verify_signatures()
.map_err(|e| format!("signature verification failed: {}", e))?;
info!("Job {} signature verification passed for signatories: {:?}",
job_id, job.signatories());
// Queue and wait for result
let mut supervisor_guard = self.supervisor.lock().await;
let result = supervisor_guard.queue_and_wait(&runner_name, job, timeout)
.await
.map_err(|e| format!("job execution failed: {}", e))?;
Ok(json!({
"job_id": job_id,
"status": "completed",
"result": result
}))
} else {
Err("invalid job.run params".into())
}
}
"job.start" => {
// Start job without waiting (non-blocking)
if let Some(param_obj) = params.as_array().and_then(|arr| arr.get(0)) {
let _secret = param_obj.get("secret")
.and_then(|v| v.as_str())
.ok_or("missing secret")?;
let job_value = param_obj.get("job")
.ok_or("missing job")?;
// Deserialize the job
let job: hero_job::Job = serde_json::from_value(job_value.clone())
.map_err(|e| format!("invalid job format: {}", e))?;
let job_id = job.id.clone();
let runner_name = job.runner.clone();
// Verify signatures
job.verify_signatures()
.map_err(|e| format!("signature verification failed: {}", e))?;
info!("Job {} signature verification passed for signatories: {:?}",
job_id, job.signatories());
// Queue the job without waiting
let mut supervisor_guard = self.supervisor.lock().await;
supervisor_guard.queue_job_to_runner(&runner_name, job)
.await
.map_err(|e| format!("failed to queue job: {}", e))?;
Ok(json!({
"job_id": job_id,
"status": "queued"
}))
} else {
Err("invalid job.start params".into())
}
}
"job.status" => {
if let Some(_job_id) = params.as_array().and_then(|arr| arr.get(0)).and_then(|v| v.as_str()) {
// TODO: Implement actual job status lookup
Ok(json!({"status": "completed"}))
} else {
Err("invalid job.status params".into())
}
}
"job.result" => {
if let Some(_job_id) = params.as_array().and_then(|arr| arr.get(0)).and_then(|v| v.as_str()) {
// TODO: Implement actual job result lookup
Ok(json!({"success": "job completed successfully"}))
} else {
Err("invalid job.result params".into())
}
}
"rpc.discover" => {
let methods = vec![
"list_runners", "register_runner", "start_runner", "stop_runner",
"get_runner_status", "get_all_runner_status", "start_all", "stop_all",
"job.run", "job.start", "job.status", "job.result", "rpc.discover"
];
Ok(json!(methods))
}
_ => {
error!("Unknown method: {}", method);
Err(format!("unknown method: {}", method).into())
}
}
}
/// Message polling loop that listens for incoming messages
async fn message_loop(
supervisor: Arc<Mutex<Supervisor>>,
http_client: HttpClient,
mycelium_url: String,
topic: String,
running: Arc<Mutex<bool>>,
) {
info!("Starting message polling loop for topic: {} (base64: {})", topic, base64::engine::general_purpose::STANDARD.encode(topic.as_bytes()));
while {
let running_guard = running.lock().await;
*running_guard
} {
// Poll for messages using Mycelium JSON-RPC API
// Topic needs to be base64 encoded for the RPC API
let topic_b64 = base64::engine::general_purpose::STANDARD.encode(topic.as_bytes());
let poll_request = json!({
"jsonrpc": "2.0",
"method": "popMessage",
"params": [null, 1, topic_b64], // Reduced timeout to 1 second
"id": 1
});
debug!("Polling for messages with request: {}", poll_request);
match tokio::time::timeout(
Duration::from_secs(10),
http_client.post(&mycelium_url).json(&poll_request).send()
).await {
Ok(Ok(response)) => {
if response.status().is_success() {
match response.json::<Value>().await {
Ok(rpc_response) => {
if let Some(message) = rpc_response.get("result") {
debug!("Received message: {}", message);
// Extract message details
if let (Some(payload), Some(src_ip), Some(msg_id)) = (
message.get("payload").and_then(|v| v.as_str()),
message.get("srcIp").and_then(|v| v.as_str()),
message.get("id").and_then(|v| v.as_str()),
) {
// Create a temporary integration instance to handle the message
let integration = MyceliumIntegration {
supervisor: Arc::clone(&supervisor),
mycelium_url: mycelium_url.clone(),
http_client: http_client.clone(),
topic: topic.clone(),
running: Arc::clone(&running),
};
let reply_info = Some((src_ip.to_string(), msg_id.to_string()));
if let Err(e) = integration.handle_supervisor_message(payload, reply_info).await {
error!("Error handling supervisor message: {}", e);
}
}
} else if let Some(error) = rpc_response.get("error") {
let error_code = error.get("code").and_then(|c| c.as_i64()).unwrap_or(0);
if error_code == -32014 {
// Timeout - no message available, continue polling
trace!("No messages available (timeout)");
} else {
error!("Mycelium RPC error: {}", error);
sleep(Duration::from_secs(1)).await;
}
} else {
trace!("No messages available");
}
}
Err(e) => {
error!("Failed to parse RPC response JSON: {}", e);
}
}
} else {
let status = response.status();
let text = response.text().await.unwrap_or_default();
error!("Message polling error: {} - {}", status, text);
sleep(Duration::from_secs(1)).await;
}
}
Ok(Err(e)) => {
error!("HTTP request failed: {}", e);
sleep(Duration::from_secs(1)).await;
}
Err(_) => {
error!("Polling request timed out after 10 seconds");
sleep(Duration::from_secs(1)).await;
}
}
}
info!("Message polling loop stopped");
}
}
// Legacy type alias for backward compatibility
pub type MyceliumServer = MyceliumIntegration;

1299
core/src/openrpc.rs Normal file

File diff suppressed because it is too large Load Diff

230
core/src/openrpc/tests.rs Normal file
View File

@@ -0,0 +1,230 @@
//! Tests for the new job API methods
#[cfg(test)]
mod job_api_tests {
use super::super::*;
use crate::supervisor::{Supervisor, SupervisorBuilder};
use crate::job::{Job, JobBuilder};
use std::sync::Arc;
use tokio::sync::Mutex;
use serde_json::json;
async fn create_test_supervisor() -> Arc<Mutex<Supervisor>> {
let supervisor = SupervisorBuilder::new()
.redis_url("redis://localhost:6379")
.namespace("test_job_api")
.build()
.await
.unwrap_or_else(|_| Supervisor::default());
let mut supervisor = supervisor;
supervisor.add_admin_secret("test-admin-secret".to_string());
supervisor.add_user_secret("test-user-secret".to_string());
Arc::new(Mutex::new(supervisor))
}
fn create_test_job() -> Job {
JobBuilder::new()
.id("test-job-123".to_string())
.caller_id("test-client".to_string())
.context_id("test-context".to_string())
.script("print('Hello World')".to_string())
.script_type(crate::job::ScriptType::Osis)
.timeout(30)
.build()
.unwrap()
}
#[tokio::test]
async fn test_jobs_create() {
let supervisor = create_test_supervisor().await;
let job = create_test_job();
let params = RunJobParams {
secret: "test-user-secret".to_string(),
job: job.clone(),
};
let result = supervisor.jobs_create(params).await;
assert!(result.is_ok());
let job_id = result.unwrap();
assert_eq!(job_id, job.id);
}
#[tokio::test]
async fn test_jobs_create_invalid_secret() {
let supervisor = create_test_supervisor().await;
let job = create_test_job();
let params = RunJobParams {
secret: "invalid-secret".to_string(),
job,
};
let result = supervisor.jobs_create(params).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_jobs_list() {
let supervisor = create_test_supervisor().await;
let result = supervisor.jobs_list().await;
// Should not error even if Redis is not available (will return empty list or error)
// The important thing is that the method signature works
assert!(result.is_ok() || result.is_err());
}
#[tokio::test]
async fn test_job_run_success_format() {
let supervisor = create_test_supervisor().await;
let job = create_test_job();
let params = RunJobParams {
secret: "test-user-secret".to_string(),
job,
};
let result = supervisor.job_run(params).await;
// The result should be a JobResult enum
match result {
Ok(JobResult::Success { success: _ }) => {
// Success case - job executed and returned output
},
Ok(JobResult::Error { error: _ }) => {
// Error case - job failed but method worked
},
Err(_) => {
// Method error (authentication, etc.)
// This is acceptable for testing without actual runners
}
}
}
#[tokio::test]
async fn test_job_start() {
let supervisor = create_test_supervisor().await;
let params = StartJobParams {
secret: "test-user-secret".to_string(),
job_id: "test-job-123".to_string(),
};
let result = supervisor.job_start(params).await;
// Should fail gracefully if job doesn't exist
assert!(result.is_err() || result.is_ok());
}
#[tokio::test]
async fn test_job_start_invalid_secret() {
let supervisor = create_test_supervisor().await;
let params = StartJobParams {
secret: "invalid-secret".to_string(),
job_id: "test-job-123".to_string(),
};
let result = supervisor.job_start(params).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_job_status() {
let supervisor = create_test_supervisor().await;
let result = supervisor.job_status("test-job-123".to_string()).await;
// Should return error for non-existent job
assert!(result.is_err());
}
#[tokio::test]
async fn test_job_result() {
let supervisor = create_test_supervisor().await;
let result = supervisor.job_result("test-job-123".to_string()).await;
// Should return error for non-existent job
assert!(result.is_err());
}
#[test]
fn test_job_result_enum_serialization() {
let success_result = JobResult::Success {
success: "Job completed successfully".to_string(),
};
let serialized = serde_json::to_string(&success_result).unwrap();
assert!(serialized.contains("success"));
assert!(serialized.contains("Job completed successfully"));
let error_result = JobResult::Error {
error: "Job failed with error".to_string(),
};
let serialized = serde_json::to_string(&error_result).unwrap();
assert!(serialized.contains("error"));
assert!(serialized.contains("Job failed with error"));
}
#[test]
fn test_job_status_response_serialization() {
let status_response = JobStatusResponse {
job_id: "test-job-123".to_string(),
status: "running".to_string(),
created_at: "2023-01-01T00:00:00Z".to_string(),
started_at: Some("2023-01-01T00:00:05Z".to_string()),
completed_at: None,
};
let serialized = serde_json::to_string(&status_response).unwrap();
assert!(serialized.contains("test-job-123"));
assert!(serialized.contains("running"));
assert!(serialized.contains("2023-01-01T00:00:00Z"));
assert!(serialized.contains("2023-01-01T00:00:05Z"));
let deserialized: JobStatusResponse = serde_json::from_str(&serialized).unwrap();
assert_eq!(deserialized.job_id, "test-job-123");
assert_eq!(deserialized.status, "running");
assert_eq!(deserialized.started_at, Some("2023-01-01T00:00:05Z".to_string()));
assert_eq!(deserialized.completed_at, None);
}
#[test]
fn test_start_job_params_serialization() {
let params = StartJobParams {
secret: "test-secret".to_string(),
job_id: "job-123".to_string(),
};
let serialized = serde_json::to_string(&params).unwrap();
assert!(serialized.contains("test-secret"));
assert!(serialized.contains("job-123"));
let deserialized: StartJobParams = serde_json::from_str(&serialized).unwrap();
assert_eq!(deserialized.secret, "test-secret");
assert_eq!(deserialized.job_id, "job-123");
}
#[test]
fn test_method_naming_convention() {
// Test that method names follow the jobs./job. convention
// These should be the actual method names in the trait
let jobs_methods = vec!["jobs.create", "jobs.list"];
let job_methods = vec!["job.run", "job.start", "job.status", "job.result"];
// Verify naming convention
for method in jobs_methods {
assert!(method.starts_with("jobs."));
}
for method in job_methods {
assert!(method.starts_with("job."));
}
}
}

207
core/src/runner.rs Normal file
View File

@@ -0,0 +1,207 @@
//! Runner implementation for actor process management.
// use sal_service_manager::{ProcessManagerError as ServiceProcessManagerError, ProcessStatus, ProcessConfig};
/// Simple process status enum to replace sal_service_manager dependency
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum ProcessStatus {
NotStarted,
Starting,
Running,
Stopping,
Stopped,
Failed,
Error(String),
}
/// Simple process config to replace sal_service_manager dependency
#[derive(Debug, Clone)]
pub struct ProcessConfig {
pub command: String,
pub args: Vec<String>,
pub working_dir: Option<String>,
pub env_vars: Vec<(String, String)>,
}
impl ProcessConfig {
pub fn new(command: String, args: Vec<String>, working_dir: Option<String>, env_vars: Vec<(String, String)>) -> Self {
Self {
command,
args,
working_dir,
env_vars,
}
}
}
/// Simple process manager error to replace sal_service_manager dependency
#[derive(Debug, thiserror::Error)]
pub enum ProcessManagerError {
#[error("Process execution failed: {0}")]
ExecutionFailed(String),
#[error("Process not found: {0}")]
ProcessNotFound(String),
#[error("IO error: {0}")]
IoError(String),
}
use std::path::PathBuf;
/// Represents the current status of an actor/runner (alias for ProcessStatus)
pub type RunnerStatus = ProcessStatus;
/// Log information structure with serialization support
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct LogInfo {
pub timestamp: String,
pub level: String,
pub message: String,
}
/// Runner configuration and state (merged from RunnerConfig)
#[derive(Debug, Clone)]
pub struct Runner {
/// Unique identifier for the runner
pub id: String,
pub name: String,
pub namespace: String,
/// Path to the actor binary
pub command: PathBuf, // Command to run runner by, used only if supervisor is used to run runners
/// Redis URL for job queue
pub redis_url: String,
/// Additional command-line arguments
pub extra_args: Vec<String>,
}
impl Runner {
/// Create a new runner from configuration
pub fn from_config(config: RunnerConfig) -> Self {
Self {
id: config.id,
name: config.name,
namespace: config.namespace,
command: config.command,
redis_url: config.redis_url,
extra_args: config.extra_args,
}
}
/// Create a new runner with extra arguments
pub fn with_args(
id: String,
name: String,
namespace: String,
command: PathBuf,
redis_url: String,
extra_args: Vec<String>,
) -> Self {
Self {
id,
name,
namespace,
command,
redis_url,
extra_args,
}
}
/// Get the queue key for this runner with the given namespace
pub fn get_queue(&self) -> String {
if self.namespace == "" {
format!("runner:{}", self.name)
} else {
format!("{}:runner:{}", self.namespace, self.name)
}
}
}
/// Result type for runner operations
pub type RunnerResult<T> = Result<T, RunnerError>;
/// Errors that can occur during runner operations
#[derive(Debug, thiserror::Error)]
pub enum RunnerError {
#[error("Actor '{actor_id}' not found")]
ActorNotFound { actor_id: String },
#[error("Actor '{actor_id}' is already running")]
ActorAlreadyRunning { actor_id: String },
#[error("Actor '{actor_id}' is not running")]
ActorNotRunning { actor_id: String },
#[error("Failed to start actor '{actor_id}': {reason}")]
StartupFailed { actor_id: String, reason: String },
#[error("Failed to stop actor '{actor_id}': {reason}")]
StopFailed { actor_id: String, reason: String },
#[error("Timeout waiting for actor '{actor_id}' to start")]
StartupTimeout { actor_id: String },
#[error("Job queue error for actor '{actor_id}': {reason}")]
QueueError { actor_id: String, reason: String },
#[error("Process manager error: {source}")]
ProcessManagerError {
#[from]
source: ProcessManagerError,
},
#[error("Configuration error: {reason}")]
ConfigError { reason: String },
#[error("Invalid secret: {0}")]
InvalidSecret(String),
#[error("IO error: {source}")]
IoError {
#[from]
source: std::io::Error,
},
#[error("Redis error: {source}")]
RedisError {
#[from]
source: redis::RedisError,
},
#[error("Job error: {source}")]
JobError {
#[from]
source: hero_job::JobError,
},
#[error("Job client error: {source}")]
JobClientError {
#[from]
source: hero_job_client::ClientError,
},
#[error("Job '{job_id}' not found")]
JobNotFound { job_id: String },
#[error("Authentication error: {message}")]
AuthenticationError { message: String },
}
// Type alias for backward compatibility
pub type RunnerConfig = Runner;
/// Convert Runner to ProcessConfig
pub fn runner_to_process_config(config: &Runner) -> ProcessConfig {
let mut args = vec![
config.id.clone(), // First positional argument is the runner ID
"--redis-url".to_string(),
config.redis_url.clone(),
];
// Add extra arguments (e.g., context configurations)
args.extend(config.extra_args.clone());
ProcessConfig::new(
config.command.to_string_lossy().to_string(),
args,
Some("/tmp".to_string()), // Default working directory since Runner doesn't have working_dir field
vec![]
)
}

312
core/src/services.rs Normal file
View File

@@ -0,0 +1,312 @@
//! Service layer for persistent storage of keys, runners, and jobs
//!
//! This module provides database/storage services for the supervisor.
//! Currently uses in-memory storage, but designed to be easily extended
//! to use Redis, PostgreSQL, or other persistent storage backends.
use crate::auth::{ApiKey, ApiKeyScope};
use crate::job::Job;
use crate::runner::Runner;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;
use serde::{Deserialize, Serialize};
/// Service for managing API keys
#[derive(Debug, Clone)]
pub struct ApiKeyService {
store: Arc<Mutex<HashMap<String, ApiKey>>>,
}
impl ApiKeyService {
/// Create a new API key service
pub fn new() -> Self {
Self {
store: Arc::new(Mutex::new(HashMap::new())),
}
}
/// Store an API key
pub async fn store(&self, key: ApiKey) -> Result<(), String> {
let mut store = self.store.lock().await;
store.insert(key.key.clone(), key);
Ok(())
}
/// Get an API key by its key string
pub async fn get(&self, key: &str) -> Option<ApiKey> {
let store = self.store.lock().await;
store.get(key).cloned()
}
/// List all API keys
pub async fn list(&self) -> Vec<ApiKey> {
let store = self.store.lock().await;
store.values().cloned().collect()
}
/// Remove an API key
pub async fn remove(&self, key: &str) -> Option<ApiKey> {
let mut store = self.store.lock().await;
store.remove(key)
}
/// Count API keys by scope
pub async fn count_by_scope(&self, scope: ApiKeyScope) -> usize {
let store = self.store.lock().await;
store.values().filter(|k| k.scope == scope).count()
}
/// Clear all API keys (for testing)
pub async fn clear(&self) {
let mut store = self.store.lock().await;
store.clear();
}
}
impl Default for ApiKeyService {
fn default() -> Self {
Self::new()
}
}
/// Service for managing runners
#[derive(Debug, Clone)]
pub struct RunnerService {
store: Arc<Mutex<HashMap<String, RunnerMetadata>>>,
}
/// Metadata about a runner for storage
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunnerMetadata {
pub id: String,
pub name: String,
pub queue: String,
pub registered_at: String,
pub registered_by: String, // API key name that registered this runner
}
impl RunnerService {
/// Create a new runner service
pub fn new() -> Self {
Self {
store: Arc::new(Mutex::new(HashMap::new())),
}
}
/// Store runner metadata
pub async fn store(&self, metadata: RunnerMetadata) -> Result<(), String> {
let mut store = self.store.lock().await;
store.insert(metadata.id.clone(), metadata);
Ok(())
}
/// Get runner metadata by ID
pub async fn get(&self, id: &str) -> Option<RunnerMetadata> {
let store = self.store.lock().await;
store.get(id).cloned()
}
/// List all runners
pub async fn list(&self) -> Vec<RunnerMetadata> {
let store = self.store.lock().await;
store.values().cloned().collect()
}
/// Remove a runner
pub async fn remove(&self, id: &str) -> Option<RunnerMetadata> {
let mut store = self.store.lock().await;
store.remove(id)
}
/// Count total runners
pub async fn count(&self) -> usize {
let store = self.store.lock().await;
store.len()
}
/// Clear all runners (for testing)
pub async fn clear(&self) {
let mut store = self.store.lock().await;
store.clear();
}
}
impl Default for RunnerService {
fn default() -> Self {
Self::new()
}
}
/// Service for managing jobs
#[derive(Debug, Clone)]
pub struct JobService {
store: Arc<Mutex<HashMap<String, JobMetadata>>>,
}
/// Metadata about a job for storage
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobMetadata {
pub job_id: String,
pub runner: String,
pub created_at: String,
pub created_by: String, // API key name that created this job
pub status: String,
pub job: Job,
}
impl JobService {
/// Create a new job service
pub fn new() -> Self {
Self {
store: Arc::new(Mutex::new(HashMap::new())),
}
}
/// Store job metadata
pub async fn store(&self, metadata: JobMetadata) -> Result<(), String> {
let mut store = self.store.lock().await;
store.insert(metadata.job_id.clone(), metadata);
Ok(())
}
/// Get job metadata by ID
pub async fn get(&self, job_id: &str) -> Option<JobMetadata> {
let store = self.store.lock().await;
store.get(job_id).cloned()
}
/// List all jobs
pub async fn list(&self) -> Vec<JobMetadata> {
let store = self.store.lock().await;
store.values().cloned().collect()
}
/// List jobs by runner
pub async fn list_by_runner(&self, runner: &str) -> Vec<JobMetadata> {
let store = self.store.lock().await;
store.values()
.filter(|j| j.runner == runner)
.cloned()
.collect()
}
/// List jobs by creator (API key name)
pub async fn list_by_creator(&self, creator: &str) -> Vec<JobMetadata> {
let store = self.store.lock().await;
store.values()
.filter(|j| j.created_by == creator)
.cloned()
.collect()
}
/// Update job status
pub async fn update_status(&self, job_id: &str, status: String) -> Result<(), String> {
let mut store = self.store.lock().await;
if let Some(metadata) = store.get_mut(job_id) {
metadata.status = status;
Ok(())
} else {
Err(format!("Job not found: {}", job_id))
}
}
/// Remove a job
pub async fn remove(&self, job_id: &str) -> Option<JobMetadata> {
let mut store = self.store.lock().await;
store.remove(job_id)
}
/// Count total jobs
pub async fn count(&self) -> usize {
let store = self.store.lock().await;
store.len()
}
/// Clear all jobs (for testing)
pub async fn clear(&self) {
let mut store = self.store.lock().await;
store.clear();
}
}
impl Default for JobService {
fn default() -> Self {
Self::new()
}
}
/// Combined service container for all storage services
#[derive(Debug, Clone)]
pub struct Services {
pub api_keys: ApiKeyService,
pub runners: RunnerService,
pub jobs: JobService,
}
impl Services {
/// Create a new services container
pub fn new() -> Self {
Self {
api_keys: ApiKeyService::new(),
runners: RunnerService::new(),
jobs: JobService::new(),
}
}
/// Clear all data (for testing)
pub async fn clear_all(&self) {
self.api_keys.clear().await;
self.runners.clear().await;
self.jobs.clear().await;
}
}
impl Default for Services {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_api_key_service() {
let service = ApiKeyService::new();
let key = ApiKey {
key: "test-key".to_string(),
name: "test".to_string(),
scope: ApiKeyScope::User,
};
service.store(key.clone()).await.unwrap();
assert_eq!(service.get("test-key").await.unwrap().name, "test");
assert_eq!(service.list().await.len(), 1);
service.remove("test-key").await;
assert!(service.get("test-key").await.is_none());
}
#[tokio::test]
async fn test_runner_service() {
let service = RunnerService::new();
let metadata = RunnerMetadata {
id: "runner1".to_string(),
name: "runner1".to_string(),
queue: "queue1".to_string(),
registered_at: "2024-01-01".to_string(),
registered_by: "admin".to_string(),
};
service.store(metadata.clone()).await.unwrap();
assert_eq!(service.get("runner1").await.unwrap().name, "runner1");
assert_eq!(service.count().await, 1);
service.remove("runner1").await;
assert!(service.get("runner1").await.is_none());
}
}

1107
core/src/supervisor.rs Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,279 @@
//! Integration tests for the job API
//!
//! These tests validate the complete job lifecycle using a real supervisor instance.
//! They require Redis and a running supervisor to execute properly.
use hero_supervisor_openrpc_client::{SupervisorClient, JobBuilder, JobResult};
use std::time::Duration;
use tokio::time::sleep;
use uuid::Uuid;
/// Test helper to create a unique job for testing
fn create_test_job(context: &str) -> Result<hero_supervisor_openrpc_client::Job, Box<dyn std::error::Error>> {
JobBuilder::new()
.caller_id("integration_test")
.context_id(context)
.payload("echo 'Test job output'")
.executor("osis")
.runner("osis_runner_1")
.timeout(30)
.env_var("TEST_VAR", "test_value")
.build()
.map_err(|e| e.into())
}
/// Test helper to check if supervisor is available
async fn is_supervisor_available() -> bool {
match SupervisorClient::new("http://localhost:3030") {
Ok(client) => client.discover().await.is_ok(),
Err(_) => false,
}
}
#[tokio::test]
async fn test_jobs_create_and_start() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let secret = "user-secret-456";
let job = create_test_job("create_and_start").unwrap();
// Test jobs.create
let job_id = client.jobs_create(secret, job).await.unwrap();
assert!(!job_id.is_empty());
// Test job.start
let result = client.job_start(secret, &job_id).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_job_status_monitoring() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let secret = "user-secret-456";
let job = create_test_job("status_monitoring").unwrap();
let job_id = client.jobs_create(secret, job).await.unwrap();
client.job_start(secret, &job_id).await.unwrap();
// Test job.status
let mut attempts = 0;
let max_attempts = 10;
while attempts < max_attempts {
let status = client.job_status(&job_id).await.unwrap();
assert!(!status.job_id.is_empty());
assert!(!status.status.is_empty());
assert!(!status.created_at.is_empty());
if status.status == "completed" || status.status == "failed" {
break;
}
attempts += 1;
sleep(Duration::from_secs(1)).await;
}
}
#[tokio::test]
async fn test_job_result_retrieval() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let secret = "user-secret-456";
let job = create_test_job("result_retrieval").unwrap();
let job_id = client.jobs_create(secret, job).await.unwrap();
client.job_start(secret, &job_id).await.unwrap();
// Wait a bit for job to complete
sleep(Duration::from_secs(3)).await;
// Test job.result
let result = client.job_result(&job_id).await.unwrap();
match result {
JobResult::Success { success } => {
assert!(!success.is_empty());
},
JobResult::Error { error } => {
assert!(!error.is_empty());
}
}
}
#[tokio::test]
async fn test_job_run_immediate() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let secret = "user-secret-456";
let job = create_test_job("immediate_run").unwrap();
// Test job.run (immediate execution)
let result = client.job_run(secret, job).await.unwrap();
match result {
JobResult::Success { success } => {
assert!(!success.is_empty());
},
JobResult::Error { error } => {
assert!(!error.is_empty());
}
}
}
#[tokio::test]
async fn test_jobs_list() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
// Test jobs.list
let job_ids = client.jobs_list().await.unwrap();
// Should return a vector (might be empty)
assert!(job_ids.len() >= 0);
}
#[tokio::test]
async fn test_authentication_failures() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let invalid_secret = "invalid-secret-123";
let job = create_test_job("auth_failure").unwrap();
// Test that invalid secrets fail
let result = client.jobs_create(invalid_secret, job.clone()).await;
assert!(result.is_err());
let result = client.job_run(invalid_secret, job.clone()).await;
assert!(result.is_err());
let result = client.job_start(invalid_secret, "fake-job-id").await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_nonexistent_job_operations() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let fake_job_id = format!("nonexistent-{}", Uuid::new_v4());
// Test operations on nonexistent job should fail
let result = client.job_status(&fake_job_id).await;
assert!(result.is_err());
let result = client.job_result(&fake_job_id).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_complete_workflow() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let secret = "user-secret-456";
let job = create_test_job("complete_workflow").unwrap();
// Complete workflow test
let job_id = client.jobs_create(secret, job).await.unwrap();
client.job_start(secret, &job_id).await.unwrap();
// Monitor until completion
let mut final_status = String::new();
for _ in 0..15 {
let status = client.job_status(&job_id).await.unwrap();
final_status = status.status.clone();
if final_status == "completed" || final_status == "failed" || final_status == "timeout" {
break;
}
sleep(Duration::from_secs(1)).await;
}
// Get final result
let result = client.job_result(&job_id).await.unwrap();
match result {
JobResult::Success { .. } => {
assert_eq!(final_status, "completed");
},
JobResult::Error { .. } => {
assert!(final_status == "failed" || final_status == "timeout");
}
}
}
#[tokio::test]
async fn test_batch_job_processing() {
if !is_supervisor_available().await {
println!("Skipping test - supervisor not available");
return;
}
let client = SupervisorClient::new("http://localhost:3030").unwrap();
let secret = "user-secret-456";
let job_count = 3;
let mut job_ids = Vec::new();
// Create multiple jobs
for i in 0..job_count {
let job = JobBuilder::new()
.caller_id("integration_test")
.context_id(&format!("batch_job_{}", i))
.payload(&format!("echo 'Batch job {}'", i))
.executor("osis")
.runner("osis_runner_1")
.timeout(30)
.build()
.unwrap();
let job_id = client.jobs_create(secret, job).await.unwrap();
job_ids.push(job_id);
}
// Start all jobs
for job_id in &job_ids {
client.job_start(secret, job_id).await.unwrap();
}
// Wait for all jobs to complete
sleep(Duration::from_secs(5)).await;
// Collect all results
let mut results = Vec::new();
for job_id in &job_ids {
let result = client.job_result(job_id).await.unwrap();
results.push(result);
}
// Verify we got results for all jobs
assert_eq!(results.len(), job_count);
}