nushell-plugins/nu_plugin_orchestrator/IMPLEMENTATION_PLAN.md

nu_plugin_orchestrator - Implementation Plan

Created: 2025-10-08 Status: Base structure complete, ready for implementation Pattern: Follows nu_plugin_tera exactly

---

Phase 1: Base Structure ✅ COMPLETE

Files Created

• Cargo.toml - Package configuration with correct dependencies
• src/main.rs - Plugin entry point with 3 commands
• src/helpers.rs - Helper functions (placeholders)
• src/tests.rs - Test infrastructure
• README.md - Comprehensive documentation
• VERIFICATION.md - Structure verification

Commands Implemented (Placeholders)

• orch status [--data-dir <path>] - Orchestrator status from local files
• orch validate <workflow.k> [--strict] - Workflow validation locally
• orch tasks [--status <status>] [--limit <n>] - Task listing from queue

Verification

cd provisioning/core/plugins/nushell-plugins/nu_plugin_orchestrator
cargo check  # ✅ SUCCESS - 264 packages locked, compiling

---

Phase 2: Core Implementation (Next Steps)

2.1 File Reading - src/helpers.rs

read_local_status() Implementation

pub fn read_local_status(data_dir: &PathBuf) -> Result<OrchStatus, String> {
    let status_file = data_dir.join("status.json");

    if !status_file.exists() {
        return Err(format!("Status file not found: {}", status_file.display()));
    }

    let content = std::fs::read_to_string(&status_file)
        .map_err(|e| format!("Failed to read status file: {}", e))?;

    serde_json::from_str(&content)
        .map_err(|e| format!("Failed to parse status JSON: {}", e))
}

Test Data: provisioning/platform/orchestrator/data/status.json

{
  "running": true,
  "tasks_pending": 5,
  "tasks_running": 2,
  "last_check": "2025-10-08T12:00:00Z"
}

read_task_queue() Implementation

use walkdir::WalkDir;

pub fn read_task_queue(
    data_dir: &PathBuf,
    status_filter: Option<String>,
) -> Result<Vec<TaskInfo>, String> {
    let tasks_dir = data_dir.join("tasks");

    if !tasks_dir.exists() {
        return Ok(vec![]);
    }

    let mut tasks = Vec::new();

    for entry in WalkDir::new(&tasks_dir)
        .max_depth(1)
        .into_iter()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().map_or(false, |ext| ext == "json"))
    {
        let content = std::fs::read_to_string(entry.path())
            .map_err(|e| format!("Failed to read task file: {}", e))?;

        let task: TaskInfo = serde_json::from_str(&content)
            .map_err(|e| format!("Failed to parse task JSON: {}", e))?;

        // Filter by status if provided
        if let Some(ref filter) = status_filter {
            if task.status != *filter {
                continue;
            }
        }

        tasks.push(task);
    }

    // Sort by priority (higher first), then by created_at (older first)
    tasks.sort_by(|a, b| {
        b.priority.cmp(&a.priority)
            .then_with(|| a.created_at.cmp(&b.created_at))
    });

    Ok(tasks)
}

Test Data: provisioning/platform/orchestrator/data/tasks/task-001.json

{
  "id": "task-001",
  "status": "pending",
  "created_at": "2025-10-08T12:00:00Z",
  "priority": 5
}

2.2 Command Integration - src/main.rs

OrchStatus::run() Update

fn run(
    &self,
    _plugin: &OrchestratorPlugin,
    _engine: &EngineInterface,
    call: &EvaluatedCall,
    _input: &Value,
) -> Result<Value, LabeledError> {
    let data_dir = if let Some(dir_str) = call.get_flag::<String>("data-dir")? {
        PathBuf::from(dir_str)
    } else {
        helpers::get_orchestrator_data_dir()
    };

    let status = helpers::read_local_status(&data_dir)
        .map_err(|e| LabeledError::new(format!("Failed to read status: {}", e)))?;

    Ok(Value::record(
        record! {
            "running" => Value::bool(status.running, call.head),
            "tasks_pending" => Value::int(status.tasks_pending as i64, call.head),
            "tasks_running" => Value::int(status.tasks_running as i64, call.head),
            "last_check" => Value::string(status.last_check, call.head),
        },
        call.head,
    ))
}

OrchTasks::run() Update

fn run(
    &self,
    _plugin: &OrchestratorPlugin,
    _engine: &EngineInterface,
    call: &EvaluatedCall,
    _input: &Value,
) -> Result<Value, LabeledError> {
    let data_dir = helpers::get_orchestrator_data_dir();
    let status_filter = call.get_flag::<String>("status")?;
    let limit = call.get_flag::<i64>("limit")?;

    let mut tasks = helpers::read_task_queue(&data_dir, status_filter)
        .map_err(|e| LabeledError::new(format!("Failed to read tasks: {}", e)))?;

    // Apply limit if provided
    if let Some(n) = limit {
        tasks.truncate(n as usize);
    }

    let task_values: Vec<Value> = tasks.into_iter().map(|task| {
        Value::record(
            record! {
                "id" => Value::string(task.id, call.head),
                "status" => Value::string(task.status, call.head),
                "created_at" => Value::string(task.created_at, call.head),
                "priority" => Value::int(task.priority as i64, call.head),
            },
            call.head,
        )
    }).collect();

    Ok(Value::list(task_values, call.head))
}

2.3 Testing - src/tests.rs

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use tempfile::tempdir;

    #[test]
    fn test_data_dir_path() {
        let dir = helpers::get_orchestrator_data_dir();
        assert!(dir.to_string_lossy().contains("orchestrator/data"));
    }

    #[test]
    fn test_read_status_file_not_found() {
        let dir = tempdir().unwrap();
        let result = helpers::read_local_status(&dir.path().to_path_buf());
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("Status file not found"));
    }

    #[test]
    fn test_read_status_valid() {
        let dir = tempdir().unwrap();
        let status_file = dir.path().join("status.json");
        fs::write(&status_file, r#"{
            "running": true,
            "tasks_pending": 5,
            "tasks_running": 2,
            "last_check": "2025-10-08T12:00:00Z"
        }"#).unwrap();

        let status = helpers::read_local_status(&dir.path().to_path_buf()).unwrap();
        assert_eq!(status.running, true);
        assert_eq!(status.tasks_pending, 5);
        assert_eq!(status.tasks_running, 2);
    }

    #[test]
    fn test_read_task_queue_empty() {
        let dir = tempdir().unwrap();
        let tasks = helpers::read_task_queue(&dir.path().to_path_buf(), None).unwrap();
        assert_eq!(tasks.len(), 0);
    }

    #[test]
    fn test_read_task_queue_with_filter() {
        let dir = tempdir().unwrap();
        let tasks_dir = dir.path().join("tasks");
        fs::create_dir(&tasks_dir).unwrap();

        // Create test tasks
        fs::write(tasks_dir.join("task-001.json"), r#"{
            "id": "task-001",
            "status": "pending",
            "created_at": "2025-10-08T12:00:00Z",
            "priority": 5
        }"#).unwrap();

        fs::write(tasks_dir.join("task-002.json"), r#"{
            "id": "task-002",
            "status": "running",
            "created_at": "2025-10-08T12:01:00Z",
            "priority": 3
        }"#).unwrap();

        // Filter by pending
        let tasks = helpers::read_task_queue(
            &dir.path().to_path_buf(),
            Some("pending".to_string())
        ).unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0].id, "task-001");
    }
}

---

Phase 3: KCL Validation (Advanced)

3.1 Research KCL Validation

• Investigate kcl-lang/kcl Rust bindings
• Determine if we can call KCL parser directly
• Alternative: Shell out to kcl binary

3.2 Implement validate_kcl_workflow()

pub fn validate_kcl_workflow(workflow_path: &str, strict: bool) -> Result<ValidationResult, String> {
    // Option 1: Shell out to kcl binary
    let output = std::process::Command::new("kcl")
        .arg("vet")
        .arg(workflow_path)
        .output()
        .map_err(|e| format!("Failed to run kcl: {}", e))?;

    if output.status.success() {
        Ok(ValidationResult {
            valid: true,
            errors: vec![],
            warnings: vec![],
        })
    } else {
        let stderr = String::from_utf8_lossy(&output.stderr);
        Ok(ValidationResult {
            valid: false,
            errors: vec![stderr.to_string()],
            warnings: vec![],
        })
    }
}

3.3 Update OrchValidate::run()

fn run(
    &self,
    _plugin: &OrchestratorPlugin,
    _engine: &EngineInterface,
    call: &EvaluatedCall,
    _input: &Value,
) -> Result<Value, LabeledError> {
    let workflow: String = call.req(0)?;
    let strict = call.has_flag("strict")?;

    let result = helpers::validate_kcl_workflow(&workflow, strict)
        .map_err(|e| LabeledError::new(format!("Validation failed: {}", e)))?;

    Ok(Value::record(
        record! {
            "valid" => Value::bool(result.valid, call.head),
            "errors" => Value::list(
                result.errors.into_iter()
                    .map(|e| Value::string(e, call.head))
                    .collect(),
                call.head
            ),
            "warnings" => Value::list(
                result.warnings.into_iter()
                    .map(|w| Value::string(w, call.head))
                    .collect(),
                call.head
            ),
        },
        call.head,
    ))
}

---

Phase 4: Enhancements (Optional)

4.1 Caching

• Add in-memory cache for status.json (TTL: 1 second)
• Add cache invalidation on file modification
• Add --no-cache flag

4.2 Advanced Filtering

• Add date range filtering for tasks (--from, --to)
• Add priority range filtering (--min-priority, --max-priority)
• Add workflow path filtering (--workflow)

4.3 Statistics

• Add orch stats command for task statistics
• Group by status, priority, date
• Show average execution time

4.4 Monitoring

• Add orch watch command for real-time updates
• Use file system notifications (notify crate)
• Stream changes to stdout

---

Testing Strategy

Unit Tests (src/tests.rs)

• Test data directory path construction
• Test file reading with mock data
• Test error handling (missing files, invalid JSON)
• Test filtering and sorting logic
• Test KCL validation (if implemented)

Integration Tests (tests/integration_tests.rs)

• Test with real orchestrator data structure
• Test concurrent access to files
• Test performance with large datasets
• Test error messages are user-friendly

Plugin Tests (using nu-plugin-test-support)

• Test command signatures are correct
• Test command outputs match expected types
• Test error cases return proper LabeledError

Benchmarks (benches/benchmark.rs)

• Benchmark file reading performance
• Compare with REST API calls
• Measure memory usage
• Test with varying dataset sizes

---

Performance Targets

Operation	Target	Baseline (REST API)	Improvement
orch status	<5ms	~50ms	10x
orch tasks	<10ms	~30ms	3x
orch validate	<20ms	~100ms	5x

---

Installation & Usage

Build

cd provisioning/core/plugins/nushell-plugins
cargo build -p nu_plugin_orchestrator --release

Install

plugin add target/release/nu_plugin_orchestrator
plugin use orchestrator

Verify

# List commands
help commands | where name =~ "orch"

# Test status
orch status

# Test with custom data dir
orch status --data-dir ./test-data

# Test tasks
orch tasks
orch tasks --status pending
orch tasks --status pending --limit 10

# Test validation (once implemented)
orch validate workflows/example.k
orch validate workflows/example.k --strict

---

Success Criteria

Phase 1 (Complete) ✅

• Structure follows nu_plugin_tera pattern
• Cargo check passes
• 3 commands implemented (placeholders)
• README documentation complete
• Verification document created

Phase 2 (Core Implementation)

• read_local_status() reads real files
• read_task_queue() reads task directory
• Commands return real data (not placeholders)
• Error handling is comprehensive
• Unit tests pass (>80% coverage)

Phase 3 (KCL Validation)

• validate_kcl_workflow() validates KCL syntax
• Validation errors are user-friendly
• Strict mode performs additional checks
• Integration with KCL tooling

Phase 4 (Production Ready)

• Performance targets met
• Integration tests pass
• Benchmarks show 5-10x improvement
• Documentation updated with real examples
• Plugin registered in project

---

Dependencies

Required

• nu-plugin (0.107.1) - Plugin SDK
• nu-protocol (0.107.1) - Nushell types
• serde (1.0) - Serialization
• serde_json (1.0) - JSON parsing
• chrono (0.4) - Timestamps
• walkdir (2.5) - Directory traversal

Optional (Phase 4)

• notify - File system notifications
• kcl-lang - KCL validation (if Rust bindings available)
• criterion - Benchmarking
• cache - In-memory caching

---

Risk Mitigation

Risk 1: Orchestrator Data Format Changes

Mitigation: Version check in status.json, graceful degradation

Risk 2: File Access Permissions

Mitigation: Clear error messages, suggest fixing permissions

Risk 3: Large Task Queues

Mitigation: Streaming, pagination, default limits

Risk 4: KCL Integration Complexity

Mitigation: Shell out to kcl binary as fallback

---

Timeline Estimate

• Phase 1 (Base Structure): ✅ Complete (1 hour)
• Phase 2 (Core Implementation): ~4 hours
• Phase 3 (KCL Validation): ~2 hours
• Phase 4 (Enhancements): ~4 hours

Total: ~11 hours (1 day of focused work)

---

Next Steps: Implement Phase 2 (Core Implementation) starting with read_local_status() in src/helpers.rs.