BitSquare/Codeman
2
0
Fork 0
mirror of https://github.com/Ark0N/Codeman.git synced 2026-10-07 16:09:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(execution): add ExecutionBridge for parallel task execution

Browse Source 
Implements the Execution Optimizer Rework plan that enables the system
to actually use the optimizer metadata (parallelGroups, agentType,
recommendedModel, requiresFreshContext, estimatedTokens) that was
previously generated but ignored.

New components:
- ExecutionBridge: Central coordinator that loads optimized plans,
  manages parallel execution within groups, and coordinates with
  SpawnOrchestrator for session-based execution
- ModelSelector: Routes tasks to appropriate models (opus/sonnet/haiku)
  based on user defaults and agent type overrides. Optimizer
  recommendations are advisory only - user preferences always win
- GroupScheduler: Builds topologically ordered execution groups,
  manages dependencies, determines execution mode (session vs task-tool)
- ContextManager: Handles fresh context requirements via /clear+/init
  or new session spawning

Features:
- Parallel task execution within groups (configurable limit)
- Group-level dependency tracking (lower groups complete first)
- Partial failure handling (continue with non-dependent tasks)
- Model configuration in App Settings > Models tab
- Agent type overrides (explore, implement, test, review)
- Execution control API endpoints (start, pause, resume, cancel)
- SSE events for real-time execution progress visibility
- Execution history tracking

API endpoints:
- GET/POST/PUT /api/execution/* for execution control
- GET/PUT /api/execution/model-config for model settings

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
arkon
2026-01-28 07:07:18 +01:00
co-authored by Claude Opus 4.5
parent 9eccd061b9
commit 55e9f9f0a5
12 changed files with 3140 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/web/server.ts
+419 -5
View File
@@ -36,6 +36,13 @@ import { v4 as uuidv4 } from 'uuid';
import { createRequire } from 'node:module';
import { RunSummaryTracker } from '../run-summary.js';
import { PlanOrchestrator, type DetailedPlanResult } from '../plan-orchestrator.js';
import {
ExecutionBridge,
getExecutionBridge,
type PlanItem as ExecutionPlanItem,
type AgentSpawner,
} from '../execution-bridge.js';
import { type ModelConfig } from '../model-selector.js';
// Load version from package.json
const require = createRequire(import.meta.url);
@@ -311,6 +318,8 @@ export class WebServer extends EventEmitter {
private pendingRespawnStarts: Map<string, NodeJS.Timeout> = new Map();
// Active plan orchestrators (for cancellation via API)
private activePlanOrchestrators: Map<string, PlanOrchestrator> = new Map();
// Execution bridge for parallel task execution
private executionBridge: ExecutionBridge;
// Grace period before starting restored respawn controllers (2 minutes)
private static readonly RESPAWN_RESTORE_GRACE_PERIOD_MS = 2 * 60 * 1000;
@@ -345,6 +354,10 @@ export class WebServer extends EventEmitter {
this.spawnOrchestrator = new SpawnOrchestrator();
this.setupSpawnOrchestratorListeners();
// Initialize execution bridge with model config from settings
this.executionBridge = getExecutionBridge(this.loadModelConfig());
this.setupExecutionBridgeListeners();
// Set up subagent watcher listeners
this.setupSubagentWatcherListeners();
@@ -1801,6 +1814,38 @@ export class WebServer extends EventEmitter {
};
});
// Toggle image watcher for a session
this.app.post('/api/sessions/:id/image-watcher', async (req) => {
const { id } = req.params as { id: string };
const body = req.body as { enabled: boolean };
const session = this.sessions.get(id);
if (!session) {
return createErrorResponse(ApiErrorCode.NOT_FOUND, 'Session not found');
}
if (body.enabled === undefined) {
return createErrorResponse(ApiErrorCode.INVALID_INPUT, 'enabled field is required');
}
if (body.enabled) {
imageWatcher.watchSession(session.id, session.workingDir);
} else {
imageWatcher.unwatchSession(session.id);
}
// Store state on session for persistence
session.imageWatcherEnabled = body.enabled;
this.persistSessionState(session);
return {
success: true,
data: {
imageWatcherEnabled: body.enabled,
},
};
});
// Quick run (create session, run prompt, return result, then cleanup)
this.app.post('/api/run', async (req) => {
// Prevent unbounded session creation
@@ -2837,6 +2882,22 @@ NOW: Generate the implementation plan for the task above. Think step by step.`;
console.log('Subagent watcher stopped via settings change');
}
// Handle image watcher toggle dynamically
const imageWatcherEnabled = settings.imageWatcherEnabled ?? true;
if (imageWatcherEnabled && !imageWatcher.isRunning()) {
imageWatcher.start();
// Re-watch all active sessions that have image watcher enabled
for (const session of this.sessions.values()) {
if (session.imageWatcherEnabled) {
imageWatcher.watchSession(session.id, session.workingDir);
}
}
console.log('Image watcher started via settings change');
} else if (!imageWatcherEnabled && imageWatcher.isRunning()) {
imageWatcher.stop();
console.log('Image watcher stopped via settings change');
}
return { success: true };
} catch (err) {
return createErrorResponse(ApiErrorCode.OPERATION_FAILED, getErrorMessage(err));
@@ -3045,6 +3106,148 @@ NOW: Generate the implementation plan for the task above. Think step by step.`;
return { success: true, data: { agentId } };
});
// ========== Execution Bridge Endpoints ==========
// Get execution status
this.app.get('/api/execution/status', async () => {
return {
success: true,
data: {
status: this.executionBridge.status,
progress: this.executionBridge.getProgress(),
},
};
});
// Get execution schedule (the current plan)
this.app.get('/api/execution/schedule', async () => {
const schedule = this.executionBridge['_scheduler'].schedule;
return { success: true, data: schedule };
});
// Load a plan for execution
// Accepts either ExecutionPlanItem format (id, title, description) or
// PlanOrchestrator PlanItem format (id, content) and converts as needed
this.app.post('/api/execution/load', async (req) => {
const { items, workingDir } = req.body as {
items: Array<{
id?: string;
title?: string;
description?: string;
content?: string;
parallelGroup?: number;
agentType?: string;
recommendedModel?: string;
requiresFreshContext?: boolean;
estimatedTokens?: number;
inputFiles?: string[];
outputFiles?: string[];
dependencies?: string[];
}>;
workingDir?: string;
};
if (!items || !Array.isArray(items)) {
return createErrorResponse(ApiErrorCode.INVALID_INPUT, 'items array is required');
}
try {
if (workingDir) {
this.executionBridge.setWorkingDir(workingDir);
}
// Convert to ExecutionPlanItem format
const execItems: ExecutionPlanItem[] = items.map((item, idx) => ({
id: item.id || `task-${idx}`,
title: item.title || item.content?.slice(0, 50) || `Task ${idx + 1}`,
description: item.description || item.content || '',
parallelGroup: item.parallelGroup,
agentType: item.agentType,
recommendedModel: item.recommendedModel,
requiresFreshContext: item.requiresFreshContext,
estimatedTokens: item.estimatedTokens,
inputFiles: item.inputFiles,
outputFiles: item.outputFiles,
dependencies: item.dependencies,
}));
const schedule = this.executionBridge.loadPlan(execItems);
return { success: true, data: schedule };
} catch (err) {
return createErrorResponse(ApiErrorCode.OPERATION_FAILED, getErrorMessage(err));
}
});
// Start execution
this.app.post('/api/execution/start', async () => {
try {
await this.executionBridge.start();
return { success: true };
} catch (err) {
return createErrorResponse(ApiErrorCode.OPERATION_FAILED, getErrorMessage(err));
}
});
// Pause execution
this.app.post('/api/execution/pause', async () => {
this.executionBridge.pause();
return { success: true };
});
// Resume execution
this.app.post('/api/execution/resume', async () => {
this.executionBridge.resume();
return { success: true };
});
// Cancel execution
this.app.post('/api/execution/cancel', async (req) => {
const { reason } = (req.body as { reason?: string }) || {};
await this.executionBridge.cancel(reason || 'Cancelled via API');
return { success: true };
});
// Reset execution bridge
this.app.post('/api/execution/reset', async () => {
this.executionBridge.reset();
return { success: true };
});
// Get execution history
this.app.get('/api/execution/history', async () => {
return { success: true, data: this.executionBridge.getHistory() };
});
// Get model configuration
this.app.get('/api/execution/model-config', async () => {
return { success: true, data: this.executionBridge.getModelConfig() };
});
// Update model configuration
this.app.put('/api/execution/model-config', async (req) => {
const config = req.body as Partial<ModelConfig>;
try {
this.executionBridge.updateModelConfig(config);
const fullConfig = this.executionBridge.getModelConfig();
this.saveModelConfig(fullConfig);
this.broadcast('execution:modelConfigUpdated', fullConfig);
return { success: true, data: fullConfig };
} catch (err) {
return createErrorResponse(ApiErrorCode.OPERATION_FAILED, getErrorMessage(err));
}
});
// Mark task complete (external signal)
this.app.post('/api/execution/tasks/:taskId/complete', async (req) => {
const { taskId } = req.params as { taskId: string };
this.executionBridge.markTaskComplete(taskId);
return { success: true };
});
// Mark task failed (external signal)
this.app.post('/api/execution/tasks/:taskId/failed', async (req) => {
const { taskId } = req.params as { taskId: string };
const { error } = (req.body as { error?: string }) || {};
this.executionBridge.markTaskFailed(taskId, error || 'Failed via API');
return { success: true };
});
// ========== Subagent Monitoring (Claude Code Background Agents) ==========
// List all known subagents
@@ -3457,8 +3660,10 @@ NOW: Generate the implementation plan for the task above. Think step by step.`;
// Set working directory for Ralph tracker to auto-load @fix_plan.md
session.ralphTracker.setWorkingDir(session.workingDir);
// Start watching for new images in this session's working directory
imageWatcher.watchSession(session.id, session.workingDir);
// Start watching for new images in this session's working directory (if enabled globally and per-session)
if (this.isImageWatcherEnabled() && session.imageWatcherEnabled) {
imageWatcher.watchSession(session.id, session.workingDir);
}
session.on('output', (data) => {
// Use batching for better performance at high throughput
@@ -3843,6 +4048,193 @@ NOW: Generate the implementation plan for the task above. Think step by step.`;
this.spawnOrchestrator.on('stateUpdate', (data) => this.broadcast('spawn:stateUpdate', data));
}
/**
* Load model configuration from settings file.
*/
private loadModelConfig(): Partial<ModelConfig> {
const settingsPath = join(homedir(), '.claudeman', 'settings.json');
try {
if (existsSync(settingsPath)) {
const content = readFileSync(settingsPath, 'utf-8');
const settings = JSON.parse(content);
return settings.modelConfig || {};
}
} catch (err) {
console.error('Failed to load model config:', getErrorMessage(err));
}
return {};
}
/**
* Save model configuration to settings file.
*/
private saveModelConfig(config: ModelConfig): void {
const settingsPath = join(homedir(), '.claudeman', 'settings.json');
try {
const dir = dirname(settingsPath);
if (!existsSync(dir)) {
mkdirSync(dir, { recursive: true });
}
let settings: Record<string, unknown> = {};
if (existsSync(settingsPath)) {
settings = JSON.parse(readFileSync(settingsPath, 'utf-8'));
}
settings.modelConfig = config;
writeFileSync(settingsPath, JSON.stringify(settings, null, 2));
} catch (err) {
console.error('Failed to save model config:', getErrorMessage(err));
}
}
/**
* Set up event listeners for execution bridge.
* Broadcasts execution progress to SSE clients.
*/
private setupExecutionBridgeListeners(): void {
// Create agent spawner interface
const spawner: AgentSpawner = {
spawnAgentWithModel: async (taskId, workingDir, prompt, _model, _options) => {
// Note: model and options are reserved for future model selection integration
const globalNice = this.getGlobalNiceConfig();
const session = new Session({
workingDir,
screenManager: this.screenManager,
useScreen: true,
mode: 'claude',
name: `exec:${taskId}`,
niceConfig: globalNice,
});
this.sessions.set(session.id, session);
this.store.incrementSessionsCreated();
this.setupSessionListeners(session);
await session.startInteractive();
this.broadcast('session:created', session.toDetailedState());
this.broadcast('session:interactive', { id: session.id });
this.persistSessionState(session);
// Configure ralph tracker for completion detection
session.ralphTracker.enable();
// Set up completion listener for execution bridge
session.on('ralphCompletionDetected', () => {
this.executionBridge.markTaskComplete(taskId);
});
// Send the prompt
setTimeout(() => {
session.writeViaScreen(prompt + '\r');
}, 2000);
return { sessionId: session.id };
},
useTaskTool: async (_sessionId, _taskId, _prompt, _model) => {
// Task tool mode - not yet implemented, falls back to session mode
throw new Error('Task tool mode not yet implemented');
},
isTaskComplete: (taskId) => {
// Check if task is marked complete in scheduler
const schedule = this.executionBridge['_scheduler'].schedule;
if (!schedule) return false;
for (const group of schedule.groups) {
const task = group.tasks.find(t => t.id === taskId);
if (task) return task.status === 'completed';
}
return false;
},
getTaskResult: (taskId) => {
const schedule = this.executionBridge['_scheduler'].schedule;
if (!schedule) return null;
for (const group of schedule.groups) {
const task = group.tasks.find(t => t.id === taskId);
if (task) {
if (task.status === 'completed') {
return { success: true };
} else if (task.status === 'failed') {
return { success: false, error: task.error };
}
}
}
return null;
},
};
this.executionBridge.setAgentSpawner(spawner);
// Set up session writer for context management
this.executionBridge.setSessionWriter({
writeToSession: (sessionId, text) => {
const session = this.sessions.get(sessionId);
if (session) {
session.writeViaScreen(text);
}
},
createSession: async (workingDir, name) => {
const globalNice = this.getGlobalNiceConfig();
const session = new Session({
workingDir,
screenManager: this.screenManager,
useScreen: true,
mode: 'claude',
name: name || 'exec-context',
niceConfig: globalNice,
});
this.sessions.set(session.id, session);
this.store.incrementSessionsCreated();
this.setupSessionListeners(session);
await session.startInteractive();
this.broadcast('session:created', session.toDetailedState());
this.persistSessionState(session);
return { sessionId: session.id };
},
});
// Forward execution bridge events as SSE broadcasts
this.executionBridge.on('planLoaded', (schedule) => {
this.broadcast('execution:planLoaded', { schedule });
});
this.executionBridge.on('started', () => {
this.broadcast('execution:started', {});
});
this.executionBridge.on('paused', () => {
this.broadcast('execution:paused', {});
});
this.executionBridge.on('resumed', () => {
this.broadcast('execution:resumed', {});
});
this.executionBridge.on('completed', (data) => {
this.broadcast('execution:completed', data);
});
this.executionBridge.on('cancelled', (reason) => {
this.broadcast('execution:cancelled', { reason });
});
this.executionBridge.on('groupStarted', (data) => {
this.broadcast('execution:groupStarted', data);
});
this.executionBridge.on('groupCompleted', (data) => {
this.broadcast('execution:groupCompleted', data);
});
this.executionBridge.on('taskAssigned', (data) => {
this.broadcast('execution:taskAssigned', data);
});
this.executionBridge.on('taskCompleted', (data) => {
this.broadcast('execution:taskCompleted', data);
});
this.executionBridge.on('taskFailed', (data) => {
this.broadcast('execution:taskFailed', data);
});
this.executionBridge.on('freshContext', (data) => {
this.broadcast('execution:freshContext', data);
});
this.executionBridge.on('modelSelected', (data) => {
this.broadcast('execution:modelSelected', data);
});
this.executionBridge.on('progress', (progress) => {
this.broadcast('execution:progress', progress);
});
}
private setupTimedRespawn(sessionId: string, durationMinutes: number): void {
// Clear existing timer if any
const existing = this.respawnTimers.get(sessionId);
@@ -4489,9 +4881,13 @@ NOW: Generate the implementation plan for the task above. Think step by step.`;
console.log('Subagent watcher disabled by user settings');
}
// Start image watcher for auto-popup of screenshots
imageWatcher.start();
console.log('Image watcher started - monitoring session directories for new images');
// Start image watcher for auto-popup of screenshots (if enabled)
if (this.isImageWatcherEnabled()) {
imageWatcher.start();
console.log('Image watcher started - monitoring session directories for new images');
} else {
console.log('Image watcher disabled by user settings');
}
}
/**
@@ -4512,6 +4908,24 @@ NOW: Generate the implementation plan for the task above. Think step by step.`;
return true; // Default enabled
}
/**
* Check if image watcher is enabled in settings (default: true)
*/
private isImageWatcherEnabled(): boolean {
const settingsPath = join(homedir(), '.claudeman', 'settings.json');
try {
if (existsSync(settingsPath)) {
const content = readFileSync(settingsPath, 'utf-8');
const settings = JSON.parse(content);
// Default to true if not explicitly set
return settings.imageWatcherEnabled ?? true;
}
} catch (err) {
console.error('Failed to read image watcher setting:', err);
}
return true; // Default enabled
}
private async restoreScreenSessions(): Promise<void> {
try {
// Reconcile screens to find which ones are still alive (also discovers unknown screens)