feat(core): implement onOutput for batch processes

Author: leosvelperez

packages/nx/src/native/index.d.ts

@@ -23,6 +23,9 @@ export declare class AppLifeCycle {
   registerForcedShutdownCallback(forcedShutdownCallback: () => any): void
   __setCloudMessage(message: string): Promise<void>
   setEstimatedTaskTimings(timings: Record<string, number>): void
+  registerRunningBatch(batchId: string, batchInfo: BatchInfo): void
+  appendBatchOutput(batchId: string, output: string): void
+  setBatchStatus(batchId: string, status: string): void
 }
 
 export declare class ChildProcess {
@@ -193,6 +196,11 @@ export declare class WorkspaceContext {
   getFilesInDirectory(directory: string): Array<string>
 }
 
+export interface BatchInfo {
+  executorName: string
+  taskIds: Array<string>
+}
+
 export interface CachedResult {
   code: number
   terminalOutput?: string

packages/nx/src/native/tui/action.rs

@@ -1,6 +1,6 @@
 use crate::native::tasks::types::{Task, TaskResult};
 
-use super::{app::Focus, components::tasks_list::TaskStatus};
+use super::{app::{BatchInfo, BatchStatus, Focus}, components::tasks_list::TaskStatus};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Action {
@@ -36,4 +36,7 @@ pub enum Action {
     SendConsoleMessage(String),
     ConsoleMessengerAvailable(bool),
     EndCommand,
+    StartBatch(String, BatchInfo),
+    UpdateBatchStatus(String, BatchStatus),
+    AppendBatchOutput(String, String),
 }

packages/nx/src/native/tui/app.rs

@@ -45,6 +45,25 @@ use crate::native::ide::nx_console::messaging::NxConsoleMessageConnection;
 use crate::native::tui::graph_utils::get_failed_dependencies;
 use crate::native::utils::time::current_timestamp_millis;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BatchStatus {
+    Running,
+    Success, 
+    Failure,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BatchInfo {
+    pub executor_name: String,
+    pub task_ids: Vec<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct BatchState {
+    pub info: BatchInfo,     // executor_name, task_ids  
+    pub status: BatchStatus, // Running/Success/Failure
+}
+
 pub struct App {
     pub components: Vec<Box<dyn Component>>,
     pub quit_at: Option<std::time::Instant>,
@@ -79,6 +98,9 @@ pub struct App {
     debug_state: TuiWidgetState,
     console_messenger: Option<NxConsoleMessageConnection>,
     estimated_task_timings: HashMap<String, i64>,
+    // Batch tracking
+    batch_states: HashMap<String, BatchState>,               // batch_id → BatchState
+    batch_pty_instances: HashMap<String, Arc<PtyInstance>>,   // batch_id → PTY
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -162,6 +184,9 @@ impl App {
             debug_state: TuiWidgetState::default().set_default_display_level(LevelFilter::Debug),
             console_messenger: None,
             estimated_task_timings: HashMap::new(),
+            // Initialize batch tracking
+            batch_states: HashMap::new(),
+            batch_pty_instances: HashMap::new(),
         })
     }
 
@@ -377,6 +402,44 @@ impl App {
         pty.process_output(output.as_bytes());
     }
 
+    // Batch methods
+    pub fn register_running_batch(&mut self, batch_id: String, batch_info: BatchInfo) {        
+        // Create PTY instance for batch output
+        let pty = PtyInstance::non_interactive();
+        self.batch_pty_instances.insert(batch_id.clone(), Arc::new(pty));
+        
+        // Store batch state with initial Running status
+        self.batch_states.insert(batch_id.clone(), BatchState {
+            info: batch_info.clone(),
+            status: BatchStatus::Running, // Always starts as Running
+        });
+        
+        // Trigger resize for new PTY instance
+        let _ = self.debounce_pty_resize();
+        
+        self.dispatch_action(Action::StartBatch(batch_id, batch_info));
+    }
+
+    pub fn append_batch_output(&mut self, batch_id: String, output: String) {
+        if let Some(pty) = self.batch_pty_instances.get_mut(&batch_id) {
+            pty.process_output(output.as_bytes());
+        }
+        self.dispatch_action(Action::AppendBatchOutput(batch_id, output));
+    }
+
+    pub fn set_batch_status(&mut self, batch_id: String, status_str: String) {
+        if let Some(batch_state) = self.batch_states.get_mut(&batch_id) {
+            let new_status = match status_str.as_str() {
+                "success" => BatchStatus::Success,
+                "failure" => BatchStatus::Failure, 
+                _ => BatchStatus::Running,
+            };
+            
+            batch_state.status = new_status;
+            self.dispatch_action(Action::UpdateBatchStatus(batch_id, new_status));
+        }
+    }
+
     pub fn handle_event(
         &mut self,
         event: tui::Event,
@@ -1512,6 +1575,16 @@ impl App {
             }
         }
 
+        for pty in self.batch_pty_instances.values() {
+            // Use the first terminal pane area as reference for batch PTY dimensions
+            // This ensures batch PTY instances have consistent dimensions with task PTYs
+            if let Some(first_pane_area) = self.layout_areas.as_ref().unwrap().terminal_panes.first() {
+                let (pty_height, pty_width) = TerminalPane::calculate_pty_dimensions(*first_pane_area);
+                let mut pty_clone = pty.as_ref().clone();
+                pty_clone.resize(pty_height, pty_width)?;
+            }
+        }
+
         // Sort tasks if needed after all resizing is complete
         if needs_sort {
             self.dispatch_action(Action::SortTasks);

packages/nx/src/native/tui/lifecycle.rs

@@ -12,7 +12,7 @@ use crate::native::{
     pseudo_terminal::pseudo_terminal::{ParserArc, WriterArc},
 };
 
-use super::app::App;
+use super::app::{App, BatchInfo as RustBatchInfo};
 use super::components::tasks_list::TaskStatus;
 use super::config::{AutoExit, TuiCliArgs as RustTuiCliArgs, TuiConfig as RustTuiConfig};
 use super::tui::Tui;
@@ -63,6 +63,22 @@ pub enum RunMode {
     RunMany,
 }
 
+#[napi(object)]
+#[derive(Clone)]
+pub struct BatchInfo {
+    pub executor_name: String,
+    pub task_ids: Vec<String>,
+}
+
+impl From<BatchInfo> for RustBatchInfo {
+    fn from(js: BatchInfo) -> Self {
+        Self {
+            executor_name: js.executor_name,
+            task_ids: js.task_ids,
+        }
+    }
+}
+
 #[napi]
 #[derive(Clone)]
 pub struct AppLifeCycle {
@@ -308,6 +324,25 @@ impl AppLifeCycle {
         self.app.lock().set_estimated_task_timings(timings);
         Ok(())
     }
+
+    // Batch lifecycle methods
+    #[napi]
+    pub fn register_running_batch(&mut self, batch_id: String, batch_info: BatchInfo) -> napi::Result<()> {
+        self.app.lock().register_running_batch(batch_id, batch_info.into());
+        Ok(())
+    }
+
+    #[napi] 
+    pub fn append_batch_output(&mut self, batch_id: String, output: String) -> napi::Result<()> {
+        self.app.lock().append_batch_output(batch_id, output);
+        Ok(())
+    }
+
+    #[napi]
+    pub fn set_batch_status(&mut self, batch_id: String, status: String) -> napi::Result<()> {
+        self.app.lock().set_batch_status(batch_id, status);
+        Ok(())
+    }
 }
 
 #[napi]

packages/nx/src/tasks-runner/forked-process-task-runner.ts

@@ -63,7 +63,7 @@ export class ForkedProcessTaskRunner {
     }
 
     const p = fork(workerPath, {
-      stdio: ['inherit', 'inherit', 'inherit', 'ipc'],
+      stdio: ['inherit', 'pipe', 'pipe', 'ipc'],
       env: {
         ...env,
         NX_FORKED_TASK_EXECUTOR: 'true',

packages/nx/src/tasks-runner/life-cycle.ts

@@ -1,6 +1,5 @@
 import { Task } from '../config/task-graph';
 import { ExternalObject, TaskStatus as NativeTaskStatus } from '../native';
-import { RunningTask } from './running-tasks/running-task';
 import { TaskStatus } from './tasks-runner';
 
 /**
@@ -22,6 +21,13 @@ export interface TaskMetadata {
   groupId: number;
 }
 
+export interface BatchInfo {
+  executorName: string;
+  taskIds: string[];
+}
+
+export type BatchStatus = 'running' | 'success' | 'failure';
+
 export interface LifeCycle {
   startCommand?(parallel?: number): void | Promise<void>;
 
@@ -70,6 +76,13 @@ export interface LifeCycle {
   registerForcedShutdownCallback?(callback: () => void): void;
 
   setEstimatedTaskTimings?(timings: Record<string, number>): void;
+
+  // Batch-specific lifecycle methods
+  registerRunningBatch?(batchId: string, batchInfo: BatchInfo): void;
+
+  appendBatchOutput?(batchId: string, output: string): void;
+
+  setBatchStatus?(batchId: string, status: BatchStatus): void;
 }
 
 export class CompositeLifeCycle implements LifeCycle {
@@ -200,4 +213,28 @@ export class CompositeLifeCycle implements LifeCycle {
       }
     }
   }
+
+  registerRunningBatch(batchId: string, batchInfo: BatchInfo): void {
+    for (let l of this.lifeCycles) {
+      if (l.registerRunningBatch) {
+        l.registerRunningBatch(batchId, batchInfo);
+      }
+    }
+  }
+
+  appendBatchOutput(batchId: string, output: string): void {
+    for (let l of this.lifeCycles) {
+      if (l.appendBatchOutput) {
+        l.appendBatchOutput(batchId, output);
+      }
+    }
+  }
+
+  setBatchStatus(batchId: string, status: BatchStatus): void {
+    for (let l of this.lifeCycles) {
+      if (l.setBatchStatus) {
+        l.setBatchStatus(batchId, status);
+      }
+    }
+  }
 }

packages/nx/src/tasks-runner/running-tasks/batch-process.ts

@@ -9,6 +9,8 @@ import { signalToCode } from '../../utils/exit-codes';
 export class BatchProcess {
   private exitCallbacks: Array<(code: number) => void> = [];
   private resultsCallbacks: Array<(results: BatchResults) => void> = [];
+  private outputCallbacks: Array<(output: string) => void> = [];
+  private terminalOutput: string = '';
 
   constructor(
     private childProcess: ChildProcess,
@@ -41,6 +43,38 @@ export class BatchProcess {
         cb(code);
       }
     });
+
+    // Capture stdout output
+    if (this.childProcess.stdout) {
+      this.childProcess.stdout.on('data', (chunk) => {
+        const output = chunk.toString();
+        this.terminalOutput += output;
+
+        // Maintain current terminal output behavior
+        process.stdout.write(chunk);
+
+        // Notify callbacks for TUI
+        for (const cb of this.outputCallbacks) {
+          cb(output);
+        }
+      });
+    }
+
+    // Capture stderr output
+    if (this.childProcess.stderr) {
+      this.childProcess.stderr.on('data', (chunk) => {
+        const output = chunk.toString();
+        this.terminalOutput += output;
+
+        // Maintain current terminal output behavior
+        process.stderr.write(chunk);
+
+        // Notify callbacks for TUI
+        for (const cb of this.outputCallbacks) {
+          cb(output);
+        }
+      });
+    }
   }
 
   onExit(cb: (code: number) => void) {
@@ -51,6 +85,10 @@ export class BatchProcess {
     this.resultsCallbacks.push(cb);
   }
 
+  onOutput(cb: (output: string) => void) {
+    this.outputCallbacks.push(cb);
+  }
+
   async getResults(): Promise<BatchResults> {
     return Promise.race<BatchResults>([
       new Promise((_, rej) => {
@@ -81,4 +119,8 @@ export class BatchProcess {
       this.childProcess.kill(signal);
     }
   }
+
+  getTerminalOutput(): string {
+    return this.terminalOutput;
+  }
 }