# Runtime Task Instance Creation and Communication Channel Flow

## Overview

In spear-next, the process of runtime creating task instances and obtaining communication channels is a multi-layered process involving collaboration between multiple components including task execution manager, runtime manager, communication factory, and others.

## Core Components

### 1. TaskExecutionManager
- Responsible for overall task execution flow management
- Manages the lifecycle of artifacts, tasks, and instances
- Coordinates runtime and communication components

### 2. RuntimeManager
- Manages different types of runtimes (Process, Kubernetes, WASM)
- Provides interfaces for runtime instance creation and management

### 3. CommunicationFactory
- Creates appropriate communication channels for different runtime types
- Supports multiple communication methods: UnixSocket, TCP, gRPC, etc.
- Manages communication channel lifecycle and connection pooling

## Detailed Flow

### Phase 1: Request Processing and Task Preparation

1. **Receive Execution Request**
   ```rust
   // In TaskExecutionManager::submit_execution
   pub async fn submit_execution(
       &self,
       request: InvokeFunctionRequest,
   ) -> ExecutionResult<super::ExecutionResponse>
   ```

2. **Prepare Artifact and Task via Events**
   - Artifact/Task are ensured before invocation in the TaskEvents subscriber:
   ```rust
   // In TaskEventSubscriber::execute_task
   let artifact = mgr.ensure_artifact_from_sms(&sms_task).await?;
   let _ = mgr.ensure_task_from_sms(&sms_task, &artifact).await?;
   // Then submit ExistingTask execution
   let _ = mgr.submit_execution(req).await;
   ```

3. **Execution Path Constraint**
   - `execute_request` only supports invoking existing tasks; it does not create artifacts or tasks.
   - If the task is missing locally on the target node, Spearlet fetches it from SMS on-demand and materializes it before executing (see `TaskExecutionManager::fetch_and_materialize_task_from_sms`).

### Phase 2: Instance Creation and Startup

4. **Get or Create Instance**
   ```rust
   async fn get_or_create_instance(&self, task: &Arc<Task>) -> ExecutionResult<Arc<TaskInstance>> {
       // Try to find an available instance
       if let Some(instance) = self.scheduler.select_instance(task).await? {
           return Ok(instance);
       }

       // Check instance limit
       if task.instance_count() >= self.config.max_instances_per_task {
           return Err(ExecutionError::ResourceExhausted { ... });
       }

       // Create new instance
       let instance_id = task.generate_instance_id();
       let runtime = self.runtime_manager.get_runtime(&task.spec.runtime_type)
           .ok_or_else(|| ExecutionError::RuntimeError { ... })?;

       let instance_config = task.create_instance_config();
       let instance = runtime.create_instance(&instance_config).await?;

       // Start the instance
       runtime.start_instance(&instance).await?;
       // Note: semantics are runtime-specific (e.g. ProcessRuntime spawns a process; WasmRuntime may start a worker thread when built with wasmedge)

       // Register instance
       self.instances.insert(instance_id.clone(), instance.clone());
       task.add_instance(instance.clone())?;
       self.scheduler.add_instance(instance.clone()).await?;

       Ok(instance)
   }
   ```

### Phase 3: Runtime Instance Creation

5. **Runtime Creates Instance**
   Using ProcessRuntime as an example:
   ```rust
   async fn create_instance(
       &self,
       config: &InstanceConfig,
   ) -> ExecutionResult<Arc<TaskInstance>> {
       let instance_id = format!("proc-{}", Uuid::new_v4());
       let instance = Arc::new(TaskInstance::new(instance_id, config.clone()));

       // Create process handle
       let handle = ProcessHandle {
           pid: 0, // Will be set during startup
           command: config.runtime_config.get("command")
               .and_then(|v| v.as_str())
               .unwrap_or(&self.config.default_executable)
               .to_string(),
           args: config.runtime_config.get("args")
               .and_then(|v| v.as_array())
               .map(|arr| arr.iter().filter_map(|v| v.as_str().map(String::from)).collect())
               .unwrap_or_default(),
           working_directory: config.runtime_config.get("working_directory")
               .and_then(|v| v.as_str())
               .unwrap_or(&self.config.working_directory)
               .to_string(),
           environment: config.environment.clone(),
           start_time: SystemTime::now(),
           child: Arc::new(Mutex::new(None)),
       };

       // Set runtime handle
       instance.set_runtime_handle(handle);
       instance.set_status(InstanceStatus::Creating);

       Ok(instance)
   }
   ```

6. **Start Instance**
   ```rust
   async fn start_instance(&self, instance: &Arc<TaskInstance>) -> ExecutionResult<()> {
       let handle = instance.get_runtime_handle::<ProcessHandle>()
           .ok_or_else(|| ExecutionError::RuntimeError { ... })?;

       let mut command = self.build_process_command(&instance.config);
       
       // Start process
       let child = command.spawn()
           .map_err(|e| ExecutionError::RuntimeError { ... })?;

       let pid = child.id().unwrap_or(0);
       
       // Update handle
       *handle.child.lock().await = Some(child);
       
       instance.set_status(InstanceStatus::Ready);
       Ok(())
   }
   ```

### Phase 4: Communication Channel Acquisition

7. **Communication Channel Creation and Management**
   
   Although the current code doesn't directly show the association between communication channels and instances, based on the architectural design, the communication channel acquisition flow should be:

   ```rust
   // In CommunicationFactory
   pub async fn get_or_create_channel(
       &mut self,
       instance_id: RuntimeInstanceId,
       custom_config: Option<ChannelConfig>,
   ) -> CommunicationResult<Arc<dyn CommunicationChannel>> {
       // Check if we already have an active channel
       if self.pool_enabled {
           if let Some(channel) = self.active_channels.get(&instance_id) {
               if channel.is_connected().await {
                   return Ok(channel.clone());
               } else {
                   self.active_channels.remove(&instance_id);
               }
           }
       }

       // Create new channel
       let channel = self.create_channel_for_instance(instance_id.clone(), custom_config).await?;
       
       // Store in pool
       if self.pool_enabled {
           self.active_channels.insert(instance_id, channel.clone());
       }
       
       Ok(channel)
   }
   ```

8. **Select Communication Strategy Based on Runtime Type**
   ```rust
   fn init_default_strategies(&mut self) {
       // Process runtime strategy
       let process_strategy = CommunicationStrategy::new(
           RuntimeType::Process,
           "unix".to_string(),
           vec!["tcp".to_string()],
           ChannelConfig {
               channel_type: "unix".to_string(),
               address: "/tmp/spear-process.sock".to_string(),
               ..Default::default()
           },
       );
       self.strategies.insert(RuntimeType::Process, process_strategy);
       
       // Kubernetes runtime strategy
       let k8s_strategy = CommunicationStrategy::new(
           RuntimeType::Kubernetes,
           "grpc".to_string(),
           vec!["tcp".to_string()],
           ChannelConfig {
               channel_type: "grpc".to_string(),
               address: "http://127.0.0.1:50051".to_string(),
               ..Default::default()
           },
       );
       self.strategies.insert(RuntimeType::Kubernetes, k8s_strategy);
       
       // WASM runtime strategy (in-process communication)
       let wasm_strategy = CommunicationStrategy::new(
           RuntimeType::Wasm,
           "memory".to_string(),
           vec![],
           ChannelConfig {
               channel_type: "memory".to_string(),
               address: "in-process".to_string(),
               ..Default::default()
           },
       );
       self.strategies.insert(RuntimeType::Wasm, wasm_strategy);
   }
   ```

### Phase 5: Execution and Communication

9. **Execute Request**
   ```rust
   async fn execute_request(
       &self,
       artifact_spec: ProtoArtifactSpec,
       execution_context: ExecutionContext,
   ) -> ExecutionResult<super::ExecutionResponse> {
       // Get instance
       let instance = self.get_or_create_instance(&task).await?;

       // Execute on instance
       let runtime = self.runtime_manager.get_runtime(&task.spec.runtime_type)?;
       let runtime_response = runtime.execute(&instance, execution_context).await?;
       
       // Convert response
       Ok(super::ExecutionResponse { ... })
   }
   ```

10. **Runtime Execution**
    ```rust
    async fn execute(
        &self,
        instance: &Arc<TaskInstance>,
        _context: ExecutionContext,
    ) -> ExecutionResult<RuntimeExecutionResponse> {
        let handle = instance.get_runtime_handle::<ProcessHandle>()?;

        // Interact with process through communication channel
        // This would use the appropriate CommunicationChannel implementation
        let child_guard = handle.child.lock().await;
        if let Some(_child) = child_guard.as_ref() {
            // Actual communication logic
            Ok(RuntimeExecutionResponse::new_sync(...))
        } else {
            Err(ExecutionError::RuntimeError { ... })
        }
    }
    ```

## Communication Channel Types

### 1. UnixSocketChannel
- Used for local inter-process communication
- High performance, low latency
- Suitable for Process Runtime

### 2. TcpChannel
- Used for network communication
- Supports remote instances
- Suitable for distributed deployments

### 3. GrpcChannel
- Used for advanced RPC communication
- Supports streaming
- Suitable for Kubernetes Runtime

## Key Data Structures

### TaskInstance
```rust
pub struct TaskInstance {
    pub id: InstanceId,
    pub task_id: TaskId,
    pub config: InstanceConfig,
    pub status: Arc<parking_lot::RwLock<InstanceStatus>>,
    pub health_status: Arc<parking_lot::RwLock<HealthStatus>>,
    pub metrics: Arc<parking_lot::RwLock<InstanceMetrics>>,
    pub runtime_handle: Arc<parking_lot::RwLock<Option<Box<dyn std::any::Any + Send + Sync>>>>,
    // ...
}
```

### RuntimeInstanceId
```rust
pub struct RuntimeInstanceId {
    pub runtime_type: RuntimeType,
    pub instance_id: String,
    pub namespace: Option<String>,
}
```

### CommunicationChannel Trait
```rust
pub trait CommunicationChannel: Send + Sync {
    async fn connect(&mut self) -> CommunicationResult<()>;
    async fn disconnect(&mut self) -> CommunicationResult<()>;
    async fn send_message(&self, message: &[u8]) -> CommunicationResult<()>;
    async fn receive_message(&self) -> CommunicationResult<Vec<u8>>;
    async fn is_connected(&self) -> bool;
    async fn close(&mut self) -> CommunicationResult<()>;
    async fn get_stats(&self) -> CommunicationResult<ChannelStats>;
}
```

## Summary

The entire flow can be summarized as:

1. **Request Reception** → TaskExecutionManager receives execution request
2. **Resource Preparation** → Create or get Artifact and Task
3. **Instance Management** → Create and start TaskInstance through RuntimeManager
4. **Communication Establishment** → CommunicationFactory creates appropriate communication channels based on Runtime type
5. **Execution Interaction** → Runtime interacts with instances through communication channels
6. **Result Return** → Return execution results to the caller

This design achieves runtime type abstraction, pluggable communication methods, and complete instance lifecycle management.
