feat(input): add mode switching between user and controller input

- Extend InputResult type to include source field
- Add mode change listeners to both user and controller input providers
- Implement immediate abort on mode switch in controller provider
- Handle mode switch signals in workflow runner
- Add proper cleanup of listeners and abort controllers

Author: moazbuilds

src/workflows/execution/runner.ts

@@ -150,6 +150,14 @@ export class WorkflowRunner {
     // Mode change listener
     const modeChangeHandler = async (data: { autonomousMode: boolean }) => {
       debug('[Runner] Mode change: autoMode=%s', data.autonomousMode);
+      // If in waiting state, let the provider's listener handle it
+      // The provider will return __SWITCH_TO_AUTO__ or __SWITCH_TO_MANUAL__
+      // and handleWaiting() will call setAutoMode()
+      if (this.machine.state === 'waiting') {
+        debug('[Runner] In waiting state, provider will handle mode switch');
+        return;
+      }
+      // In other states (running, idle), set auto mode directly
       await this.setAutoMode(data.autonomousMode);
     };
     process.on('workflow:mode-change', modeChangeHandler);
@@ -402,6 +410,14 @@ export class WorkflowRunner {
       return;
     }
 
+    // Handle special switch-to-auto signal
+    if (result.type === 'input' && result.value === '__SWITCH_TO_AUTO__') {
+      debug('[Runner] Switching to autonomous mode');
+      await this.setAutoMode(true);
+      // Re-run waiting with controller input
+      return;
+    }
+
     // Handle result
     const step = this.moduleSteps[ctx.currentStepIndex];
     const uniqueAgentId = `${step.agentId}-step-${ctx.currentStepIndex}`;
@@ -419,7 +435,7 @@ export class WorkflowRunner {
           this.machine.send({ type: 'INPUT_RECEIVED', input: '' });
         } else {
           // Has input = resume current step with input, then wait again
-          await this.resumeWithInput(result.value, result.resumeMonitoringId);
+          await this.resumeWithInput(result.value, result.resumeMonitoringId, result.source);
         }
         break;
 
@@ -442,12 +458,12 @@ export class WorkflowRunner {
   /**
    * Resume current step with input (for chained prompts or steering)
    */
-  private async resumeWithInput(input: string, monitoringId?: number): Promise<void> {
+  private async resumeWithInput(input: string, monitoringId?: number, source?: 'user' | 'controller'): Promise<void> {
     const ctx = this.machine.context;
     const step = this.moduleSteps[ctx.currentStepIndex];
     const uniqueAgentId = `${step.agentId}-step-${ctx.currentStepIndex}`;
 
-    debug('[Runner] Resuming step with input: %s...', input.slice(0, 50));
+    debug('[Runner] Resuming step with input: %s... (source=%s)', input.slice(0, 50), source ?? 'user');
 
     // Get sessionId from step data for resume
     const stepData = await getStepData(this.cmRoot, ctx.currentStepIndex);
@@ -467,8 +483,8 @@ export class WorkflowRunner {
       }
     }
 
-    // Log custom user input (magenta)
-    if (!isQueuedPrompt) {
+    // Log custom user input (magenta) - skip for controller input (already logged during streaming)
+    if (!isQueuedPrompt && source !== 'controller') {
       const formatted = formatUserInput(input);
       this.emitter.logMessage(uniqueAgentId, formatted);
       if (monitoringId !== undefined) {
@@ -480,6 +496,16 @@ export class WorkflowRunner {
     this.emitter.updateAgentStatus(uniqueAgentId, 'running');
     this.emitter.setWorkflowStatus('running');
 
+    // Track if mode switch was requested during execution
+    let modeSwitchRequested: 'manual' | 'auto' | null = null;
+    const modeChangeHandler = (data: { autonomousMode: boolean }) => {
+      debug('[Runner] Mode change during resumeWithInput: autoMode=%s', data.autonomousMode);
+      modeSwitchRequested = data.autonomousMode ? 'auto' : 'manual';
+      // Abort the current step execution
+      this.abortController?.abort();
+    };
+    process.on('workflow:mode-change', modeChangeHandler);
+
     try {
       const output = await executeStep(step, this.cwd, {
         logger: () => {},
@@ -508,11 +534,21 @@ export class WorkflowRunner {
       if (error instanceof Error && error.name === 'AbortError') {
         if (this.pauseRequested) {
           this.machine.send({ type: 'PAUSE' });
+          return;
+        }
+        // Handle mode switch during execution
+        if (modeSwitchRequested) {
+          debug('[Runner] Step aborted due to mode switch to %s', modeSwitchRequested);
+          this.emitter.updateAgentStatus(uniqueAgentId, 'checkpoint');
+          await this.setAutoMode(modeSwitchRequested === 'auto');
+          // Return to let handleWaiting loop with new provider
+          return;
         }
         return;
       }
       this.machine.send({ type: 'STEP_ERROR', error: error as Error });
     } finally {
+      process.removeListener('workflow:mode-change', modeChangeHandler);
       this.abortController = null;
     }
   }

src/workflows/input/controller.ts

@@ -46,6 +46,7 @@ export class ControllerInputProvider implements InputProvider {
   private getControllerConfig: () => Promise<ControllerConfig | null>;
   private cwd: string;
   private aborted = false;
+  private abortController: AbortController | null = null;
   private modeChangeListener: ((data: { autonomousMode: boolean }) => void) | null = null;
 
   constructor(options: ControllerInputProviderOptions) {
@@ -67,13 +68,17 @@ export class ControllerInputProvider implements InputProvider {
 
     const loggerService = AgentLoggerService.getInstance();
 
+    // Set up abort controller for this execution
+    this.abortController = new AbortController();
+
     // Listen for mode change (user switches to manual)
     let switchToManual = false;
     this.modeChangeListener = (data) => {
       if (!data.autonomousMode) {
-        debug('[Controller] Mode change to manual requested');
+        debug('[Controller] Mode change to manual requested, aborting execution');
         switchToManual = true;
-        // Note: We don't abort here - let current execution complete
+        // Abort the current execution immediately
+        this.abortController?.abort();
       }
     };
     process.on('workflow:mode-change', this.modeChangeListener);
@@ -87,6 +92,7 @@ export class ControllerInputProvider implements InputProvider {
         workingDir: this.cwd,
         resumeSessionId: config.sessionId,
         resumePrompt: prompt,
+        abortSignal: this.abortController.signal,
         logger: (chunk) => {
           // Log controller output to step's log
           if (context.stepOutput.monitoringId !== undefined) {
@@ -105,7 +111,7 @@ export class ControllerInputProvider implements InputProvider {
         return { type: 'stop' };
       }
 
-      // Check if user switched to manual mode
+      // Check if user switched to manual mode (shouldn't reach here if aborted, but just in case)
       if (switchToManual) {
         debug('[Controller] Switching to manual mode');
         this.emitter.emitCanceled();
@@ -144,9 +150,24 @@ export class ControllerInputProvider implements InputProvider {
         type: 'input',
         value: cleanedResponse,
         resumeMonitoringId: context.stepOutput.monitoringId,
+        source: 'controller',
       };
+    } catch (error) {
+      // Handle abort (mode switch to manual)
+      if (error instanceof Error && error.name === 'AbortError') {
+        if (switchToManual) {
+          debug('[Controller] Aborted due to mode switch to manual');
+          this.emitter.emitCanceled();
+          return { type: 'input', value: '__SWITCH_TO_MANUAL__' };
+        }
+        // General abort (stop workflow)
+        debug('[Controller] Aborted');
+        return { type: 'stop' };
+      }
+      throw error;
     } finally {
-      // Clean up listener
+      // Clean up
+      this.abortController = null;
       if (this.modeChangeListener) {
         process.removeListener('workflow:mode-change', this.modeChangeListener);
         this.modeChangeListener = null;
@@ -168,6 +189,10 @@ export class ControllerInputProvider implements InputProvider {
     debug('[Controller] Aborting');
     this.aborted = true;
 
+    // Abort any running execution
+    this.abortController?.abort();
+    this.abortController = null;
+
     if (this.modeChangeListener) {
       process.removeListener('workflow:mode-change', this.modeChangeListener);
       this.modeChangeListener = null;

src/workflows/input/types.ts

@@ -34,7 +34,7 @@ export interface InputContext {
  * Result from an input provider
  */
 export type InputResult =
-  | { type: 'input'; value: string; resumeMonitoringId?: number }
+  | { type: 'input'; value: string; resumeMonitoringId?: number; source?: 'user' | 'controller' }
   | { type: 'skip' }
   | { type: 'stop' };
 

src/workflows/input/user.ts

@@ -33,6 +33,7 @@ export class UserInputProvider implements InputProvider {
   private resolver: ((result: InputResult) => void) | null = null;
   private currentContext: InputContext | null = null;
   private inputListener: ((data?: { prompt?: string; skip?: boolean }) => void) | null = null;
+  private modeChangeListener: ((data: { autonomousMode: boolean }) => void) | null = null;
 
   constructor(options: UserInputProviderOptions) {
     this.emitter = options.emitter;
@@ -57,16 +58,32 @@ export class UserInputProvider implements InputProvider {
     };
     process.on('workflow:input', this.inputListener);
 
+    // Set up listener for mode change (switch to autonomous)
+    this.modeChangeListener = (data) => {
+      if (data.autonomousMode && this.resolver) {
+        debug('[UserInput] Mode change to autonomous, signaling switch');
+        this.emitter.emitCanceled();
+        this.resolver({ type: 'input', value: '__SWITCH_TO_AUTO__' });
+        this.resolver = null;
+        this.currentContext = null;
+      }
+    };
+    process.on('workflow:mode-change', this.modeChangeListener);
+
     try {
       return await new Promise<InputResult>((resolve) => {
         this.resolver = resolve;
       });
     } finally {
-      // Clean up listener
+      // Clean up listeners
       if (this.inputListener) {
         process.removeListener('workflow:input', this.inputListener);
         this.inputListener = null;
       }
+      if (this.modeChangeListener) {
+        process.removeListener('workflow:mode-change', this.modeChangeListener);
+        this.modeChangeListener = null;
+      }
     }
   }
 
@@ -141,6 +158,11 @@ export class UserInputProvider implements InputProvider {
       this.inputListener = null;
     }
 
+    if (this.modeChangeListener) {
+      process.removeListener('workflow:mode-change', this.modeChangeListener);
+      this.modeChangeListener = null;
+    }
+
     if (this.resolver) {
       debug('[UserInput] Aborting pending input wait');
       this.emitter.emitCanceled();