feat(minecraft): async concious

update dashboard

Author: shinohara-rin

services/minecraft/src/agents/planning/index.ts

@@ -138,7 +138,7 @@ export class PlanningAgentImpl extends AbstractAgent implements PlanningAgent {
     }
   }
 
-  public async executePlan(plan: Plan): Promise<void> {
+  public async executePlan(plan: Plan, cancellationToken?: any): Promise<void> {
     if (!this.initialized) {
       throw new Error('Planning agent not initialized')
     }
@@ -160,22 +160,28 @@ export class PlanningAgentImpl extends AbstractAgent implements PlanningAgent {
 
       // Execute each step
       for (const step of plan.steps) {
+        // Check for cancellation before each step
+        if (cancellationToken?.isCancelled) {
+          this.logger.log('Plan execution cancelled')
+          plan.status = 'cancelled'
+          return
+        }
+
         try {
           this.logger.withField('step', step).log('Executing step')
           await this.actionAgent.performAction(step)
         }
         catch (stepError) {
-          
           if (stepError instanceof ActionError) {
-             this.logger.withError(stepError).warn('Step execution failed with ActionError')
-             // If it's a resource failure or crafting failure that we've already tried to fix (implied by fail-fast skills),
-             // then we should abort and report failure instead of looping.
-             // We can check error types or context.
-             if (stepError.code === 'RESOURCE_MISSING' || stepError.code === 'CRAFTING_FAILED' || stepError.code === 'INVENTORY_FULL') {
-                 // For now, fail fast on these hard errors.
-                 // In the future we might want a "replanning" phase here, but NOT a blind retry.
-                 throw stepError;
-             }
+            this.logger.withError(stepError).warn('Step execution failed with ActionError')
+            // If it's a resource failure or crafting failure that we've already tried to fix (implied by fail-fast skills),
+            // then we should abort and report failure instead of looping.
+            // We can check error types or context.
+            if (stepError.code === 'RESOURCE_MISSING' || stepError.code === 'CRAFTING_FAILED' || stepError.code === 'INVENTORY_FULL') {
+              // For now, fail fast on these hard errors.
+              // In the future we might want a "replanning" phase here, but NOT a blind retry.
+              throw stepError
+            }
           }
 
           this.logger.withError(stepError).error('Failed to execute step')
@@ -189,7 +195,7 @@ export class PlanningAgentImpl extends AbstractAgent implements PlanningAgent {
               stepError instanceof Error ? stepError.message : 'Unknown error',
               'system',
             )
-            await this.executePlan(adjustedPlan)
+            await this.executePlan(adjustedPlan, cancellationToken)
             return
           }
 

services/minecraft/src/cognitive/conscious/orchestrator.ts

@@ -3,23 +3,30 @@ import type { Neuri, NeuriContext } from 'neuri'
 
 import type { EventManager } from '../perception/event-manager'
 import type { BotEvent, MineflayerWithAgents, UserIntentPayload } from '../types'
+import type { CancellationToken } from './task-state'
 
 import { withRetry } from '@moeru/std'
 import { system, user } from 'neuri/openai'
 
+import { DebugServer } from '../../debug-server'
 import { handleLLMCompletion } from './completion'
 import { generateStatusPrompt } from './prompt'
+import { TaskManager } from './task-manager'
 
 export class Orchestrator {
-  private isProcessing = false
+  private taskManager: TaskManager
+  private eventQueue: Array<BotEvent<UserIntentPayload>> = []
+  private isProcessingQueue = false
 
   constructor(
     private readonly deps: {
       eventManager: EventManager
       neuri: Neuri
       logger: Logg
     },
-  ) {}
+  ) {
+    this.taskManager = new TaskManager(deps.logger)
+  }
 
   public init(bot: MineflayerWithAgents): void {
     this.deps.eventManager.on<UserIntentPayload>('user_intent', async (event) => {
@@ -38,54 +45,49 @@ export class Orchestrator {
       return
     }
 
-    if (this.isProcessing) {
-      this.deps.logger.warn('Still processing previous intent, skipping or queuing (TBD)')
-      // For now, let's just abort or we could queue. Implementation plan said we'd decide.
-      // Let's implement a simple queue or at least a lock.
-      return
+    // Check if there's a current task
+    if (this.taskManager.hasCurrentTask()) {
+      // Determine if we should cancel current task or queue this event
+      if (this.shouldCancelCurrentTask(event)) {
+        this.deps.logger
+          .withFields({ currentTaskId: this.taskManager.getCurrentTask()?.id, priority: event.priority })
+          .log('Orchestrator: Cancelling current task for high-priority event')
+        this.taskManager.cancelCurrentTask('High-priority event received')
+        this.broadcastTaskStatus()
+      }
+      else {
+        // Queue the event for later processing
+        this.eventQueue.push(event)
+        this.deps.logger
+          .withFields({ queueSize: this.eventQueue.length, username, event: content })
+          .log('Orchestrator: Event queued')
+        this.broadcastTaskStatus()
+
+        // Notify user that we're busy
+        const busyMessage = 'I\'m busy right now, I\'ll get to that in a moment!'
+        if (source.reply) {
+          source.reply(busyMessage)
+        }
+        else {
+          bot.bot.chat(busyMessage)
+        }
+        return
+      }
     }
 
-    this.isProcessing = true
-    this.deps.logger.withFields({ username, content }).log('Orchestrator: Handling user intent')
+    // Create new task
+    const task = this.taskManager.createTask(content)
+    this.deps.logger
+      .withFields({ username, content, taskId: task.id })
+      .log('Orchestrator: Starting new task')
+    this.broadcastTaskStatus()
 
     try {
       // 1. Update memory
       bot.memory.chatHistory.push(user(`${username}: ${content}`))
 
-      // 2. Planning
-      const plan = await bot.planning.createPlan(content)
-      this.deps.logger.withFields({ plan }).log('Orchestrator: Plan created')
-
-      // 3. Execution
-      await bot.planning.executePlan(plan)
-      this.deps.logger.log('Orchestrator: Plan executed successfully')
-
-      // 4. Response Generation
-      const statusPrompt = await generateStatusPrompt(bot)
-      const response = await this.deps.neuri.handleStateless(
-        [...bot.memory.chatHistory, system(statusPrompt)],
-        async (c: NeuriContext) => {
-          this.deps.logger.log('Orchestrator: thinking...')
-          return withRetry<NeuriContext, string>(
-            ctx => handleLLMCompletion(ctx, bot, this.deps.logger),
-            {
-              retry: 3,
-              retryDelay: 1000,
-            },
-          )(c)
-        },
-      )
-
-      // 5. Reply
-      if (response) {
-        this.deps.logger.withFields({ response }).log('Orchestrator: Responded')
-        if (source.reply) {
-          source.reply(response)
-        }
-        else {
-          bot.bot.chat(response)
-        }
-      }
+      // 2. Execute task with cancellation support
+      await this.executeTaskWithCancellation(bot, event, task.cancellationToken)
     }
     catch (error) {
       this.deps.logger.withError(error).warn('Orchestrator: Failed to process intent')
@@ -98,7 +100,114 @@ export class Orchestrator {
       }
     }
     finally {
-      this.isProcessing = false
+      this.taskManager.completeCurrentTask()
+      this.broadcastTaskStatus()
+      // Process next queued event
+      this.processNextQueuedEvent(bot)
+    }
+  }
+
+  private async executeTaskWithCancellation(
+    bot: MineflayerWithAgents,
+    event: BotEvent<UserIntentPayload>,
+    cancellationToken: CancellationToken,
+  ): Promise<void> {
+    const { payload, source } = event
+    const { content } = payload
+
+    // Planning phase
+    if (cancellationToken.isCancelled)
+      return
+    this.taskManager.updateTaskStatus('planning')
+    this.broadcastTaskStatus()
+    this.deps.logger.log('Orchestrator: Starting planning phase')
+
+    const plan = await bot.planning.createPlan(content)
+    this.taskManager.setTaskPlan(plan)
+    this.deps.logger.withFields({ steps: plan.steps.length }).log('Orchestrator: Plan created')
+
+    // Execution phase
+    if (cancellationToken.isCancelled)
+      return
+    this.taskManager.updateTaskStatus('executing')
+    this.broadcastTaskStatus()
+    this.deps.logger.log('Orchestrator: Executing plan')
+
+    await bot.planning.executePlan(plan, cancellationToken)
+    this.deps.logger.log('Orchestrator: Plan executed successfully')
+
+    // Response generation phase
+    if (cancellationToken.isCancelled)
+      return
+    this.taskManager.updateTaskStatus('responding')
+    this.broadcastTaskStatus()
+    this.deps.logger.log('Orchestrator: Generating response')
+
+    const statusPrompt = await generateStatusPrompt(bot)
+    const taskContext = this.taskManager.getTaskContextForLLM()
+
+    const response = await this.deps.neuri.handleStateless(
+      [
+        ...bot.memory.chatHistory,
+        system(statusPrompt),
+        system(`Task Context: ${taskContext}`),
+      ],
+      async (c: NeuriContext) => {
+        this.deps.logger.log('Orchestrator: thinking...')
+        return withRetry<NeuriContext, string>(
+          ctx => handleLLMCompletion(ctx, bot, this.deps.logger),
+          {
+            retry: 3,
+            retryDelay: 1000,
+          },
+        )(c)
+      },
+    )
+
+    // Reply
+    if (cancellationToken.isCancelled)
+      return
+    if (response) {
+      this.deps.logger.withFields({ response }).log('Orchestrator: Responded')
+      if (source.reply) {
+        source.reply(response)
+      }
+      else {
+        bot.bot.chat(response)
+      }
+    }
+  }
+
+  private shouldCancelCurrentTask(event: BotEvent<UserIntentPayload>): boolean {
+    // High priority events should cancel current task
+    const HIGH_PRIORITY_THRESHOLD = 8
+    return (event.priority ?? 5) >= HIGH_PRIORITY_THRESHOLD
+  }
+
+  private async processNextQueuedEvent(bot: MineflayerWithAgents): Promise<void> {
+    // Prevent concurrent queue processing
+    if (this.isProcessingQueue || this.eventQueue.length === 0) {
+      return
+    }
+
+    this.isProcessingQueue = true
+    const nextEvent = this.eventQueue.shift()
+
+    if (nextEvent) {
+      this.deps.logger.log('Orchestrator: Processing next queued event')
+      await this.handleUserIntent(bot, nextEvent)
     }
+
+    this.isProcessingQueue = false
+  }
+
+  private broadcastTaskStatus(): void {
+    const debugServer = DebugServer.getInstance()
+    debugServer.broadcast('task-status', {
+      currentTask: this.taskManager.getCurrentTask(),
+      queueSize: this.eventQueue.length,
+      queue: this.eventQueue,
+      history: this.taskManager.getTaskHistory(),
+    })
   }
 }