feat(minecraft): async concious

update dashboard
2026-02-18 11:09:58 +08:00
parent cc56e6bfec
commit 9b698ae42d
6 changed files with 497 additions and 59 deletions
@@ -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
          }

@@ -3,15 +3,20 @@ 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: {
@@ -19,7 +24,9 @@ export class Orchestrator {
      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(),
+    })
+  }
 }
@@ -0,0 +1,172 @@
+import type { Logg } from '@guiiai/logg'
+
+import type { Plan } from '../../libs/mineflayer/base-agent'
+import type { TaskContext, TaskStatus } from './task-state'
+
+import { createCancellationToken } from './task-state'
+
+export class TaskManager {
+  private currentTask: TaskContext | null = null
+  private taskHistory: TaskContext[] = []
+  private readonly maxHistorySize = 10
+
+  constructor(private readonly logger: Logg) {}
+
+  /**
+   * Create a new task with cancellation support
+   */
+  public createTask(goal: string): TaskContext {
+    const task: TaskContext = {
+      id: this.generateTaskId(),
+      goal,
+      status: 'idle',
+      startTime: Date.now(),
+      cancellationToken: createCancellationToken(),
+    }
+
+    this.currentTask = task
+    this.logger.withFields({ taskId: task.id, goal }).log('TaskManager: Created new task')
+
+    return task
+  }
+
+  /**
+   * Update the status of the current task
+   */
+  public updateTaskStatus(status: TaskStatus, currentStep?: string): void {
+    if (!this.currentTask) {
+      this.logger.warn('TaskManager: No current task to update')
+      return
+    }
+
+    this.currentTask.status = status
+    if (currentStep) {
+      this.currentTask.currentStep = currentStep
+    }
+
+    this.logger.withFields({
+      taskId: this.currentTask.id,
+      status,
+      currentStep,
+    }).log('TaskManager: Updated task status')
+  }
+
+  /**
+   * Set the plan for the current task
+   */
+  public setTaskPlan(plan: Plan): void {
+    if (!this.currentTask) {
+      this.logger.warn('TaskManager: No current task to set plan for')
+      return
+    }
+
+    this.currentTask.plan = plan
+    this.logger.withFields({ taskId: this.currentTask.id }).log('TaskManager: Set task plan')
+  }
+
+  /**
+   * Cancel the current task
+   */
+  public cancelCurrentTask(reason?: string): void {
+    if (!this.currentTask) {
+      this.logger.warn('TaskManager: No current task to cancel')
+      return
+    }
+
+    this.logger.withFields({
+      taskId: this.currentTask.id,
+      reason,
+    }).log('TaskManager: Cancelling current task')
+
+    this.currentTask.status = 'cancelling'
+    this.currentTask.cancellationToken.cancel()
+
+    // Move to history
+    this.addToHistory(this.currentTask)
+    this.currentTask = null
+  }
+
+  /**
+   * Complete the current task
+   */
+  public completeCurrentTask(): void {
+    if (!this.currentTask) {
+      return
+    }
+
+    this.logger.withFields({ taskId: this.currentTask.id }).log('TaskManager: Task completed')
+
+    // Move to history
+    this.addToHistory(this.currentTask)
+    this.currentTask = null
+  }
+
+  /**
+   * Get the current task
+   */
+  public getCurrentTask(): TaskContext | null {
+    return this.currentTask
+  }
+
+  /**
+   * Check if there is a current task
+   */
+  public hasCurrentTask(): boolean {
+    return this.currentTask !== null
+  }
+
+  /**
+   * Check if can accept a new task
+   */
+  public canAcceptNewTask(): boolean {
+    return this.currentTask === null
+  }
+
+  /**
+   * Get formatted task context for LLM
+   */
+  public getTaskContextForLLM(): string {
+    if (!this.currentTask) {
+      return 'No active task'
+    }
+
+    const { goal, status, startTime, currentStep, plan } = this.currentTask
+    const elapsedSeconds = Math.floor((Date.now() - startTime) / 1000)
+
+    const lines = [
+      `Current Task: [${status.toUpperCase()}] ${goal}`,
+      `- Started: ${elapsedSeconds} seconds ago`,
+    ]
+
+    if (currentStep) {
+      lines.push(`- Current Step: ${currentStep}`)
+    }
+
+    if (plan) {
+      const totalSteps = plan.steps.length
+      lines.push(`- Plan: ${totalSteps} steps total`)
+    }
+
+    return lines.join('\n')
+  }
+
+  /**
+   * Get task history
+   */
+  public getTaskHistory(): TaskContext[] {
+    return [...this.taskHistory]
+  }
+
+  private generateTaskId(): string {
+    return `task_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`
+  }
+
+  private addToHistory(task: TaskContext): void {
+    this.taskHistory.push(task)
+
+    // Limit history size
+    if (this.taskHistory.length > this.maxHistorySize) {
+      this.taskHistory.shift()
+    }
+  }
+}
@@ -0,0 +1,37 @@
+import type { Plan } from '../../libs/mineflayer/base-agent'
+
+export type TaskStatus = 'idle' | 'planning' | 'executing' | 'responding' | 'cancelling'
+
+export interface CancellationToken {
+  isCancelled: boolean
+  cancel: () => void
+  onCancelled: (callback: () => void) => void
+}
+
+export interface TaskContext {
+  id: string
+  goal: string
+  status: TaskStatus
+  startTime: number
+  currentStep?: string
+  plan?: Plan
+  cancellationToken: CancellationToken
+}
+
+export function createCancellationToken(): CancellationToken {
+  let isCancelled = false
+  const callbacks: Array<() => void> = []
+
+  return {
+    get isCancelled() {
+      return isCancelled
+    },
+    cancel() {
+      isCancelled = true
+      callbacks.forEach(cb => cb())
+    },
+    onCancelled(callback: () => void) {
+      callbacks.push(callback)
+    },
+  }
+}
@@ -38,14 +38,14 @@ export interface MemoryAgent extends BaseAgent {
 export interface Plan {
  goal: string
  steps: PlanStep[]
-  status: 'pending' | 'in_progress' | 'completed' | 'failed'
+  status: 'pending' | 'in_progress' | 'completed' | 'failed' | 'cancelled'
  requiresAction: boolean
 }

 export interface PlanningAgent extends BaseAgent {
  type: 'planning'
  createPlan: (goal: string) => Promise<Plan>
-  executePlan: (plan: Plan) => Promise<void>
+  executePlan: (plan: Plan, cancellationToken?: any) => Promise<void>
  adjustPlan: (plan: Plan, feedback: string, sender: string) => Promise<Plan>
 }

@@ -235,10 +235,34 @@
    </header>
    <main>
        <nav>
-            <div class="nav-item active" onclick="switchTab('logs')">Logs</div>
+            <div class="nav-item active" onclick="switchTab('task-status')">Task Status</div>
+            <div class="nav-item" onclick="switchTab('logs')">Logs</div>
            <div class="nav-item" onclick="switchTab('llm')">LLM Traces</div>
        </nav>
        
+        <div id="task-status-panel" class="content-panel active">
+            <div class="toolbar">
+                <h3 style="margin: 0;">Current Task State</h3>
+            </div>
+            <div class="log-container" style="padding: 2rem;">
+                <div id="current-task" style="margin-bottom: 2rem;">
+                    <div class="section-title">Active Task</div>
+                    <div id="task-info" class="llm-card" style="background-color: #1e1e1e;">
+                        <div style="padding: 1rem; color: #666;">No active task</div>
+                    </div>
+                </div>
+
+                <div id="task-queue" style="margin-bottom: 2rem;">
+                    <div class="section-title">Event Queue (<span id="queue-count">0</span>)</div>
+                    <div id="queue-list" style="display: flex; flex-direction: column; gap: 0.5rem;"></div>
+                </div>
+
+                <div id="task-history">
+                    <div class="section-title">Recent Tasks (Last 5)</div>
+                    <div id="history-list" style="display: flex; flex-direction: column; gap: 0.5rem;"></div>
+                </div>
+            </div>
+        </div>
        <div id="logs-panel" class="content-panel active">
            <div class="toolbar">
                <button onclick="clearLogs()">Clear</button>
@@ -410,6 +434,96 @@
                    console.error('Failed to parse llm trace', e);
                }
            });
+
+            eventSource.addEventListener('task-status', (e) => {
+                try {
+                    const data = JSON.parse(e.data);
+                    updateTaskStatus(data);
+                } catch (e) {
+                    console.error('Failed to parse task status', e);
+                }
+            });
+        }
+
+        function updateTaskStatus(status) {
+            // Update current task
+            const taskInfo = document.getElementById('task-info');
+            if (status.currentTask) {
+                const task = status.currentTask;
+                const elapsed = Math.floor((Date.now() - task.startTime) / 1000);
+                const statusColor = {
+                    'idle': '#888',
+                    'planning': '#ffb74d',
+                    'executing': '#4a9eff',
+                    'responding': '#9c27b0',
+                    'cancelling': '#ef5350'
+                }[task.status] || '#888';
+
+                taskInfo.innerHTML = `
+                    <div style="padding: 1rem;">
+                        <div style="display: flex; justify-content: space-between; margin-bottom: 0.5rem;">
+                            <span style="font-weight: bold; font-size: 1.1em;">${task.goal}</span>
+                            <span class="badge" style="background-color: ${statusColor}">${task.status.toUpperCase()}</span>
+                        </div>
+                        <div style="color: #888; font-size: 0.9em;">
+                            <div>Task ID: ${task.id}</div>
+                            <div>Elapsed: ${elapsed}s</div>
+                            ${task.currentStep ? `<div>Current Step: ${task.currentStep}</div>` : ''}
+                            ${task.plan ? `<div>Plan: ${task.plan.steps.length} steps (${task.plan.status})</div>` : ''}
+                        </div>
+                    </div>
+                `;
+            } else {
+                taskInfo.innerHTML = '<div style="padding: 1rem; color: #666;">No active task</div>';
+            }
+
+            // Update queue
+            const queueCount = document.getElementById('queue-count');
+            const queueList = document.getElementById('queue-list');
+            queueCount.textContent = status.queueSize || 0;
+            
+            if (status.queue && status.queue.length > 0) {
+                queueList.innerHTML = status.queue.map((event, idx) => `
+                    <div class="message" style="background-color: #1e1e1e; border-left-color: #ffb74d;">
+                        <div style="display: flex; justify-content: space-between;">
+                            <span>#${idx + 1}: ${event.payload?.content || 'Unknown event'}</span>
+                            <span class="badge">Priority: ${event.priority || 5}</span>
+                        </div>
+                        <div style="color: #666; font-size: 0.85em; margin-top: 0.5rem;">
+                            From: ${event.source?.id || 'unknown'}
+                        </div>
+                    </div>
+                `).join('');
+            } else {
+                queueList.innerHTML = '<div style="color: #666; padding: 1rem;">Queue is empty</div>';
+            }
+
+            // Update history
+            const historyList = document.getElementById('history-list');
+            if (status.history && status.history.length > 0) {
+                historyList.innerHTML = status.history.slice(0, 5).map(task => {
+                    const elapsed = Math.floor((Date.now() - task.startTime) / 1000);
+                    const statusColor = {
+                        'completed': '#4caf50',
+                        'failed': '#ef5350',
+                        'cancelled': '#ff9800'
+                    }[task.plan?.status] || '#888';
+
+                    return `
+                        <div class="message" style="background-color: #1e1e1e; border-left-color: ${statusColor};">
+                            <div style="display: flex; justify-content: space-between;">
+                                <span>${task.goal}</span>
+                                <span class="badge" style="background-color: ${statusColor}">${task.plan?.status || task.status}</span>
+                            </div>
+                            <div style="color: #666; font-size: 0.85em; margin-top: 0.5rem;">
+                                Duration: ${elapsed}s
+                            </div>
+                        </div>
+                    `;
+                }).join('');
+            } else {
+                historyList.innerHTML = '<div style="color: #666; padding: 1rem;">No task history</div>';
+            }
        }

        connectSSE();