mirror of
https://github.com/catlog22/Claude-Code-Workflow.git
synced 2026-02-28 09:23:08 +08:00
fix(orchestrator): resolve high-priority issues from code review
1. Race condition fix: Removed frontend direct lockSession call in useOrchestratorExecution.ts - session locking now handled purely via backend WebSocket broadcast (CLI_SESSION_LOCKED) 2. WebSocket handlers: Added handleSessionLockedMessage and handleSessionUnlockedMessage to sessionManagerStore.ts 3. useWebSocket integration: Added case handlers for CLI_SESSION_LOCKED and CLI_SESSION_UNLOCKED messages 4. API input validation: Added validation for sessionConfig, stepTimeout, and errorStrategy in execute-in-session endpoint 5. Fixed wsBroadcast reference: Changed to broadcastToClients from context
This commit is contained in:
@@ -9,6 +9,10 @@ import { useExecutionStore } from '@/stores/executionStore';
|
|||||||
import { useFlowStore } from '@/stores';
|
import { useFlowStore } from '@/stores';
|
||||||
import { useCliStreamStore } from '@/stores/cliStreamStore';
|
import { useCliStreamStore } from '@/stores/cliStreamStore';
|
||||||
import { useCliSessionStore } from '@/stores/cliSessionStore';
|
import { useCliSessionStore } from '@/stores/cliSessionStore';
|
||||||
|
import {
|
||||||
|
handleSessionLockedMessage,
|
||||||
|
handleSessionUnlockedMessage,
|
||||||
|
} from '@/stores/sessionManagerStore';
|
||||||
import {
|
import {
|
||||||
OrchestratorMessageSchema,
|
OrchestratorMessageSchema,
|
||||||
type OrchestratorWebSocketMessage,
|
type OrchestratorWebSocketMessage,
|
||||||
@@ -212,6 +216,30 @@ export function useWebSocket(options: UseWebSocketOptions = {}): UseWebSocketRet
|
|||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
case 'CLI_SESSION_LOCKED': {
|
||||||
|
const { sessionKey, reason, executionId, timestamp } = data.payload ?? {};
|
||||||
|
if (typeof sessionKey === 'string') {
|
||||||
|
handleSessionLockedMessage({
|
||||||
|
sessionKey,
|
||||||
|
reason: reason ?? 'Workflow execution',
|
||||||
|
executionId,
|
||||||
|
timestamp: timestamp ?? new Date().toISOString(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
case 'CLI_SESSION_UNLOCKED': {
|
||||||
|
const { sessionKey, timestamp } = data.payload ?? {};
|
||||||
|
if (typeof sessionKey === 'string') {
|
||||||
|
handleSessionUnlockedMessage({
|
||||||
|
sessionKey,
|
||||||
|
timestamp: timestamp ?? new Date().toISOString(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
case 'CLI_OUTPUT': {
|
case 'CLI_OUTPUT': {
|
||||||
const { executionId, chunkType, data: outputData, unit } = data.payload;
|
const { executionId, chunkType, data: outputData, unit } = data.payload;
|
||||||
|
|
||||||
|
|||||||
@@ -35,6 +35,50 @@ const initialState: SessionManagerState = {
|
|||||||
/** Module-level worker reference. Worker objects are not serializable. */
|
/** Module-level worker reference. Worker objects are not serializable. */
|
||||||
let _workerRef: Worker | null = null;
|
let _workerRef: Worker | null = null;
|
||||||
|
|
||||||
|
// ========== WebSocket Session Lock Message Handler ==========
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Handle CLI_SESSION_LOCKED WebSocket message from backend.
|
||||||
|
* Updates session metadata to reflect locked state.
|
||||||
|
*/
|
||||||
|
export function handleSessionLockedMessage(payload: {
|
||||||
|
sessionKey: string;
|
||||||
|
reason: string;
|
||||||
|
executionId?: string;
|
||||||
|
timestamp: string;
|
||||||
|
}): void {
|
||||||
|
const store = useSessionManagerStore.getState();
|
||||||
|
store.updateTerminalMeta(payload.sessionKey, {
|
||||||
|
status: 'locked',
|
||||||
|
isLocked: true,
|
||||||
|
lockReason: payload.reason,
|
||||||
|
lockedByExecutionId: payload.executionId,
|
||||||
|
lockedAt: payload.timestamp,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Handle CLI_SESSION_UNLOCKED WebSocket message from backend.
|
||||||
|
* Updates session metadata to reflect unlocked state.
|
||||||
|
*/
|
||||||
|
export function handleSessionUnlockedMessage(payload: {
|
||||||
|
sessionKey: string;
|
||||||
|
timestamp: string;
|
||||||
|
}): void {
|
||||||
|
const store = useSessionManagerStore.getState();
|
||||||
|
const existing = store.terminalMetas[payload.sessionKey];
|
||||||
|
// Only unlock if currently locked
|
||||||
|
if (existing?.isLocked) {
|
||||||
|
store.updateTerminalMeta(payload.sessionKey, {
|
||||||
|
status: 'active',
|
||||||
|
isLocked: false,
|
||||||
|
lockReason: undefined,
|
||||||
|
lockedByExecutionId: undefined,
|
||||||
|
lockedAt: undefined,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ========== Worker Message Handler ==========
|
// ========== Worker Message Handler ==========
|
||||||
|
|
||||||
function _handleWorkerMessage(event: MessageEvent<MonitorAlert>): void {
|
function _handleWorkerMessage(event: MessageEvent<MonitorAlert>): void {
|
||||||
|
|||||||
@@ -1307,6 +1307,37 @@ export async function handleOrchestratorRoutes(ctx: RouteContext): Promise<boole
|
|||||||
errorStrategy?: 'pause' | 'skip' | 'stop';
|
errorStrategy?: 'pause' | 'skip' | 'stop';
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Input validation
|
||||||
|
const validTools = ['claude', 'gemini', 'qwen', 'codex', 'opencode'];
|
||||||
|
const validShells = ['bash', 'pwsh', 'cmd'];
|
||||||
|
const validErrorStrategies = ['pause', 'skip', 'stop'];
|
||||||
|
|
||||||
|
if (sessionConfig) {
|
||||||
|
if (sessionConfig.tool && !validTools.includes(sessionConfig.tool)) {
|
||||||
|
return { success: false, error: `Invalid tool. Must be one of: ${validTools.join(', ')}`, status: 400 };
|
||||||
|
}
|
||||||
|
if (sessionConfig.preferredShell && !validShells.includes(sessionConfig.preferredShell)) {
|
||||||
|
return { success: false, error: `Invalid preferredShell. Must be one of: ${validShells.join(', ')}`, status: 400 };
|
||||||
|
}
|
||||||
|
if (sessionConfig.model && typeof sessionConfig.model !== 'string') {
|
||||||
|
return { success: false, error: 'model must be a string', status: 400 };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (inputVariables && typeof inputVariables !== 'object') {
|
||||||
|
return { success: false, error: 'variables must be an object', status: 400 };
|
||||||
|
}
|
||||||
|
|
||||||
|
if (stepTimeout !== undefined) {
|
||||||
|
if (typeof stepTimeout !== 'number' || stepTimeout < 1000 || stepTimeout > 3600000) {
|
||||||
|
return { success: false, error: 'stepTimeout must be a number between 1000 and 3600000 (ms)', status: 400 };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!validErrorStrategies.includes(errorStrategy)) {
|
||||||
|
return { success: false, error: `Invalid errorStrategy. Must be one of: ${validErrorStrategies.join(', ')}`, status: 400 };
|
||||||
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
// Verify flow exists
|
// Verify flow exists
|
||||||
const flow = await readFlowStorage(workflowDir, flowId);
|
const flow = await readFlowStorage(workflowDir, flowId);
|
||||||
@@ -1355,31 +1386,27 @@ export async function handleOrchestratorRoutes(ctx: RouteContext): Promise<boole
|
|||||||
broadcastExecutionStateUpdate(execution);
|
broadcastExecutionStateUpdate(execution);
|
||||||
|
|
||||||
// Broadcast EXECUTION_STARTED to WebSocket clients
|
// Broadcast EXECUTION_STARTED to WebSocket clients
|
||||||
if (wsBroadcast) {
|
broadcastToClients({
|
||||||
wsBroadcast({
|
type: 'EXECUTION_STARTED',
|
||||||
type: 'EXECUTION_STARTED',
|
payload: {
|
||||||
payload: {
|
executionId: execId,
|
||||||
executionId: execId,
|
flowId: flowId,
|
||||||
flowId: flowId,
|
sessionKey: sessionKey,
|
||||||
sessionKey: sessionKey,
|
stepName: flow.name,
|
||||||
stepName: flow.name,
|
timestamp: now
|
||||||
timestamp: now
|
}
|
||||||
}
|
});
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
// Lock the session (via WebSocket broadcast for frontend to handle)
|
// Lock the session (via WebSocket broadcast for frontend to handle)
|
||||||
if (wsBroadcast) {
|
broadcastToClients({
|
||||||
wsBroadcast({
|
type: 'CLI_SESSION_LOCKED',
|
||||||
type: 'CLI_SESSION_LOCKED',
|
payload: {
|
||||||
payload: {
|
sessionKey: sessionKey,
|
||||||
sessionKey: sessionKey,
|
reason: `Executing workflow: ${flow.name}`,
|
||||||
reason: `Executing workflow: ${flow.name}`,
|
executionId: execId,
|
||||||
executionId: execId,
|
timestamp: now
|
||||||
timestamp: now
|
}
|
||||||
}
|
});
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO: Implement actual step-by-step execution in PTY session
|
// TODO: Implement actual step-by-step execution in PTY session
|
||||||
// For now, mark as running and let the frontend handle the orchestration
|
// For now, mark as running and let the frontend handle the orchestration
|
||||||
|
|||||||
Reference in New Issue
Block a user