From 9b698ae42d3b1677e94be7e7b4544e48c3a39ae6 Mon Sep 17 00:00:00 2001 From: Rin Date: Mon, 29 Dec 2025 01:18:09 +0800 Subject: [PATCH] feat(minecraft): async concious update dashboard --- .../minecraft/src/agents/planning/index.ts | 30 +-- .../src/cognitive/conscious/orchestrator.ts | 197 ++++++++++++++---- .../src/cognitive/conscious/task-manager.ts | 172 +++++++++++++++ .../src/cognitive/conscious/task-state.ts | 37 ++++ .../src/libs/mineflayer/base-agent.ts | 4 +- services/minecraft/src/web/dashboard.html | 116 ++++++++++- 6 files changed, 497 insertions(+), 59 deletions(-) create mode 100644 services/minecraft/src/cognitive/conscious/task-manager.ts create mode 100644 services/minecraft/src/cognitive/conscious/task-state.ts diff --git a/services/minecraft/src/agents/planning/index.ts b/services/minecraft/src/agents/planning/index.ts index 950801ae7..5cba9e878 100644 --- a/services/minecraft/src/agents/planning/index.ts +++ b/services/minecraft/src/agents/planning/index.ts @@ -138,7 +138,7 @@ export class PlanningAgentImpl extends AbstractAgent implements PlanningAgent { } } - public async executePlan(plan: Plan): Promise { + public async executePlan(plan: Plan, cancellationToken?: any): Promise { 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 } diff --git a/services/minecraft/src/cognitive/conscious/orchestrator.ts b/services/minecraft/src/cognitive/conscious/orchestrator.ts index 045b32354..6e308c0f9 100644 --- a/services/minecraft/src/cognitive/conscious/orchestrator.ts +++ b/services/minecraft/src/cognitive/conscious/orchestrator.ts @@ -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> = [] + 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('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( - 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, + cancellationToken: CancellationToken, + ): Promise { + 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( + 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): 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 { + // 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(), + }) + } } diff --git a/services/minecraft/src/cognitive/conscious/task-manager.ts b/services/minecraft/src/cognitive/conscious/task-manager.ts new file mode 100644 index 000000000..a69af1919 --- /dev/null +++ b/services/minecraft/src/cognitive/conscious/task-manager.ts @@ -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() + } + } +} diff --git a/services/minecraft/src/cognitive/conscious/task-state.ts b/services/minecraft/src/cognitive/conscious/task-state.ts new file mode 100644 index 000000000..be244a595 --- /dev/null +++ b/services/minecraft/src/cognitive/conscious/task-state.ts @@ -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) + }, + } +} diff --git a/services/minecraft/src/libs/mineflayer/base-agent.ts b/services/minecraft/src/libs/mineflayer/base-agent.ts index 24bf9fccc..b3023d44e 100644 --- a/services/minecraft/src/libs/mineflayer/base-agent.ts +++ b/services/minecraft/src/libs/mineflayer/base-agent.ts @@ -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 - executePlan: (plan: Plan) => Promise + executePlan: (plan: Plan, cancellationToken?: any) => Promise adjustPlan: (plan: Plan, feedback: string, sender: string) => Promise } diff --git a/services/minecraft/src/web/dashboard.html b/services/minecraft/src/web/dashboard.html index 509e245d0..9e7864bf7 100644 --- a/services/minecraft/src/web/dashboard.html +++ b/services/minecraft/src/web/dashboard.html @@ -235,10 +235,34 @@
+
+
+

Current Task State

+
+
+
+
Active Task
+
+
No active task
+
+
+ +
+
Event Queue (0)
+
+
+ +
+
Recent Tasks (Last 5)
+
+
+
+
@@ -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 = ` +
+
+ ${task.goal} + ${task.status.toUpperCase()} +
+
+
Task ID: ${task.id}
+
Elapsed: ${elapsed}s
+ ${task.currentStep ? `
Current Step: ${task.currentStep}
` : ''} + ${task.plan ? `
Plan: ${task.plan.steps.length} steps (${task.plan.status})
` : ''} +
+
+ `; + } else { + taskInfo.innerHTML = '
No active task
'; + } + + // 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) => ` +
+
+ #${idx + 1}: ${event.payload?.content || 'Unknown event'} + Priority: ${event.priority || 5} +
+
+ From: ${event.source?.id || 'unknown'} +
+
+ `).join(''); + } else { + queueList.innerHTML = '
Queue is empty
'; + } + + // 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 ` +
+
+ ${task.goal} + ${task.plan?.status || task.status} +
+
+ Duration: ${elapsed}s +
+
+ `; + }).join(''); + } else { + historyList.innerHTML = '
No task history
'; + } } connectSSE();