fix(core): resolve scheduler hang and improve policy violation visibility

This PR addresses three core issues with the Policy Engine and Scheduler:
1. Scheduler Hang: Removed a redundant MessageBus listener in Scheduler.ts that caused race conditions in TTY environments.
2. Policy Visibility: Updated ToolGroupMessage.tsx to always display policy violation errors, regardless of verbosity settings.
3. User Feedback: Added emitFeedback to MessageBus.ts to ensure blocked tool calls are reported to the UI.

packages/core/src/confirmation-bus/message-bus.ts

@@ -11,6 +11,7 @@ import { PolicyDecision } from '../policy/types.js';
 import { MessageBusType, type Message } from './types.js';
 import { safeJsonStringify } from '../utils/safeJsonStringify.js';
 import { debugLogger } from '../utils/debugLogger.js';
+import { coreEvents } from '../utils/events.js';
 
 export class MessageBus extends EventEmitter {
   constructor(
@@ -70,6 +71,10 @@ export class MessageBus extends EventEmitter {
             break;
           case PolicyDecision.DENY:
             // Emit both rejection and response messages
+            coreEvents.emitFeedback(
+              'error',
+              `Tool call "${message.toolCall.name}" was blocked by policy.`,
+            );
             this.emitMessage({
               type: MessageBusType.TOOL_POLICY_REJECTION,
               toolCall: message.toolCall,

packages/core/src/scheduler/scheduler.ts

@@ -35,11 +35,7 @@ import { runInDevTraceSpan } from '../telemetry/trace.js';
 import { logToolCall } from '../telemetry/loggers.js';
 import { ToolCallEvent } from '../telemetry/types.js';
 import type { EditorType } from '../utils/editor.js';
-import {
-  MessageBusType,
-  type SerializableConfirmationDetails,
-  type ToolConfirmationRequest,
-} from '../confirmation-bus/types.js';
+import { type SerializableConfirmationDetails } from '../confirmation-bus/types.js';
 import { runWithToolCallContext } from '../utils/toolCallContext.js';
 import {
   coreEvents,
@@ -91,9 +87,6 @@ const createErrorResponse = (
  * Coordinates execution via state updates and event listening.
  */
 export class Scheduler {
-  // Tracks which MessageBus instances have the legacy listener attached to prevent duplicates.
-  private static subscribedMessageBuses = new WeakSet<MessageBus>();
-
   private readonly state: SchedulerStateManager;
   private readonly executor: ToolExecutor;
   private readonly modifier: ToolModificationHandler;
@@ -127,8 +120,6 @@ export class Scheduler {
     this.executor = new ToolExecutor(this.context);
     this.modifier = new ToolModificationHandler();
 
-    this.setupMessageBusListener(this.messageBus);
-
     coreEvents.on(CoreEvent.McpProgress, this.handleMcpProgress);
   }
 
@@ -161,28 +152,6 @@ export class Scheduler {
     });
   };
 
-  private setupMessageBusListener(messageBus: MessageBus): void {
-    if (Scheduler.subscribedMessageBuses.has(messageBus)) {
-      return;
-    }
-
-    // TODO: Optimize policy checks. Currently, tools check policy via
-    // MessageBus even though the Scheduler already checked it.
-    messageBus.subscribe(
-      MessageBusType.TOOL_CONFIRMATION_REQUEST,
-      async (request: ToolConfirmationRequest) => {
-        await messageBus.publish({
-          type: MessageBusType.TOOL_CONFIRMATION_RESPONSE,
-          correlationId: request.correlationId,
-          confirmed: false,
-          requiresUserConfirmation: true,
-        });
-      },
-    );
-
-    Scheduler.subscribedMessageBuses.add(messageBus);
-  }
-
   /**
    * Schedules a batch of tool calls.
    * @returns A promise that resolves with the results of the completed batch.