diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7d23ca30..d2909907 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -49,7 +49,7 @@ jobs: run: | export PATH="$HOME/.local/bin:$PATH" export PYTHONPATH="${PWD}/src:${PYTHONPATH}" - PYTHONPATH="${PWD}/src:${PYTHONPATH}" uv run pytest tests/ -v + PYTHONPATH="${PWD}/src:${PYTHONPATH}" uv run pytest tests/ -v -m "not manual" - name: Check Pylint minimum score run: | diff --git a/.github/workflows/pr-checks.yml b/.github/workflows/pr-checks.yml index eedd6904..09c22194 100644 --- a/.github/workflows/pr-checks.yml +++ b/.github/workflows/pr-checks.yml @@ -49,7 +49,7 @@ jobs: run: | export PATH="$HOME/.local/bin:$PATH" export PYTHONPATH="${PWD}/src:${PYTHONPATH}" - PYTHONPATH="${PWD}/src:${PYTHONPATH}" uv run pytest tests/ -v --tb=short + PYTHONPATH="${PWD}/src:${PYTHONPATH}" uv run pytest tests/ -v --tb=short -m "not manual" - name: Display Results Summary if: always() diff --git a/pyproject.toml b/pyproject.toml index 03ce28aa..3ce6b446 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -126,6 +126,9 @@ addopts = [ "--verbose" ] pythonpath = ["src"] +markers = [ + "manual: marks tests that require manual execution (e.g., integration tests with external services)" +] [tool.pylint.main] # Disable specific warnings that are not relevant for this project diff --git a/src/cyberautoagent.py b/src/cyberautoagent.py index e1ce86df..0da047af 100644 --- a/src/cyberautoagent.py +++ b/src/cyberautoagent.py @@ -46,9 +46,12 @@ from modules.agents.cyber_autoagent import ( AgentConfig, create_agent, - _ensure_prompt_within_budget, ) -from modules.config.system.environment import auto_setup, clean_operation_memory, setup_logging +from modules.config.system.environment import ( + auto_setup, + clean_operation_memory, + setup_logging, +) from modules.config.manager import get_config_manager from modules.handlers.base import StepLimitReached from modules.handlers.utils import ( @@ -480,7 +483,12 @@ def main(): mcp_config = config_manager.get_mcp_config(args.provider, **config_overrides) if mcp_config.enabled: - mcp_connections = list(filter(lambda c: '*' in c.plugins or args.module in c.plugins, mcp_config.connections)) + mcp_connections = list( + filter( + lambda c: "*" in c.plugins or args.module in c.plugins, + mcp_config.connections, + ) + ) else: mcp_connections = [] @@ -636,7 +644,7 @@ def cleanup_logging(): module=args.module, mcp_connections=mcp_connections, ) - agent, callback_handler = create_agent( + agent, callback_handler, feedback_manager = create_agent( target=args.target, objective=args.objective, config=config, @@ -669,14 +677,40 @@ def _initial_prompt_accessor(): ) current_message = initial_prompt - # Continue until stop condition is met while not interrupted: try: - _ensure_prompt_within_budget(agent) # Execute agent with current message + # Note: HITL feedback is now injected via HITLFeedbackInjectionHook + # which modifies the system prompt in BeforeModelInvocationEvent result = agent(current_message) + # Check for HITL pause AFTER agent execution + # This ensures pause is honored before starting next iteration + if feedback_manager: + is_paused = feedback_manager.is_paused() + logger.info( + "[HITL] Pause check: feedback_manager exists, is_paused=%s", + is_paused, + ) + if is_paused: + logger.info( + "[HITL] Execution paused after iteration - blocking until resume" + ) + print_status( + "⏸️ Execution paused - awaiting user feedback", + "INFO", + ) + # Poll until pause is cleared (by feedback or timeout) + poll_count = 0 + while feedback_manager.is_paused(): + time.sleep(0.5) + poll_count += 1 + logger.info( + "[HITL] Resumed after pause - continuing execution" + ) + print_status("▶️ Execution resumed", "INFO") + # Pass the metrics from the result to the callback handler if ( callback_handler @@ -739,7 +773,12 @@ def __init__(self, accumulated_usage): if remaining_steps > 0: # Simple continuation message current_message = f"Continue the security assessment. You have {remaining_steps} steps remaining out of {args.iterations} total. Focus on achieving the objective efficiently." + logger.debug( + "Generated continuation message for next iteration (length=%d)", + len(current_message), + ) else: + logger.info("No remaining steps - breaking execution loop") break except StepLimitReached: diff --git a/src/modules/agents/cyber_autoagent.py b/src/modules/agents/cyber_autoagent.py index 9d35a7e5..ad0130d8 100644 --- a/src/modules/agents/cyber_autoagent.py +++ b/src/modules/agents/cyber_autoagent.py @@ -33,7 +33,7 @@ configure_sdk_logging, get_config_manager, ) -from modules.config.types import MCPConnection, ServerConfig +from modules.config.types import ServerConfig from modules.config.system.logger import get_logger from modules.config.models.factory import ( create_bedrock_model, @@ -77,6 +77,12 @@ initialize_memory_system, mem0_memory, ) +from modules.handlers.hitl import ( + FeedbackInputHandler, + FeedbackManager, + HITLHookProvider, +) +from modules.handlers.hitl.feedback_injection_hook import HITLFeedbackInjectionHook from modules.tools.prompt_optimizer import prompt_optimizer warnings.filterwarnings("ignore", category=DeprecationWarning) @@ -90,12 +96,14 @@ # for better separation of concerns. See imports above for available functions. -def _discover_mcp_tools(config: AgentConfig, server_config: ServerConfig) -> List[AgentTool]: +def _discover_mcp_tools( + config: AgentConfig, server_config: ServerConfig +) -> List[AgentTool]: """Discover and register MCP tools from configured connections.""" mcp_tools = [] environ = os.environ.copy() - for mcp_conn in (config.mcp_connections or []): - if '*' in mcp_conn.plugins or config.module in mcp_conn.plugins: + for mcp_conn in config.mcp_connections or []: + if "*" in mcp_conn.plugins or config.module in mcp_conn.plugins: logger.debug("Discover MCP tools from: %s", mcp_conn) try: headers = resolve_env_vars_in_dict(mcp_conn.headers, environ) @@ -103,25 +111,36 @@ def _discover_mcp_tools(config: AgentConfig, server_config: ServerConfig) -> Lis case "stdio": if not mcp_conn.command: raise ValueError(f"{mcp_conn.transport} requires command") - command_list: List[str] = resolve_env_vars_in_list(mcp_conn.command, environ) - transport = lambda: stdio_client(StdioServerParameters( - command = command_list[0], args=command_list[1:], - env=environ, - )) + command_list: List[str] = resolve_env_vars_in_list( + mcp_conn.command, environ + ) + transport = lambda: stdio_client( # noqa: E731 + StdioServerParameters( + command=command_list[0], + args=command_list[1:], + env=environ, + ) + ) case "streamable-http": - transport = lambda: streamablehttp_client( + transport = lambda: streamablehttp_client( # noqa: E731 url=mcp_conn.server_url, headers=headers, - timeout=mcp_conn.timeoutSeconds if mcp_conn.timeoutSeconds else 30, + timeout=mcp_conn.timeoutSeconds + if mcp_conn.timeoutSeconds + else 30, ) case "sse": - transport = lambda: sse_client( + transport = lambda: sse_client( # noqa: E731 url=mcp_conn.server_url, headers=headers, - timeout=mcp_conn.timeoutSeconds if mcp_conn.timeoutSeconds else 30, + timeout=mcp_conn.timeoutSeconds + if mcp_conn.timeoutSeconds + else 30, ) case _: - raise ValueError(f"Unsupported MCP transport {mcp_conn.transport}") + raise ValueError( + f"Unsupported MCP transport {mcp_conn.transport}" + ) client = MCPClient(transport, prefix=mcp_conn.id) prefix_idx = len(mcp_conn.id) + 1 client.start() @@ -131,7 +150,10 @@ def _discover_mcp_tools(config: AgentConfig, server_config: ServerConfig) -> Lis page_token = tools.pagination_token for tool in tools: logger.debug(f"Considering tool: {tool.tool_name}") - if '*' in mcp_conn.allowed_tools or tool.tool_name[prefix_idx:] in mcp_conn.allowed_tools: + if ( + "*" in mcp_conn.allowed_tools + or tool.tool_name[prefix_idx:] in mcp_conn.allowed_tools + ): logger.debug(f"Allowed tool: {tool.tool_name}") # Wrap output and save into output path output_base_path = get_output_path( @@ -145,7 +167,9 @@ def _discover_mcp_tools(config: AgentConfig, server_config: ServerConfig) -> Lis client_used = True if not page_token: break - client_stop = lambda *_: client.stop(exc_type=None, exc_val=None, exc_tb=None) + client_stop = lambda *_: client.stop( # noqa: E731 + exc_type=None, exc_val=None, exc_tb=None + ) if client_used: atexit.register(client_stop) signal.signal(signal.SIGTERM, client_stop) @@ -195,6 +219,9 @@ def create_agent( server_config = config_manager.get_server_config(config.provider, **overrides) + # Get HITL configuration + hitl_config = server_config.hitl + # Get centralized region configuration if config.region_name is None: config.region_name = config_manager.get_default_region() @@ -667,6 +694,9 @@ def create_agent( except Exception: pass + # Check if HITL is enabled before creating handler so we can include it in init_context + hitl_enabled = hitl_config.enabled + callback_handler = ReactBridgeHandler( max_steps=config.max_steps, operation_id=operation_id, @@ -708,8 +738,11 @@ def create_agent( else {} ), }, - "observability": config_manager.getenv_bool("ENABLE_OBSERVABILITY", False), - "ui_mode": config_manager.getenv("CYBER_UI_MODE", "cli").lower(), + "observability": ( + os.getenv("ENABLE_OBSERVABILITY", "false").lower() == "true" + ), + "ui_mode": os.getenv("CYBER_UI_MODE", "cli").lower(), + "hitl_enabled": hitl_enabled, }, ) @@ -768,6 +801,59 @@ def create_agent( rebuild_interval=20, ) hooks.append(prompt_rebuild_hook) + # Create HITL hook if enabled + hitl_hook = None + feedback_manager = None + feedback_handler = None + + if hitl_enabled: + # Initialize feedback manager with configuration + feedback_manager = FeedbackManager( + memory=memory_client, + operation_id=operation_id, + emitter=callback_handler.emitter, + hitl_config=hitl_config, + ) + + # Initialize feedback input handler for receiving UI commands + feedback_handler = FeedbackInputHandler(feedback_manager=feedback_manager) + feedback_handler.start_listening() + + # Verify thread is actually running + import time + + time.sleep(0.5) # Give thread time to start + if ( + feedback_handler._listener_thread + and feedback_handler._listener_thread.is_alive() + ): + logger.info( + "[HITL] Listener thread verified: ID=%s, alive=%s", + feedback_handler._listener_thread.ident, + feedback_handler._listener_thread.is_alive(), + ) + else: + logger.error("[HITL] WARNING: Listener thread failed to start!") + + # Create HITL hook provider using centralized configuration + hitl_hook = HITLHookProvider( + feedback_manager=feedback_manager, + auto_pause_on_destructive=hitl_config.auto_pause_on_destructive, + auto_pause_on_low_confidence=hitl_config.auto_pause_on_low_confidence, + confidence_threshold=hitl_config.confidence_threshold, + ) + + # Create feedback injection hook for system prompt modification + feedback_injection_hook = HITLFeedbackInjectionHook( + feedback_manager=feedback_manager + ) + + print_status("HITL system enabled - human feedback available", "SUCCESS") + + # Add HITL hooks if enabled + if hitl_hook: + hooks.append(hitl_hook) + hooks.append(feedback_injection_hook) # Create model based on provider type try: @@ -967,4 +1053,4 @@ def create_agent( pass agent_logger.debug("Agent initialized successfully") - return agent, callback_handler + return agent, callback_handler, feedback_manager diff --git a/src/modules/config/manager.py b/src/modules/config/manager.py index de32c79f..c7f8a283 100644 --- a/src/modules/config/manager.py +++ b/src/modules/config/manager.py @@ -36,6 +36,7 @@ EvaluationConfig, SwarmConfig, SDKConfig, + HITLConfig, OutputConfig, ServerConfig, MCPConnection, @@ -63,6 +64,16 @@ logger = get_logger("Config.Manager") +LITELLM_EMBEDDING_DEFAULTS: Dict[str, Tuple[str, int]] = { + "openai": ("openai/text-embedding-3-small", 1536), + "azure": ("azure/text-embedding-3-small", 1536), + "gemini": ("models/text-embedding-004", 768), + "google": ("models/text-embedding-004", 768), + "mistral": ("multi-qa-MiniLM-L6-cos-v1", 384), + "sagemaker": ("multi-qa-MiniLM-L6-cos-v1", 384), + "xai": ("multi-qa-MiniLM-L6-cos-v1", 384), +} +DEFAULT_LITELLM_EMBEDDING: Tuple[str, int] = ("multi-qa-MiniLM-L6-cos-v1", 384) class ConfigManager: @@ -354,6 +365,9 @@ def get_server_config(self, provider: str, **overrides) -> ServerConfig: enable_telemetry=self.getenv_bool("ENABLE_SDK_TELEMETRY", True), ) + # Build HITL configuration (reads enabled from environment, others use code defaults) + hitl_config = HITLConfig() + config = ServerConfig( server_type=provider, llm=defaults["llm"], @@ -364,6 +378,7 @@ def get_server_config(self, provider: str, **overrides) -> ServerConfig: mcp=mcp_config, output=output_config, sdk=sdk_config, + hitl=hitl_config, host=host, region=defaults["region"], ) @@ -447,6 +462,11 @@ def get_swarm_model_id(self, server: Optional[str] = None, **overrides) -> str: # Final fallback to safe default aligned with Bedrock memory/evaluation defaults return "us.anthropic.claude-3-5-sonnet-20241022-v2:0" + def get_hitl_config(self, server: str, **overrides) -> HITLConfig: + """Get HITL configuration for the specified server.""" + server_config = self.get_server_config(server, **overrides) + return server_config.hitl + def get_unified_output_path( self, server: str, @@ -733,9 +753,13 @@ def get_mem0_service_config(self, server: str, **overrides) -> Dict[str, Any]: def validate_requirements(self, provider: str) -> None: """Validate that all requirements are met for the specified provider.""" # Delegate to validation module - ollama_host = _get_ollama_host_from_env(self.env) if provider == "ollama" else None + ollama_host = ( + _get_ollama_host_from_env(self.env) if provider == "ollama" else None + ) region = self.get_default_region() if provider == "bedrock" else None - server_config = self.get_server_config(provider) if provider == "ollama" else None + server_config = ( + self.get_server_config(provider) if provider == "ollama" else None + ) validate_provider(provider, self.env, ollama_host, region, server_config) @@ -900,7 +924,7 @@ def get_safe_max_tokens(self, model_id: str, buffer: float = 0.5) -> int: if not (0 < buffer <= 1.0): logger.warning( "Invalid buffer %.2f (must be between 0 and 1), using default 0.5", - buffer + buffer, ) buffer = 0.5 @@ -914,7 +938,10 @@ def get_safe_max_tokens(self, model_id: str, buffer: float = 0.5) -> int: safe = int(limits.output * buffer) logger.debug( "Safe max_tokens from models.dev: model=%s, limit=%d, safe=%d (%.0f%%)", - model_id, limits.output, safe, buffer * 100 + model_id, + limits.output, + safe, + buffer * 100, ) return safe except (ValueError, KeyError, AttributeError) as e: @@ -922,13 +949,15 @@ def get_safe_max_tokens(self, model_id: str, buffer: float = 0.5) -> int: except Exception as e: logger.error( "Unexpected error in models.dev lookup for %s: %s", - model_id, e, exc_info=True + model_id, + e, + exc_info=True, ) # Fallback to 4096 if model not found logger.warning( "Model not found in models.dev, using safe default: model=%s, safe=4096", - model_id + model_id, ) return 4096 @@ -948,26 +977,33 @@ def _get_swarm_llm_config( logger.info( "Swarm config: model=%s, max_tokens=%d (source=env override)", swarm_cfg.model_id, - swarm_cfg.max_tokens + swarm_cfg.max_tokens, ) else: swarm_cfg.max_tokens = safe_max logger.info( "Swarm config: model=%s, max_tokens=%d (source=models.dev safe default)", swarm_cfg.model_id, - swarm_cfg.max_tokens + swarm_cfg.max_tokens, ) return swarm_cfg - def _get_mcp_config(self, _server: str, defaults: Dict[str, Any], overrides: Dict[str, Any]) -> MCPConfig: + def _get_mcp_config( + self, _server: str, defaults: Dict[str, Any], overrides: Dict[str, Any] + ) -> MCPConfig: """Get MCP configuration with validation.""" - enabled = overrides.get("mcp_enabled") or os.getenv("CYBER_MCP_ENABLED", "false").lower() == "true" + enabled = ( + overrides.get("mcp_enabled") + or os.getenv("CYBER_MCP_ENABLED", "false").lower() == "true" + ) connections = [] if enabled: - conns_json = overrides.get("mcp_conns") or os.getenv("CYBER_MCP_CONNECTIONS") + conns_json = overrides.get("mcp_conns") or os.getenv( + "CYBER_MCP_CONNECTIONS" + ) if conns_json and conns_json.strip(): try: conns = json.loads(conns_json) @@ -978,53 +1014,81 @@ def _get_mcp_config(self, _server: str, defaults: Dict[str, Any], overrides: Dic for conn in conns: mcp_id = conn.get("id") if mcp_id is None or len(mcp_id) == 0: - raise ValueError("CYBER_MCP_CONNECTIONS requires an id property") + raise ValueError( + "CYBER_MCP_CONNECTIONS requires an id property" + ) if mcp_id in map(lambda x: x.id, connections): - raise ValueError("CYBER_MCP_CONNECTIONS id property must be unique") + raise ValueError( + "CYBER_MCP_CONNECTIONS id property must be unique" + ) mcp_transport = conn.get("transport") if mcp_transport not in ["stdio", "sse", "streamable-http"]: - raise ValueError(f"CYBER_MCP_CONNECTIONS {mcp_id} does not have a valid transport: {mcp_transport}") + raise ValueError( + f"CYBER_MCP_CONNECTIONS {mcp_id} does not have a valid transport: {mcp_transport}" + ) mcp_command = conn.get("command") or None if mcp_transport == "stdio": if not mcp_command: - raise ValueError("CYBER_MCP_CONNECTIONS stdio transport requires the command property") + raise ValueError( + "CYBER_MCP_CONNECTIONS stdio transport requires the command property" + ) if isinstance(mcp_command, str): mcp_command = [str] if not isinstance(mcp_command, list): - raise ValueError("CYBER_MCP_CONNECTIONS command property is expected to be a list") + raise ValueError( + "CYBER_MCP_CONNECTIONS command property is expected to be a list" + ) else: if mcp_command is not None: - raise ValueError("CYBER_MCP_CONNECTIONS network transports do not use the command property") + raise ValueError( + "CYBER_MCP_CONNECTIONS network transports do not use the command property" + ) mcp_server_url = conn.get("server_url") or None if mcp_transport == "stdio": if mcp_server_url: - raise ValueError("CYBER_MCP_CONNECTIONS stdio transport does not use the server_url property") + raise ValueError( + "CYBER_MCP_CONNECTIONS stdio transport does not use the server_url property" + ) else: if mcp_server_url is None: - raise ValueError("CYBER_MCP_CONNECTIONS network transports require the server_url property") + raise ValueError( + "CYBER_MCP_CONNECTIONS network transports require the server_url property" + ) mcp_headers = conn.get("headers") if mcp_headers is not None and not isinstance(mcp_headers, dict): - raise ValueError("CYBER_MCP_CONNECTIONS headers property is expected to be a dictionary") + raise ValueError( + "CYBER_MCP_CONNECTIONS headers property is expected to be a dictionary" + ) mcp_plugins = conn.get("plugins") if mcp_plugins is not None and not isinstance(mcp_plugins, list): - raise ValueError("CYBER_MCP_CONNECTIONS plugins property is expected to be a list") + raise ValueError( + "CYBER_MCP_CONNECTIONS plugins property is expected to be a list" + ) if not mcp_plugins or "*" in mcp_plugins: mcp_plugins = ["*"] mcp_timeout = conn.get("timeoutSeconds") if mcp_timeout is not None and not isinstance(mcp_timeout, int): - raise ValueError("CYBER_MCP_CONNECTIONS timeoutSeconds is expected to be an integer") + raise ValueError( + "CYBER_MCP_CONNECTIONS timeoutSeconds is expected to be an integer" + ) if mcp_timeout is not None and mcp_timeout < 0: - raise ValueError("CYBER_MCP_CONNECTIONS timeoutSeconds is expected to be a positive integer") + raise ValueError( + "CYBER_MCP_CONNECTIONS timeoutSeconds is expected to be a positive integer" + ) mcp_allowed_tools = conn.get("allowedTools") - if mcp_allowed_tools is not None and not isinstance(mcp_allowed_tools, list): - raise ValueError("CYBER_MCP_CONNECTIONS allowedTools property is expected to be a list") + if mcp_allowed_tools is not None and not isinstance( + mcp_allowed_tools, list + ): + raise ValueError( + "CYBER_MCP_CONNECTIONS allowedTools property is expected to be a list" + ) if not mcp_allowed_tools or "*" in mcp_allowed_tools: mcp_allowed_tools = ["*"] diff --git a/src/modules/config/models/dev_client.py b/src/modules/config/models/dev_client.py index 79ebca87..8275ba93 100644 --- a/src/modules/config/models/dev_client.py +++ b/src/modules/config/models/dev_client.py @@ -51,6 +51,7 @@ class ModelLimits: context: Maximum input tokens (context window) output: Maximum output tokens (completion limit) """ + context: int output: int @@ -66,6 +67,7 @@ class ModelPricing: cache_write: Cost per million cached write tokens (optional) reasoning: Cost per million reasoning tokens (optional, for o1/o3 models) """ + input: float output: float cache_read: Optional[float] = None @@ -91,6 +93,7 @@ class ModelCapabilities: modalities_input: Supported input modalities (text, image, audio, video, pdf) modalities_output: Supported output modalities (text, image, audio) """ + name: str reasoning: bool tool_call: bool @@ -117,6 +120,7 @@ class ModelInfo: limits: Token limits pricing: Pricing information (None if not available) """ + provider: str model_id: str full_id: str @@ -258,14 +262,14 @@ def list_models(self, provider: Optional[str] = None) -> List[str]: if provider: provider_data = data.get(provider, {}) - models_data = provider_data.get('models', {}) + models_data = provider_data.get("models", {}) return sorted(models_data.keys()) # Return all models across all providers all_models = [] for provider_id, provider_data in data.items(): - if 'models' in provider_data: - for model_id in provider_data['models'].keys(): + if "models" in provider_data: + for model_id in provider_data["models"].keys(): all_models.append(f"{provider_id}/{model_id}") return sorted(all_models) @@ -310,7 +314,9 @@ def _get_data(self) -> Dict: with open(self.cache_file) as f: self._data = json.load(f) self._data_source = "cache" - logger.info(f"Loaded models from cache ({len(self._data)} providers)") + logger.info( + f"Loaded models from cache ({len(self._data)} providers)" + ) return self._data except Exception as e: logger.warning(f"Failed to load cache: {e}") @@ -322,11 +328,13 @@ def _get_data(self) -> Dict: # Use httpx if available, fall back to urllib try: import httpx + response = httpx.get(self.API_URL, timeout=10.0, follow_redirects=True) response.raise_for_status() self._data = response.json() except ImportError: import urllib.request + with urllib.request.urlopen(self.API_URL, timeout=10) as response: self._data = json.loads(response.read().decode()) @@ -347,7 +355,9 @@ def _get_data(self) -> Dict: with open(self.snapshot_file) as f: self._data = json.load(f) self._data_source = "snapshot" - logger.info(f"Loaded models from snapshot ({len(self._data)} providers)") + logger.info( + f"Loaded models from snapshot ({len(self._data)} providers)" + ) return self._data except Exception as e: logger.error(f"Failed to load snapshot: {e}") @@ -388,7 +398,7 @@ def _save_cache(self, data: Dict): """ try: self.cache_dir.mkdir(parents=True, exist_ok=True) - with open(self.cache_file, 'w') as f: + with open(self.cache_file, "w") as f: json.dump(data, f, indent=2) logger.debug(f"Saved cache: {self.cache_file}") except Exception as e: @@ -405,18 +415,18 @@ def _lookup_model(self, data: Dict, model_id: str) -> Optional[ModelInfo]: ModelInfo if found, None otherwise """ # Handle provider/model format - if '/' in model_id: - parts = model_id.split('/', 1) + if "/" in model_id: + parts = model_id.split("/", 1) if len(parts) == 2: provider, model = parts return self._parse_model(data, provider, model) # Search across all providers for exact match for provider_id, provider_data in data.items(): - if 'models' not in provider_data: + if "models" not in provider_data: continue - if model_id in provider_data['models']: + if model_id in provider_data["models"]: return self._parse_model(data, provider_id, model_id) return None @@ -439,14 +449,14 @@ def _fuzzy_lookup(self, data: Dict, model_id: str) -> Optional[ModelInfo]: """ # Provider aliases mapping provider_aliases = { - 'moonshot': 'moonshotai', - 'anthropic': 'amazon-bedrock', # When used with ARN format - 'gemini': 'google', # Gemini models are under google provider + "moonshot": "moonshotai", + "anthropic": "amazon-bedrock", # When used with ARN format + "gemini": "google", # Gemini models are under google provider } # Handle provider/model format with alias resolution - if '/' in model_id: - parts = model_id.split('/', 1) + if "/" in model_id: + parts = model_id.split("/", 1) if len(parts) == 2: provider, model = parts # Try with aliased provider @@ -457,25 +467,25 @@ def _fuzzy_lookup(self, data: Dict, model_id: str) -> Optional[ModelInfo]: return info # Normalize dots to dashes (e.g., claude-3.5-haiku → claude-3-5-haiku) - normalized = model_id.replace('.', '-') + normalized = model_id.replace(".", "-") if normalized != model_id: info = self._lookup_model(data, normalized) if info: return info # Handle Bedrock ARN format (us.anthropic.claude-sonnet-4-5-20250929-v1:0) - if model_id.startswith('us.') or model_id.startswith('anthropic.'): + if model_id.startswith("us.") or model_id.startswith("anthropic."): # Extract the actual model name - parts = model_id.split('.') + parts = model_id.split(".") if len(parts) >= 2: # Try "anthropic/claude-sonnet-4-5-20250929" - bedrock_model = '.'.join(parts[1:]) + bedrock_model = ".".join(parts[1:]) info = self._lookup_model(data, f"amazon-bedrock/{bedrock_model}") if info: return info # Try with -latest suffix removed - if model_id.endswith('-latest'): + if model_id.endswith("-latest"): base = model_id[:-7] info = self._lookup_model(data, base) if info: @@ -483,7 +493,9 @@ def _fuzzy_lookup(self, data: Dict, model_id: str) -> Optional[ModelInfo]: return None - def _parse_model(self, data: Dict, provider: str, model: str) -> Optional[ModelInfo]: + def _parse_model( + self, data: Dict, provider: str, model: str + ) -> Optional[ModelInfo]: """Parse model data into ModelInfo. Args: @@ -496,41 +508,45 @@ def _parse_model(self, data: Dict, provider: str, model: str) -> Optional[ModelI """ try: provider_data = data[provider] - model_data = provider_data['models'][model] + model_data = provider_data["models"][model] # Parse capabilities capabilities = ModelCapabilities( - name=model_data.get('name', model), - reasoning=model_data.get('reasoning', False), - tool_call=model_data.get('tool_call', False), - attachment=model_data.get('attachment', False), - temperature=model_data.get('temperature', True), - structured_output=model_data.get('structured_output'), - knowledge=model_data.get('knowledge'), - release_date=model_data.get('release_date'), - last_updated=model_data.get('last_updated'), - open_weights=model_data.get('open_weights', False), - modalities_input=model_data.get('modalities', {}).get('input', ['text']), - modalities_output=model_data.get('modalities', {}).get('output', ['text']), + name=model_data.get("name", model), + reasoning=model_data.get("reasoning", False), + tool_call=model_data.get("tool_call", False), + attachment=model_data.get("attachment", False), + temperature=model_data.get("temperature", True), + structured_output=model_data.get("structured_output"), + knowledge=model_data.get("knowledge"), + release_date=model_data.get("release_date"), + last_updated=model_data.get("last_updated"), + open_weights=model_data.get("open_weights", False), + modalities_input=model_data.get("modalities", {}).get( + "input", ["text"] + ), + modalities_output=model_data.get("modalities", {}).get( + "output", ["text"] + ), ) # Parse limits - limit_data = model_data.get('limit', {}) + limit_data = model_data.get("limit", {}) limits = ModelLimits( - context=limit_data.get('context', 0), - output=limit_data.get('output', 0), + context=limit_data.get("context", 0), + output=limit_data.get("output", 0), ) # Parse pricing (optional) pricing = None - if 'cost' in model_data: - cost_data = model_data['cost'] + if "cost" in model_data: + cost_data = model_data["cost"] pricing = ModelPricing( - input=cost_data.get('input', 0.0), - output=cost_data.get('output', 0.0), - cache_read=cost_data.get('cache_read'), - cache_write=cost_data.get('cache_write'), - reasoning=cost_data.get('reasoning'), + input=cost_data.get("input", 0.0), + output=cost_data.get("output", 0.0), + cache_read=cost_data.get("cache_read"), + cache_write=cost_data.get("cache_write"), + reasoning=cost_data.get("reasoning"), ) return ModelInfo( @@ -564,10 +580,10 @@ def get_models_client() -> ModelsDevClient: # Public API __all__ = [ - 'ModelLimits', - 'ModelPricing', - 'ModelCapabilities', - 'ModelInfo', - 'ModelsDevClient', - 'get_models_client', + "ModelLimits", + "ModelPricing", + "ModelCapabilities", + "ModelInfo", + "ModelsDevClient", + "get_models_client", ] diff --git a/src/modules/config/models/factory.py b/src/modules/config/models/factory.py index 820135b5..fc2b830c 100644 --- a/src/modules/config/models/factory.py +++ b/src/modules/config/models/factory.py @@ -28,6 +28,7 @@ def _get_config_manager(): """Lazy import to avoid circular dependency.""" from modules.config.manager import get_config_manager + return get_config_manager() @@ -254,9 +255,7 @@ def _parse_spec(spec: str) -> Optional[List[Dict[str, List[str]]]]: return parsed try: config_manager = _get_config_manager() - config_fallbacks = ( - config_manager.get_context_window_fallbacks("litellm") or [] - ) + config_fallbacks = config_manager.get_context_window_fallbacks("litellm") or [] if config_fallbacks: copied: List[Dict[str, List[str]]] = [] for mapping in config_fallbacks: @@ -436,9 +435,7 @@ def create_bedrock_model( # Add additional request fields if present (e.g., anthropic_beta for extended context) if config.get("additional_request_fields"): - model_kwargs["additional_request_fields"] = config[ - "additional_request_fields" - ] + model_kwargs["additional_request_fields"] = config["additional_request_fields"] return BedrockModel(**model_kwargs) diff --git a/src/modules/config/providers/ollama_config.py b/src/modules/config/providers/ollama_config.py index 26dfd3fd..75910cb9 100644 --- a/src/modules/config/providers/ollama_config.py +++ b/src/modules/config/providers/ollama_config.py @@ -46,9 +46,7 @@ def get_ollama_host(env_reader: EnvironmentReader) -> str: except (requests.exceptions.RequestException, ConnectionError): pass # Fallback to host.docker.internal if no connection works - logger.debug( - "No Ollama connection found, falling back to host.docker.internal" - ) + logger.debug("No Ollama connection found, falling back to host.docker.internal") return "http://host.docker.internal:11434" # Native execution - use localhost return "http://localhost:11434" diff --git a/src/modules/config/system/defaults.py b/src/modules/config/system/defaults.py index 22b27791..3e0a32ad 100644 --- a/src/modules/config/system/defaults.py +++ b/src/modules/config/system/defaults.py @@ -136,8 +136,7 @@ def build_litellm_defaults() -> Dict[str, Any]: provider=ModelProvider.LITELLM, model_id="bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0", # Default to Bedrock via LiteLLM temperature=0.95, - max_tokens=32000, - + max_tokens=32000, ), "embedding": EmbeddingConfig( provider=ModelProvider.LITELLM, diff --git a/src/modules/config/system/environment.py b/src/modules/config/system/environment.py index e18e0465..2a70eb2d 100644 --- a/src/modules/config/system/environment.py +++ b/src/modules/config/system/environment.py @@ -383,7 +383,8 @@ def cleanup_tee_outputs(): file_handler.setFormatter(formatter) # Console handler - only show warnings and above unless verbose - console_handler = logging.StreamHandler(sys.__stdout__) # Use original stdout + # Use current stdout (TeeOutput) so React UI can capture logs + console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.INFO if verbose else logging.WARNING) console_handler.setFormatter(formatter) @@ -409,6 +410,14 @@ def cleanup_tee_outputs(): root_file_handler.setFormatter(formatter) root_logger.addHandler(root_file_handler) + # In verbose mode, also send INFO logs to console for all modules + # Use current stdout (TeeOutput) so React UI can capture logs + if verbose: + root_console_handler = logging.StreamHandler(sys.stdout) + root_console_handler.setLevel(logging.INFO) + root_console_handler.setFormatter(formatter) + root_logger.addHandler(root_console_handler) + # Suppress verbose AWS credential detection messages logging.getLogger("boto3").setLevel(logging.WARNING) logging.getLogger("botocore").setLevel(logging.WARNING) diff --git a/src/modules/config/system/logger.py b/src/modules/config/system/logger.py index ec634991..ed866868 100644 --- a/src/modules/config/system/logger.py +++ b/src/modules/config/system/logger.py @@ -62,6 +62,7 @@ def configure_sdk_logging(enable_debug: bool = False) -> None: Args: enable_debug: If True, enable verbose logging for SDK components """ + # Suppress unrecognized tool specification warnings from Strands toolkit registry # These are benign warnings from the Strands SDK when built-in tools (stop, http_request, python_repl) # are processed during tool registration. The tools work correctly despite the warnings. diff --git a/src/modules/config/system/validation.py b/src/modules/config/system/validation.py index 6b9a17df..3e6c8aa8 100644 --- a/src/modules/config/system/validation.py +++ b/src/modules/config/system/validation.py @@ -108,7 +108,8 @@ def validate_ollama_requirements( # Require at least one required model to be available has_required = any( - any(req in model for model in available_models) for req in required_models + any(req in model for model in available_models) + for req in required_models ) if not has_required: @@ -151,7 +152,9 @@ def validate_bedrock_model_access(region: str) -> None: # Model-specific errors will be handled by strands-agents during actual usage -def validate_aws_requirements(env_reader: EnvironmentReader, region: str = None) -> None: +def validate_aws_requirements( + env_reader: EnvironmentReader, region: str = None +) -> None: """Validate AWS requirements including Bedrock model access. Supports either standard AWS credentials (ACCESS_KEY/SECRET or PROFILE) @@ -195,7 +198,9 @@ def validate_aws_requirements(env_reader: EnvironmentReader, region: str = None) validate_bedrock_model_access(region) -def validate_litellm_requirements(env_reader: EnvironmentReader, model_id: str = "") -> None: +def validate_litellm_requirements( + env_reader: EnvironmentReader, model_id: str = "" +) -> None: """Validate LiteLLM requirements based on model provider prefix. LiteLLM handles most validation internally: diff --git a/src/modules/config/types.py b/src/modules/config/types.py index aa1b99b7..75f86f76 100644 --- a/src/modules/config/types.py +++ b/src/modules/config/types.py @@ -323,6 +323,28 @@ class SDKConfig: tool_timeout_seconds: int = 300 +@dataclass +class HITLConfig: + """Configuration for Human-in-the-Loop (HITL) system.""" + + # Feature toggle - only this reads from environment + enabled: bool = field( + default_factory=lambda: os.getenv("CYBER_AGENT_HITL_ENABLED", "false").lower() + == "true" + ) + + # Timeout for manual (user-triggered via [i] key) pauses in seconds + manual_pause_timeout: int = 180 + + # Timeout for auto-pause (destructive operations/low confidence) in seconds + auto_pause_timeout: int = 120 + + # Auto-pause triggers + auto_pause_on_destructive: bool = True + auto_pause_on_low_confidence: bool = True + confidence_threshold: int = 90 # Threshold for low confidence (0-100) + + @dataclass class OutputConfig: """Configuration for output directory management.""" @@ -346,6 +368,6 @@ class ServerConfig: mcp: MCPConfig = field(default_factory=MCPConfig) output: OutputConfig = field(default_factory=OutputConfig) sdk: SDKConfig = field(default_factory=SDKConfig) + hitl: HITLConfig = field(default_factory=HITLConfig) host: Optional[str] = None region: str = "us-east-1" # Default, can be overridden via environment - diff --git a/src/modules/handlers/conversation_budget.py b/src/modules/handlers/conversation_budget.py index d4069733..d0067192 100644 --- a/src/modules/handlers/conversation_budget.py +++ b/src/modules/handlers/conversation_budget.py @@ -8,7 +8,7 @@ import os import time from dataclasses import dataclass -from typing import Any, Optional, Callable, Sequence, TypedDict +from typing import Any, Optional, Callable, Sequence, TypedDict, Dict from strands import Agent from strands.agent.conversation_manager import ( @@ -606,7 +606,7 @@ def _apply_mapper(self, agent: Agent) -> None: logger.debug( "Skipping pruning for small conversation: %d messages (agent=%s)", total, - getattr(agent, "name", "unknown") + getattr(agent, "name", "unknown"), ) return @@ -693,7 +693,7 @@ def _safe_estimate_tokens(agent: Agent) -> Optional[int]: if messages is None: logger.warning( "TOKEN ESTIMATION FAILED: agent.messages is None (agent=%s)", - getattr(agent, "name", "unknown") + getattr(agent, "name", "unknown"), ) return None @@ -701,14 +701,14 @@ def _safe_estimate_tokens(agent: Agent) -> Optional[int]: logger.warning( "TOKEN ESTIMATION FAILED: agent.messages is not a list (type=%s, agent=%s)", type(messages).__name__, - getattr(agent, "name", "unknown") + getattr(agent, "name", "unknown"), ) return None if len(messages) == 0: logger.info( "TOKEN ESTIMATION: agent.messages is empty, returning 0 tokens (agent=%s)", - getattr(agent, "name", "unknown") + getattr(agent, "name", "unknown"), ) return 0 @@ -717,7 +717,7 @@ def _safe_estimate_tokens(agent: Agent) -> Optional[int]: "TOKEN ESTIMATION: Estimated %d tokens from %d messages (agent=%s)", estimated, len(messages), - getattr(agent, "name", "unknown") + getattr(agent, "name", "unknown"), ) return estimated except Exception as e: @@ -725,7 +725,7 @@ def _safe_estimate_tokens(agent: Agent) -> Optional[int]: "TOKEN ESTIMATION ERROR: Exception during estimation (agent=%s, error=%s)", getattr(agent, "name", "unknown"), str(e), - exc_info=True + exc_info=True, ) return None @@ -825,7 +825,9 @@ def _get_char_to_token_ratio_dynamic(model_id: str) -> float: provider = info.provider.lower() # Provider-specific ratios based on tokenizer characteristics - if "anthropic" in provider or ("bedrock" in provider and "claude" in model_id.lower()): + if "anthropic" in provider or ( + "bedrock" in provider and "claude" in model_id.lower() + ): ratio = 3.7 # Claude tokenizer elif "google" in provider or "gemini" in provider or "vertex" in provider: ratio = 4.2 # Gemini tokenizer (SentencePiece) @@ -834,10 +836,14 @@ def _get_char_to_token_ratio_dynamic(model_id: str) -> float: elif "openai" in provider or "azure" in provider: # Check if it's a GPT model model_lower = model_id.lower() - if any(gpt in model_lower for gpt in ["gpt-4", "gpt-5", "gpt4", "gpt5"]): + if any( + gpt in model_lower for gpt in ["gpt-4", "gpt-5", "gpt4", "gpt5"] + ): ratio = 4.0 # GPT tokenizer except Exception as e: - logger.debug("models.dev lookup failed for ratio: model=%s, error=%s", model_id, e) + logger.debug( + "models.dev lookup failed for ratio: model=%s, error=%s", model_id, e + ) # Cache and return _RATIO_CACHE[model_id] = ratio @@ -911,7 +917,10 @@ def _estimate_prompt_tokens(agent: Agent) -> int: logger.debug( "TOKEN ESTIMATION: %d chars / %.1f ratio = %d tokens (model=%s)", - total_chars, ratio, estimated_tokens, model_id + total_chars, + ratio, + estimated_tokens, + model_id, ) return estimated_tokens @@ -976,14 +985,14 @@ def _ensure_prompt_within_budget(agent: Agent) -> None: "BUDGET CHECK FAILED: Token estimation returned None for agent=%s. " "Cannot perform budget enforcement without token count. " "This may indicate empty messages or estimation error.", - getattr(agent, "name", "unknown") + getattr(agent, "name", "unknown"), ) # Try to use telemetry as fallback if telemetry_tokens is not None and telemetry_tokens > 0: logger.info( "BUDGET CHECK FALLBACK: Using telemetry tokens (%d) as proxy for context size", - telemetry_tokens + telemetry_tokens, ) current_tokens = telemetry_tokens else: diff --git a/src/modules/handlers/events/emitters.py b/src/modules/handlers/events/emitters.py index e0cc2f66..9c3c2c10 100644 --- a/src/modules/handlers/events/emitters.py +++ b/src/modules/handlers/events/emitters.py @@ -1,4 +1,25 @@ -"""Event emitters for different transport mechanisms.""" +"""Event emitters for different transport mechanisms. + +HITL Event Types: +----------------- +The following events are emitted by the HITL system: + +hitl_pause_requested: + Emitted when tool execution is paused for review. + Fields: tool_name, tool_id, parameters, confidence, reason + +hitl_feedback_submitted: + Emitted when user provides feedback. + Fields: feedback_type, content, tool_id, timestamp + +hitl_agent_interpretation: + Emitted when agent interprets user feedback. + Fields: tool_id, interpretation, modified_parameters, awaiting_approval + +hitl_resume: + Emitted when execution resumes after feedback. + Fields: tool_id, modified_parameters, approved +""" import hashlib import json diff --git a/src/modules/handlers/hitl/__init__.py b/src/modules/handlers/hitl/__init__.py new file mode 100644 index 00000000..9121b6c0 --- /dev/null +++ b/src/modules/handlers/hitl/__init__.py @@ -0,0 +1,16 @@ +""" +Human-in-the-Loop (HITL) feedback system for Cyber-AutoAgent. + +This module provides real-time intervention capabilities during agent execution, +allowing users to pause, review, correct, and guide agent actions. +""" + +from .feedback_handler import FeedbackInputHandler +from .feedback_manager import FeedbackManager +from .hitl_hook_provider import HITLHookProvider + +__all__ = [ + "FeedbackManager", + "HITLHookProvider", + "FeedbackInputHandler", +] diff --git a/src/modules/handlers/hitl/feedback_handler.py b/src/modules/handlers/hitl/feedback_handler.py new file mode 100644 index 00000000..2dc29b9d --- /dev/null +++ b/src/modules/handlers/hitl/feedback_handler.py @@ -0,0 +1,183 @@ +"""Handler for receiving feedback from React UI via stdin.""" + +import json +import logging +import select +import sys +import threading +from typing import Optional + +from .feedback_manager import FeedbackManager +from .types import FeedbackType + +logger = logging.getLogger(__name__) + + +class FeedbackInputHandler: + """Handles incoming feedback from React UI via stdin commands.""" + + def __init__(self, feedback_manager: FeedbackManager): + """Initialize feedback input handler. + + Args: + feedback_manager: FeedbackManager instance + """ + self.feedback_manager = feedback_manager + self._running = False + self._listener_thread: Optional[threading.Thread] = None + + logger.info("FeedbackInputHandler initialized") + + def start_listening(self) -> None: + """Start listening for feedback commands in background thread.""" + if self._running: + logger.warning("Feedback listener already running") + return + + self._running = True + self._listener_thread = threading.Thread( + target=self._listen_loop, + daemon=True, + name="HITLFeedbackListener", + ) + self._listener_thread.start() + logger.info("Feedback listener started") + + def stop_listening(self) -> None: + """Stop listening for feedback commands.""" + self._running = False + if self._listener_thread: + self._listener_thread.join(timeout=1.0) + logger.info("Feedback listener stopped") + + def _listen_loop(self) -> None: + """Main listening loop for stdin commands (runs in background thread).""" + import time + + logger.info("[HITL-InputHandler] Listener thread STARTED - monitoring stdin") + + iteration = 0 + last_heartbeat = time.time() + + while self._running: + iteration += 1 + current_time = time.time() + + # Heartbeat every 5 seconds to prove thread is alive + if current_time - last_heartbeat > 5: + logger.info(f"[HITL-InputHandler] Heartbeat - iteration {iteration}") + last_heartbeat = current_time + + try: + # Check if stdin has data available (non-blocking) + if select.select([sys.stdin], [], [], 0.5)[0]: + logger.info( + f"[HITL-InputHandler] Stdin data available at iteration {iteration}" + ) + line = sys.stdin.readline() + if line: + logger.warning( + f"[HITL-InputHandler] Line received: {line[:200]}" + ) + self._process_input_line(line) + except Exception as e: + logger.error("Error in feedback listener: %s", e, exc_info=True) + + logger.info("[HITL-InputHandler] Listener thread EXITED") + + def _process_input_line(self, line: str) -> None: + """Process a line of input from stdin. + + Args: + line: Input line to process + """ + # Check for test marker + if "TEST_STDIN_WORKS" in line: + logger.warning( + "[HITL-InputHandler] TEST STDIN WORKS - stdin is functional!" + ) + + # Look for HITL command format: __HITL_COMMAND____HITL_COMMAND_END__ + if "__HITL_COMMAND__" in line: + logger.info("[HITL-InputHandler] HITL command detected, parsing...") + try: + start = line.index("__HITL_COMMAND__") + len("__HITL_COMMAND__") + end = line.index("__HITL_COMMAND_END__") + command_json = line[start:end] + command = json.loads(command_json) + logger.info( + f"[HITL-InputHandler] Command parsed: type={command.get('type')}" + ) + self.handle_feedback_command(command) + except (ValueError, json.JSONDecodeError) as e: + logger.warning("Failed to parse HITL command: %s", e) + + def handle_feedback_command(self, command: dict) -> None: + """Process feedback command from UI. + + Args: + command: Feedback command dictionary with fields: + - type: Command type ("submit_feedback", "request_pause") + - Additional fields depending on type + """ + command_type = command.get("type") + + logger.info("Received HITL command: %s", command_type) + + if command_type == "submit_feedback": + self._handle_submit_feedback(command) + elif command_type == "request_pause": + self._handle_pause_request(command) + else: + logger.warning("Unknown feedback command type: %s", command_type) + + def _handle_submit_feedback(self, command: dict) -> None: + """Handle feedback submission command. + + Args: + command: Command dict with feedback_type, content, tool_id + """ + try: + feedback_type_str = command.get("feedback_type", "correction") + feedback_type = FeedbackType(feedback_type_str) + content = command.get("content", "") + tool_id = command.get("tool_id", "") + + self.feedback_manager.submit_feedback( + feedback_type=feedback_type, + content=content, + tool_id=tool_id, + ) + + logger.info( + "Feedback submitted: type=%s, tool_id=%s", + feedback_type.value, + tool_id, + ) + + except Exception as e: + logger.error("Failed to submit feedback: %s", e, exc_info=True) + + def _handle_pause_request(self, command: dict) -> None: + """Handle pause request from user. + + Blocks listener thread until feedback received or timeout. + + Args: + command: Command dict with optional 'is_manual' field + """ + try: + is_manual = command.get("is_manual", True) + self.feedback_manager.request_pause(is_manual=is_manual) + + # Block until feedback or timeout + # This runs on listener thread, so it doesn't block agent + feedback_received = self.feedback_manager.wait_for_feedback() + + if feedback_received: + logger.info("Pause resumed after feedback") + else: + logger.warning("Pause timed out - auto-resumed") + + except Exception as e: + logger.error("Failed to handle pause request: %s", e, exc_info=True) diff --git a/src/modules/handlers/hitl/feedback_injection_hook.py b/src/modules/handlers/hitl/feedback_injection_hook.py new file mode 100644 index 00000000..4926cf62 --- /dev/null +++ b/src/modules/handlers/hitl/feedback_injection_hook.py @@ -0,0 +1,81 @@ +"""HITL feedback injection hook for modifying agent system prompt.""" + +import logging +from typing import TYPE_CHECKING, Any + +from strands.hooks import BeforeModelCallEvent, HookProvider, HookRegistry + +if TYPE_CHECKING: + from .feedback_manager import FeedbackManager + +logger = logging.getLogger(__name__) + + +class HITLFeedbackInjectionHook(HookProvider): + """Hook that injects pending HITL feedback into agent system prompt. + + This hook uses the BeforeModelInvocationEvent to append pending user + feedback to the agent's system prompt before each model invocation. + This ensures feedback is processed as part of the agent's core context + rather than as a conversation message. + + Pattern based on prompt_rebuild_hook.py which modifies + event.agent.system_prompt directly. + """ + + def __init__(self, feedback_manager: "FeedbackManager"): + """Initialize hook with feedback manager. + + Args: + feedback_manager: FeedbackManager instance to check for pending feedback + """ + self.feedback_manager = feedback_manager + logger.info( + "[HITL-HOOK] HITLFeedbackInjectionHook initialized for operation %s", + feedback_manager.operation_id, + ) + + def register_hooks(self, registry: HookRegistry, **kwargs: Any): + """Register BeforeModelCallEvent callback. + + Args: + registry: Hook registry to register callback with + **kwargs: Additional keyword arguments from base class + """ + registry.add_callback(BeforeModelCallEvent, self.inject_feedback) + logger.debug("[HITL-HOOK] Registered BeforeModelCallEvent callback") + + def inject_feedback(self, event: BeforeModelCallEvent): + """Inject pending feedback into system prompt before model invocation. + + This method is called before each model invocation. If feedback is + pending, it appends the formatted feedback message to the agent's + system prompt and clears the pending feedback. + + Args: + event: BeforeModelCallEvent containing agent context + """ + feedback_message = self.feedback_manager.get_pending_feedback_message() + + if feedback_message: + logger.info( + "[HITL-HOOK] Injecting feedback into system prompt (length=%d chars)", + len(feedback_message), + ) + logger.debug( + "[HITL-HOOK] Feedback preview:\n%s", + feedback_message[:300] + "..." + if len(feedback_message) > 300 + else feedback_message, + ) + + # Append feedback to system prompt + current_prompt = event.agent.system_prompt or "" + event.agent.system_prompt = f"{current_prompt}\n\n{feedback_message}" + + # Clear feedback after injection to prevent duplicate injection + self.feedback_manager.clear_pending_feedback() + + logger.info("[HITL-HOOK] Feedback successfully injected into system prompt") + else: + logger.debug("[HITL-HOOK] No pending feedback to inject") diff --git a/src/modules/handlers/hitl/feedback_manager.py b/src/modules/handlers/hitl/feedback_manager.py new file mode 100644 index 00000000..4cbd88a6 --- /dev/null +++ b/src/modules/handlers/hitl/feedback_manager.py @@ -0,0 +1,339 @@ +"""Feedback manager for HITL workflows.""" + +import logging +import threading +import time +from typing import TYPE_CHECKING, Any, Dict, Optional + +from .types import ( + FeedbackType, + HITLState, + ToolInvocation, + UserFeedback, +) + +if TYPE_CHECKING: + # Import only during type checking to avoid circular dependencies at runtime + # This pattern allows type hints without creating import cycles + from modules.config.manager import HITLConfig + +logger = logging.getLogger(__name__) + + +class FeedbackManager: + """Manages HITL feedback state and workflow.""" + + def __init__( + self, + memory=None, + operation_id: Optional[str] = None, + emitter=None, + hitl_config: Optional["HITLConfig"] = None, + ): + """Initialize feedback manager. + + Args: + memory: Memory client for storing interventions + operation_id: Operation identifier + emitter: Event emitter for UI communication + hitl_config: HITL configuration with timeout settings + """ + self.memory = memory + self.operation_id = operation_id + self.emitter = emitter + + # Store timeout configuration for pause mechanism + if hitl_config: + self.manual_pause_timeout = hitl_config.manual_pause_timeout + self.auto_pause_timeout = hitl_config.auto_pause_timeout + else: + # Default timeouts if no config provided + self.manual_pause_timeout = 120 + self.auto_pause_timeout = 30 + + # State tracking + self.state = HITLState.ACTIVE + self.pending_tool: Optional[ToolInvocation] = None + self.pending_feedback: Optional[UserFeedback] = None + + # Pause mechanism using threading.Event for blocking coordination + self._pause_event = threading.Event() + self._pause_event.set() # Start in non-paused state (event is set) + self._is_manual_pause = False # Track if current pause is manual vs auto + + # Feedback queue for tools awaiting approval + self.feedback_queue: Dict[str, UserFeedback] = {} + + logger.info("FeedbackManager initialized for operation %s", operation_id) + + def request_pause( + self, + tool_name: Optional[str] = None, + tool_id: Optional[str] = None, + parameters: Optional[Dict[str, Any]] = None, + confidence: Optional[float] = None, + reason: Optional[str] = None, + is_manual: bool = False, + ) -> None: + """Request execution pause (auto or manual). + + Args: + tool_name: Name of tool to review (auto-generated for manual pause) + tool_id: Unique tool invocation ID (auto-generated for manual pause) + parameters: Tool parameters (empty dict for manual pause) + confidence: Confidence score 0-100 (None for manual pause) + reason: Reason for pause + is_manual: True for user-triggered pause, False for auto-pause + """ + # Generate synthetic data for manual pauses + if is_manual: + tool_name = tool_name or "manual_intervention" + tool_id = tool_id or f"manual_{int(time.time() * 1000)}" + parameters = parameters or {} + reason = reason or "User requested manual intervention" + + log_msg = "manual pause" if is_manual else f"auto-pause for {tool_name}" + logger.info( + "Pause requested: %s (id=%s, reason=%s)", + log_msg, + tool_id, + reason, + ) + + self.state = HITLState.PAUSED + self.pending_tool = ToolInvocation( + tool_name=tool_name, + tool_id=tool_id, + parameters=parameters, + confidence=confidence, + reason=reason, + ) + + # Block execution by clearing the event + self._is_manual_pause = is_manual + self._pause_event.clear() + + # Emit pause event to UI + if self.emitter: + # Include timeout for UI display + timeout_seconds = ( + self.manual_pause_timeout if is_manual else self.auto_pause_timeout + ) + + self.emitter.emit( + { + "type": "hitl_pause_requested", + "tool_name": tool_name, + "tool_id": tool_id, + "parameters": parameters, + "confidence": confidence, + "reason": reason, + "is_manual": is_manual, + "timeout_seconds": timeout_seconds, + } + ) + + def wait_for_feedback(self) -> bool: + """Block execution until feedback is received or timeout expires. + + Uses appropriate timeout based on pause type (manual vs auto). + Returns True if feedback received, False if timeout. + """ + if self.state != HITLState.PAUSED: + # Not paused, no need to wait + return True + + # Use appropriate timeout based on pause type + timeout = ( + self.manual_pause_timeout + if self._is_manual_pause + else self.auto_pause_timeout + ) + pause_type = "manual" if self._is_manual_pause else "auto" + + logger.info( + "[HITL-FM] Blocking execution - waiting for feedback (%s pause, timeout=%ds)", + pause_type, + timeout, + ) + + # Block until event is set (feedback received) or timeout expires + feedback_received = self._pause_event.wait(timeout=timeout) + + if feedback_received: + logger.info("[HITL-FM] Feedback received, execution resuming") + return True + else: + logger.warning( + "[HITL-FM] Timeout expired after %ds, auto-resuming execution", timeout + ) + # Auto-resume on timeout + self.resume() + return False + + def submit_feedback( + self, + feedback_type: FeedbackType, + content: str, + tool_id: str, + ) -> None: + """Submit user feedback and auto-resume execution. + + Feedback submission indicates user intent to continue. + Execution resumes immediately after storing feedback. + + Args: + feedback_type: Type of feedback + content: Feedback content + tool_id: Tool invocation ID + """ + logger.info( + "[HITL-FM] Feedback submitted for tool %s: type=%s, operation=%s", + tool_id, + feedback_type.value, + self.operation_id, + ) + logger.info( + "[HITL-FM] Feedback content (length=%d):\n%s", + len(content), + content[:200] + "..." if len(content) > 200 else content, + ) + + feedback = UserFeedback( + feedback_type=feedback_type, + content=content, + tool_id=tool_id, + timestamp=time.time(), + ) + self.pending_feedback = feedback + self.feedback_queue[tool_id] = feedback + + logger.debug( + "[HITL-FM] Feedback stored - pending_feedback=%s, queue_size=%d", + self.pending_feedback is not None, + len(self.feedback_queue), + ) + + # Emit feedback event to backend + if self.emitter: + self.emitter.emit( + { + "type": "hitl_feedback_submitted", + "feedback_type": feedback_type.value, + "content": content, + "tool_id": tool_id, + "timestamp": feedback.timestamp, + } + ) + + # Store intervention in memory + if self.memory: + self._store_intervention(feedback) + + # Auto-resume execution (user intent to continue) + # Note: Don't clear pending_feedback yet - injection hook needs it + self._pause_event.set() + self.state = HITLState.ACTIVE + self._is_manual_pause = False + logger.info("[HITL-FM] Execution auto-resumed after feedback submission") + + def get_pending_feedback(self, tool_id: str) -> Optional[UserFeedback]: + """Get pending feedback for tool. + + Args: + tool_id: Tool invocation ID + + Returns: + UserFeedback if exists, None otherwise + """ + return self.feedback_queue.get(tool_id) + + def is_paused(self) -> bool: + """Check if currently paused.""" + return self.state == HITLState.PAUSED + + def resume(self) -> None: + """Resume execution from paused state.""" + logger.info("[HITL-FM] Resuming execution from paused state") + self.state = HITLState.ACTIVE + self.pending_tool = None + self.pending_feedback = None + + # Signal the pause event to unblock wait_for_feedback() + self._pause_event.set() + self._is_manual_pause = False + + def get_pending_feedback_message(self) -> Optional[str]: + """Get pending feedback formatted as agent message. + + Returns: + Formatted message if feedback pending, None otherwise + """ + if not self.pending_feedback: + logger.debug("[HITL-FM] No pending feedback to retrieve") + return None + + feedback = self.pending_feedback + + message = f"""HUMAN FEEDBACK RECEIVED: + +Type: {feedback.feedback_type.value} +Content: {feedback.content} + +Please incorporate this feedback and adjust your approach accordingly. Continue the security assessment with this guidance in mind.""" + + logger.info( + "[HITL-FM] Formatted pending feedback into message (type=%s, length=%d)", + feedback.feedback_type.value, + len(message), + ) + logger.debug("[HITL-FM] Formatted message preview:\n%s", message[:300]) + + return message + + def clear_pending_feedback(self) -> None: + """Clear pending feedback after it has been injected into agent context.""" + if self.pending_feedback: + logger.info( + "[HITL-FM] Clearing pending feedback after injection (type=%s, tool_id=%s)", + self.pending_feedback.feedback_type.value, + self.pending_feedback.tool_id, + ) + self.pending_feedback = None + else: + logger.warning( + "[HITL-FM] clear_pending_feedback called but no feedback was pending" + ) + + def _store_intervention(self, feedback: UserFeedback) -> None: + """Store intervention in memory and logs. + + Args: + feedback: User feedback to store + """ + try: + if self.memory and self.pending_tool: + intervention_data = { + "category": "hitl_intervention", + "tool_name": self.pending_tool.tool_name, + "tool_id": feedback.tool_id, + "feedback_type": feedback.feedback_type.value, + "feedback_content": feedback.content, + "original_parameters": self.pending_tool.parameters, + "timestamp": feedback.timestamp, + } + + # Store in Mem0 + if hasattr(self.memory, "add"): + self.memory.add( + str(intervention_data), + user_id="cyber_agent", + metadata=intervention_data, + ) + + logger.info( + "Intervention stored in memory for tool %s", feedback.tool_id + ) + + except Exception as e: + logger.warning("Failed to store intervention in memory: %s", e) diff --git a/src/modules/handlers/hitl/hitl_hook_provider.py b/src/modules/handlers/hitl/hitl_hook_provider.py new file mode 100644 index 00000000..a286eec6 --- /dev/null +++ b/src/modules/handlers/hitl/hitl_hook_provider.py @@ -0,0 +1,206 @@ +"""HITL Hook Provider for intercepting tool calls.""" + +import logging +from typing import Optional + +from strands.hooks import ( + BeforeModelCallEvent, + BeforeToolCallEvent, + HookProvider, + HookRegistry, +) + +from .feedback_manager import FeedbackManager + +logger = logging.getLogger(__name__) + + +class HITLHookProvider(HookProvider): + """Hook provider for HITL tool interception.""" + + def __init__( + self, + feedback_manager: FeedbackManager, + auto_pause_on_destructive: bool = True, + auto_pause_on_low_confidence: bool = True, + confidence_threshold: float = 70.0, + ): + """Initialize HITL hook provider. + + Args: + feedback_manager: FeedbackManager instance + auto_pause_on_destructive: Auto-pause before destructive operations + auto_pause_on_low_confidence: Auto-pause on low confidence + confidence_threshold: Confidence threshold for auto-pause (0-100) + """ + self.feedback_manager = feedback_manager + self.auto_pause_on_destructive = auto_pause_on_destructive + self.auto_pause_on_low_confidence = auto_pause_on_low_confidence + self.confidence_threshold = confidence_threshold + + # Track tools that should trigger auto-pause + self.destructive_patterns = [ + "rm ", + "delete ", + "drop ", + "truncate ", + "format ", + "erase ", + ] + + logger.info( + "HITLHookProvider initialized (destructive=%s, low_conf=%s, threshold=%.1f)", + auto_pause_on_destructive, + auto_pause_on_low_confidence, + confidence_threshold, + ) + + def register_hooks(self, registry: HookRegistry, **kwargs) -> None: + """Register hook callbacks. + + Args: + registry: Hook registry from Strands SDK + **kwargs: Additional keyword arguments (unused) + """ + logger.debug("Registering HITL hooks") + registry.add_callback(BeforeToolCallEvent, self._on_before_tool_call) + registry.add_callback(BeforeModelCallEvent, self._check_manual_pause) + logger.info("HITL hooks registered successfully (tool + model invocation)") + + def _on_before_tool_call(self, event: BeforeToolCallEvent) -> None: + """Handle before tool call event. + + Args: + event: BeforeToolCallEvent from Strands SDK + + Raises: + RuntimeError: If user rejects the tool execution + """ + tool_use = event.tool_use + tool_name = tool_use.get("name", "unknown") + tool_id = tool_use.get("toolUseId", tool_use.get("id", "unknown")) + tool_input = tool_use.get("input", {}) + + logger.debug("HITL hook intercepted tool: %s (id=%s)", tool_name, tool_id) + + # Determine if we should pause + should_pause, reason = self._should_pause_for_tool(tool_name, tool_input) + + if should_pause: + logger.info( + "Auto-pause triggered for tool %s (reason=%s)", + tool_name, + reason, + ) + + # Request pause through feedback manager + self.feedback_manager.request_pause( + tool_name=tool_name, + tool_id=tool_id, + parameters=tool_input, + confidence=None, # TODO: Extract confidence from event if available + reason=reason, + ) + + # Block execution until feedback received or timeout + feedback_received = self.feedback_manager.wait_for_feedback() + if not feedback_received: + logger.warning( + "Timeout expired waiting for feedback on tool %s - auto-resuming", + tool_name, + ) + return + + # Check if user rejected the operation + feedback = self.feedback_manager.get_pending_feedback(tool_id) + if feedback and feedback.feedback_type.value == "rejection": + logger.info( + "Tool %s rejected by user - preventing execution", + tool_name, + ) + # Raise exception to prevent tool from executing + # The agent will see the rejection feedback at next model invocation + raise RuntimeError( + f"Tool execution cancelled by user (tool={tool_name}, reason=rejection)" + ) + + def _check_manual_pause(self, event: BeforeModelCallEvent) -> None: + """Check for manual pause before each model invocation. + + This ensures manual pause requests (via [i] key) are honored even when + the agent is not calling tools. + + Args: + event: BeforeModelCallEvent from Strands SDK + """ + if self.feedback_manager.is_paused(): + logger.info("[HITL-Hook] Manual pause detected - waiting for feedback") + + # Block until feedback received or timeout + feedback_received = self.feedback_manager.wait_for_feedback() + + if feedback_received: + logger.info("[HITL-Hook] Feedback received - resuming execution") + else: + logger.warning( + "[HITL-Hook] Manual pause timeout expired - auto-resuming" + ) + + def _should_pause_for_tool( + self, + tool_name: str, + tool_input: dict, + ) -> tuple[bool, Optional[str]]: + """Determine if tool should trigger auto-pause. + + Args: + tool_name: Name of the tool + tool_input: Tool input parameters + + Returns: + Tuple of (should_pause, reason) + """ + # Check for destructive operations + if self.auto_pause_on_destructive: + if self._is_destructive_operation(tool_name, tool_input): + return True, "destructive_operation" + + # Check for low confidence (if confidence scoring is available) + if self.auto_pause_on_low_confidence: + # TODO: Extract confidence from tool invocation metadata + # For now, we don't have access to model confidence scores + pass + + return False, None + + def _is_destructive_operation(self, tool_name: str, tool_input: dict) -> bool: + """Check if operation is potentially destructive. + + Args: + tool_name: Name of the tool + tool_input: Tool input parameters + + Returns: + True if potentially destructive, False otherwise + """ + # Check shell commands + if tool_name == "shell": + command = tool_input.get("command", "") + if isinstance(command, str): + command_lower = command.lower() + for pattern in self.destructive_patterns: + if pattern in command_lower: + logger.debug( + "Destructive pattern '%s' found in command: %s", + pattern, + command[:50], + ) + return True + + # Check editor operations (file deletions) + if tool_name == "editor": + operation = tool_input.get("operation", "") + if operation in ["delete", "remove"]: + return True + + return False diff --git a/src/modules/handlers/hitl/types.py b/src/modules/handlers/hitl/types.py new file mode 100644 index 00000000..e101eda8 --- /dev/null +++ b/src/modules/handlers/hitl/types.py @@ -0,0 +1,42 @@ +"""Type definitions for HITL system.""" + +from dataclasses import dataclass +from enum import Enum +from typing import Any, Dict, Optional + + +class HITLState(Enum): + """HITL workflow states.""" + + ACTIVE = "active" # Normal execution + PAUSED = "paused" # Execution paused, waiting for feedback + + +class FeedbackType(Enum): + """Types of user feedback.""" + + CORRECTION = "correction" # Modify tool parameters + SUGGESTION = "suggestion" # Propose alternative approach + APPROVAL = "approval" # Approve as-is + REJECTION = "rejection" # Reject and abort + + +@dataclass +class ToolInvocation: + """Tool invocation details for HITL review.""" + + tool_name: str + tool_id: str + parameters: Dict[str, Any] + confidence: Optional[float] = None + reason: Optional[str] = None + + +@dataclass +class UserFeedback: + """User feedback on tool invocation.""" + + feedback_type: FeedbackType + content: str + tool_id: str + timestamp: float diff --git a/src/modules/handlers/prompt_rebuild_hook.py b/src/modules/handlers/prompt_rebuild_hook.py index d6704d52..c484e22a 100644 --- a/src/modules/handlers/prompt_rebuild_hook.py +++ b/src/modules/handlers/prompt_rebuild_hook.py @@ -17,13 +17,23 @@ from pathlib import Path from typing import Any, Dict, Optional -from strands.experimental.hooks.events import BeforeModelInvocationEvent -from strands.hooks import HookProvider, HookRegistry +from strands.hooks import BeforeModelCallEvent, HookProvider, HookRegistry from modules.config.system.logger import get_logger logger = get_logger("Handlers.PromptRebuildHook") +# Import HITL logger for debugging hook interactions +try: + from modules.handlers.hitl.hitl_logger import log_hitl + + HITL_LOGGING_AVAILABLE = True +except ImportError: + HITL_LOGGING_AVAILABLE = False + + def log_hitl(*args, **kwargs): + pass + class PromptRebuildHook(HookProvider): """Trigger-based prompt rebuilding (not every step). @@ -106,16 +116,16 @@ def __init__( operation_id, ) - def register_hooks(self, registry: HookRegistry): - """Register BeforeModelInvocationEvent callback.""" - registry.add_callback(BeforeModelInvocationEvent, self.check_if_rebuild_needed) - logger.debug("PromptRebuildHook registered for BeforeModelInvocationEvent") + def register_hooks(self, registry: HookRegistry, **kwargs: Any): + """Register BeforeModelCallEvent callback.""" + registry.add_callback(BeforeModelCallEvent, self.check_if_rebuild_needed) + logger.debug("PromptRebuildHook registered for BeforeModelCallEvent") - def check_if_rebuild_needed(self, event: BeforeModelInvocationEvent): + def check_if_rebuild_needed(self, event: BeforeModelCallEvent): """Check triggers and rebuild prompt if needed. Args: - event: BeforeModelInvocationEvent from Strands SDK + event: BeforeModelCallEvent from Strands SDK """ current_step = self.callback_handler.current_step @@ -128,6 +138,16 @@ def check_if_rebuild_needed(self, event: BeforeModelInvocationEvent): ) if not should_rebuild: + logger.debug( + "Prompt rebuild skipped at step %d (last rebuild: step %d)", + current_step, + self.last_rebuild_step, + ) + log_hitl( + "PromptRebuild", + f"Rebuild skipped at step {current_step} (interval not reached)", + "DEBUG", + ) return # Keep using existing prompt logger.info( @@ -135,6 +155,11 @@ def check_if_rebuild_needed(self, event: BeforeModelInvocationEvent): current_step, self.last_rebuild_step, ) + log_hitl( + "PromptRebuild", + f"⚠️ Prompt rebuild TRIGGERED at step {current_step} (last: {self.last_rebuild_step})", + "WARNING", + ) # Rebuild prompt with fresh context try: @@ -188,7 +213,17 @@ def check_if_rebuild_needed(self, event: BeforeModelInvocationEvent): ) # Update agent's system prompt + old_prompt_len = ( + len(event.agent.system_prompt) if event.agent.system_prompt else 0 + ) event.agent.system_prompt = new_prompt + new_prompt_len = len(new_prompt) + + log_hitl( + "PromptRebuild", + f"✓ Prompt completely rebuilt: {old_prompt_len} → {new_prompt_len} chars", + "WARNING", + ) # Update tracking self.last_rebuild_step = current_step diff --git a/src/modules/handlers/react/react_bridge_handler.py b/src/modules/handlers/react/react_bridge_handler.py index 583c9b5c..3acaeb57 100644 --- a/src/modules/handlers/react/react_bridge_handler.py +++ b/src/modules/handlers/react/react_bridge_handler.py @@ -225,7 +225,13 @@ def __call__(self, **kwargs): When in swarm operation context, callbacks are attributed to the currently active swarm agent for proper visibility in the UI. """ - # Minimal logging for production + # Log callback invocations for HITL debugging + callback_type = kwargs.get("event", {}).get("type", "unknown") + logger.debug( + "[HITL-RBH] Callback invoked: type=%s, step=%d", + callback_type, + self.current_step, + ) # Transform SDK events to UI events self._transform_sdk_event(kwargs) diff --git a/src/modules/handlers/report_generator.py b/src/modules/handlers/report_generator.py index baeb3c92..7c186e48 100644 --- a/src/modules/handlers/report_generator.py +++ b/src/modules/handlers/report_generator.py @@ -21,6 +21,7 @@ logger = get_logger("Handlers.ReportGenerator") + def generate_security_report( target: str, objective: str, diff --git a/src/modules/interfaces/react/src/App.tsx b/src/modules/interfaces/react/src/App.tsx index ae78f00d..b6895c22 100644 --- a/src/modules/interfaces/react/src/App.tsx +++ b/src/modules/interfaces/react/src/App.tsx @@ -79,7 +79,7 @@ const AppContent: React.FC = ({ const currentTheme = themeManager.getCurrentTheme(); // Consolidated state management - const { state: appState, actions } = useApplicationState(); + const { state: appState, actions, dispatch } = useApplicationState(); // Command parser service const commandParser = React.useMemo(() => new InputParser(), []); @@ -387,6 +387,7 @@ const AppContent: React.FC = ({ const mainAppViewProps = React.useMemo(() => ({ appState, actions, + dispatch, currentTheme, operationHistoryEntries: operationManager.operationHistoryEntries, assessmentFlowState: operationManager.assessmentFlowState, @@ -404,6 +405,7 @@ const AppContent: React.FC = ({ }), [ appState, actions, + dispatch, currentTheme, operationManager.operationHistoryEntries, operationManager.assessmentFlowState, diff --git a/src/modules/interfaces/react/src/components/HITLInterventionPanel.tsx b/src/modules/interfaces/react/src/components/HITLInterventionPanel.tsx new file mode 100644 index 00000000..32afaaf1 --- /dev/null +++ b/src/modules/interfaces/react/src/components/HITLInterventionPanel.tsx @@ -0,0 +1,222 @@ +/** + * HITLInterventionPanel - Human-in-the-Loop Intervention Interface + * + * Interactive panel for reviewing and providing feedback on tool executions + * before they run. Enables human oversight of potentially destructive operations. + */ + +import React, { useState } from 'react'; +import { Box, Text, useInput } from 'ink'; +import TextInput from 'ink-text-input'; + +interface HITLInterventionPanelProps { + /** Tool name being reviewed */ + toolName: string; + /** Unique tool invocation ID */ + toolId: string; + /** Tool parameters to review */ + parameters: Record; + /** Reason for pause (e.g., "destructive_operation") */ + reason?: string; + /** Confidence score if available (0-100) */ + confidence?: number; + /** Timeout in seconds for pause duration */ + timeoutSeconds?: number; + /** Whether panel is currently active */ + isActive: boolean; + /** Callback for submitting feedback */ + onSubmitFeedback: (feedbackType: string, content: string) => void; +} + +/** + * HITLInterventionPanel Component + */ +export const HITLInterventionPanel: React.FC = ({ + toolName, + toolId, + parameters, + reason, + confidence, + timeoutSeconds, + isActive, + onSubmitFeedback, +}) => { + const isManualIntervention = toolName === 'manual_intervention'; + const [feedbackText, setFeedbackText] = useState(''); + const [mode, setMode] = useState<'review' | 'feedback'>('review'); + + // Keyboard handler for destructive operations + useInput((input, key) => { + // Only handle keyboard for non-manual interventions when active + if (!isActive || isManualIntervention) return; + + // Switch to feedback mode when [c] pressed for destructive operations + if (mode === 'review' && input === 'c') { + setMode('feedback'); + return; + } + + // Escape to go back to review mode + if (mode === 'feedback' && key.escape) { + setMode('review'); + setFeedbackText(''); + return; + } + }); + + // Show idle state when HITL is enabled but no intervention needed + if (!isActive) { + return ( + + + ✓ HITL: Active - monitoring operations (press [i] for manual intervention) + + + ); + } + + // Format parameters for display + const formatParameters = (params: Record): string => { + try { + return JSON.stringify(params, null, 2); + } catch { + return String(params); + } + }; + + // Manual Intervention - Direct text input + if (isManualIntervention) { + return ( + + + + 💬 Provide Feedback to Agent + + {timeoutSeconds && ( + + ⏱ Timeout: {timeoutSeconds}s + + )} + + + + + > + { + if (value.trim()) { + onSubmitFeedback('suggestion', value); + setFeedbackText(''); + } + }} + /> + + + + + Press [Esc] to cancel + + + ); + } + + // Auto-pause (Destructive Operation) - Show tool details with approval options + if (!isManualIntervention && mode === 'review') { + const hasParameters = parameters && Object.keys(parameters).length > 0; + + return ( + + + + ⚠️ Potentially destructive operation - review required + + {timeoutSeconds && ( + + ⏱ Timeout: {timeoutSeconds}s + + )} + + + + + Tool: {toolName} + + + + {reason && ( + + + Reason: {reason} + + + )} + + {hasParameters && ( + + Parameters: + {formatParameters(parameters)} + + )} + + + Options: + [a] Approve - proceed with operation + [r] Reject - cancel this operation + [c] Correction - provide modified parameters + [Esc] Cancel and resume + + + + Press a key to choose... + + + ); + } + + // Feedback input mode (for destructive operations when user presses [c]) + if (!isManualIntervention && mode === 'feedback') { + return ( + + + + 💬 Provide Correction + + + + + + Tool: {toolName} + + + + + Enter modified parameters or instructions: + + > + { + if (value.trim()) { + onSubmitFeedback('correction', value); + setFeedbackText(''); + setMode('review'); + } + }} + /> + + + + + Press Esc to cancel + + + ); + } + + return null; +}; diff --git a/src/modules/interfaces/react/src/components/MainAppView.tsx b/src/modules/interfaces/react/src/components/MainAppView.tsx index 3d76584f..f45fe375 100644 --- a/src/modules/interfaces/react/src/components/MainAppView.tsx +++ b/src/modules/interfaces/react/src/components/MainAppView.tsx @@ -15,15 +15,18 @@ import { Footer } from './Footer.js'; import { UnifiedInputPrompt } from './UnifiedInputPrompt.js'; import { Terminal } from './Terminal.js'; import { ModalRegistry } from './ModalRegistry.js'; +import { HITLInterventionPanel } from './HITLInterventionPanel.js'; +import { submitFeedback } from '../utils/hitlCommands.js'; // Types -import { ApplicationState } from '../hooks/useApplicationState.js'; +import { ApplicationState, ActionType } from '../hooks/useApplicationState.js'; import { OperationHistoryEntry } from '../hooks/useOperationManager.js'; import { ModalType } from '../hooks/useModalManager.js'; interface MainAppViewProps { appState: ApplicationState; actions: any; // Application state actions + dispatch?: any; // Application state dispatch function currentTheme: any; // Theme configuration object operationHistoryEntries: OperationHistoryEntry[]; assessmentFlowState: any; // Assessment flow state object @@ -47,6 +50,7 @@ interface MainAppViewProps { export const MainAppView: React.FC = ({ appState, actions, + dispatch, currentTheme, operationHistoryEntries, assessmentFlowState, @@ -276,6 +280,10 @@ export const MainAppView: React.FC = ({ onMetricsUpdate={handleMetricsUpdate} animationsEnabled={isAutoScrollEnabled && activeModal === ModalType.NONE} cleanupRef={terminalCleanupRef} + dispatch={dispatch} + hitlEnabled={appState.hitlEnabled} + hitlPendingTool={appState.hitlPendingTool} + hitlInterpretation={appState.hitlInterpretation} /> ) )} @@ -283,7 +291,27 @@ export const MainAppView: React.FC = ({ {/* INPUT & FOOTER AREA: Static at the bottom */} - {!hideInput && activeModal === ModalType.NONE && (!showOperationStream || appState.userHandoffActive) && ( + {/* HITL Intervention Panel - Pinned above footer */} + {appState.hitlEnabled && activeModal === ModalType.NONE && ( + { + if (appState.hitlPendingTool) { + submitFeedback(feedbackType as any, content, appState.hitlPendingTool.toolId); + } + }} + /> + )} + + {!hideInput && activeModal === ModalType.NONE && + !appState.hitlPendingTool && + (!showOperationStream || appState.userHandoffActive) && ( = ({ userHandoffActive={appState.userHandoffActive} /> )} - + {/* Spacer above footer when streaming to preserve breathing room */} {showOperationStream && ( diff --git a/src/modules/interfaces/react/src/components/StreamDisplay.tsx b/src/modules/interfaces/react/src/components/StreamDisplay.tsx index 9efb6ecf..335f43cf 100644 --- a/src/modules/interfaces/react/src/components/StreamDisplay.tsx +++ b/src/modules/interfaces/react/src/components/StreamDisplay.tsx @@ -83,7 +83,10 @@ export type AdditionalStreamEvent = | { type: 'batch'; id?: string; events: DisplayStreamEvent[]; [key: string]: any } | { type: 'tool_output'; tool: string; status?: string; output?: any; [key: string]: any } | { type: 'operation_init'; operation_id?: string; target?: string; objective?: string; memory?: any; [key: string]: any } - | { type: 'report_paths'; operation_id?: string; target?: string; outputDir?: string; reportPath?: string; logPath?: string; memoryPath?: string; [key: string]: any }; + | { type: 'report_paths'; operation_id?: string; target?: string; outputDir?: string; reportPath?: string; logPath?: string; memoryPath?: string; [key: string]: any } + | { type: 'hitl_pause_requested'; tool_name?: string; tool_id?: string; parameters?: any; reason?: string; confidence?: number; [key: string]: any } + | { type: 'hitl_feedback_submitted'; feedback_type?: string; content?: string; tool_id?: string; [key: string]: any } + | { type: 'hitl_resume'; tool_id?: string; [key: string]: any }; // Combined event type supporting both SDK-aligned and additional events export type DisplayStreamEvent = StreamEvent | AdditionalStreamEvent; @@ -2066,6 +2069,9 @@ const method = latestInput.method || 'GET'; {('observability' in event) && ( Observability: {event.observability ? 'enabled' : 'disabled'} )} + {('hitl_enabled' in event) && event.hitl_enabled && ( + HITL: enabled - human feedback available + )} {('tools_available' in event && event.tools_available) ? ( Available Tools: {event.tools_available} ) : null} @@ -2104,6 +2110,68 @@ const method = latestInput.method || 'GET'; ); } + case 'hitl_pause_requested': { + const toolName = 'tool_name' in event ? String(event.tool_name) : 'unknown'; + const reason = 'reason' in event ? String(event.reason) : undefined; + const confidence = 'confidence' in event && typeof event.confidence === 'number' ? event.confidence : undefined; + + return ( + + ⚠️ HITL: Tool execution paused for review + + Tool: {toolName} + + {reason && ( + + Reason: {reason} + + )} + {confidence !== undefined && ( + + Confidence: {confidence}% + + )} + + ); + } + + case 'hitl_feedback_submitted': { + const feedbackType = 'feedback_type' in event ? String(event.feedback_type) : 'unknown'; + const content = 'content' in event ? String(event.content) : ''; + const preview = content.length > 80 ? content.substring(0, 80) + '...' : content; + + return ( + + ✓ Feedback Submitted to Agent + + Type: + {feedbackType} + + {preview && ( + + Content: + {preview} + + )} + + → Agent will process in next step + + + ); + } + + case 'hitl_agent_interpretation': { + const interpretation = 'interpretation' in event ? String(event.interpretation) : ''; + return ( + + ✓ Agent Interpretation: + + {interpretation} + + + ); + } + case 'specialist_progress': { const status = event.status || 'Processing'; const gate = event.gate; @@ -2174,6 +2242,14 @@ const method = latestInput.method || 'GET'; ); } + case 'hitl_resume': { + return ( + + ✓ Execution resumed + + ); + } + default: return null; } diff --git a/src/modules/interfaces/react/src/components/Terminal.tsx b/src/modules/interfaces/react/src/components/Terminal.tsx index 80a1de01..d4af4f86 100644 --- a/src/modules/interfaces/react/src/components/Terminal.tsx +++ b/src/modules/interfaces/react/src/components/Terminal.tsx @@ -8,7 +8,7 @@ */ import React, { useState, useEffect, useRef, useCallback } from 'react'; -import { Box, Text } from 'ink'; +import { Box, Text, useInput } from 'ink'; import { StreamDisplay, StaticStreamDisplay, DisplayStreamEvent } from './StreamDisplay.js'; import { ExecutionService } from '../services/ExecutionService.js'; import { themeManager } from '../themes/theme-manager.js'; @@ -20,6 +20,8 @@ import { ByteBudgetRingBuffer } from '../utils/ByteBudgetRingBuffer.js'; import { DISPLAY_LIMITS } from '../constants/config.js'; import { useTerminalSize } from '../hooks/useTerminalSize.js'; import { calculateAvailableHeight } from '../utils/layoutConstants.js'; +import { submitFeedback, requestManualIntervention, setExecutionServiceForHITL } from '../utils/hitlCommands.js'; +import { ActionType } from '../hooks/useApplicationState.js'; // Exported helper: build a trimmed report preview to avoid storing huge content in memory export const buildTrimmedReportContent = (raw: string): string => { @@ -50,6 +52,10 @@ interface TerminalProps { onMetricsUpdate?: (metrics: { tokens?: number; cost?: number; duration: string; memoryOps: number; evidence: number }) => void; animationsEnabled?: boolean; cleanupRef?: React.MutableRefObject<(() => void) | null>; + dispatch?: (action: any) => void; + hitlEnabled?: boolean; + hitlPendingTool?: any; + hitlInterpretation?: any; } export const Terminal: React.FC = React.memo(({ @@ -60,11 +66,67 @@ export const Terminal: React.FC = React.memo(({ onEvent, onMetricsUpdate, animationsEnabled = true, - cleanupRef + cleanupRef, + dispatch, + hitlEnabled = false, + hitlPendingTool, + hitlInterpretation }) => { // Use production-grade terminal size hook with resize handling const { availableWidth, availableHeight, columns } = useTerminalSize(); const terminalWidth = propsTerminalWidth || availableWidth; + + // Set execution service for HITL commands + useEffect(() => { + setExecutionServiceForHITL(executionService); + return () => setExecutionServiceForHITL(null); + }, [executionService]); + + // Manual intervention handler - [i] key (always active when HITL enabled) + useInput((input, key) => { + if (!hitlEnabled) return; + if (input?.toLowerCase() === 'i' && !hitlPendingTool && !hitlInterpretation) { + requestManualIntervention(); + } + }); + + // HITL keyboard handler + useInput((input, key) => { + if (!hitlPendingTool && !hitlInterpretation) return; + + const isManualIntervention = hitlPendingTool?.toolName === 'manual_intervention'; + + // Escape key - cancel intervention and resume + if (key.escape) { + if (dispatch) { + dispatch({ type: ActionType.CLEAR_HITL_STATE }); + } + return; + } + + // Manual intervention - only handle Esc, all other keys handled by panel + if (isManualIntervention) { + // TextInput in panel handles all input + return; + } + + // Destructive operation review mode: a/c/r keys + if (hitlPendingTool && !hitlInterpretation) { + if (input === 'a') { + submitFeedback('approval', 'Approved - continuing as planned', hitlPendingTool.toolId); + if (dispatch) { + dispatch({ type: ActionType.CLEAR_HITL_STATE }); + } + } else if (input === 'r') { + submitFeedback('rejection', 'Rejected - stopping execution', hitlPendingTool.toolId); + if (dispatch) { + dispatch({ type: ActionType.CLEAR_HITL_STATE }); + } + } + // [c] is handled by HITLInterventionPanel to switch to feedback mode + } + }); + // Test marker utility for diagnosing spinner/timer behavior const emitTestMarker = (msg: string) => { try { @@ -577,6 +639,43 @@ export const Terminal: React.FC = React.memo(({ }; switch (event.type) { + case 'hitl_pause_requested': + // Update HITL state when tool execution is paused + if (dispatch) { + dispatch({ + type: ActionType.SET_HITL_PENDING_TOOL, + payload: { + toolName: event.tool_name || 'unknown', + toolId: event.tool_id || '', + parameters: event.parameters || {}, + reason: event.reason, + confidence: event.confidence, + timeoutSeconds: event.timeout_seconds + } + }); + // Set userHandoffActive to prevent ESC from terminating the operation + dispatch({ + type: ActionType.SET_USER_HANDOFF, + payload: true + }); + } + results.push(event as DisplayStreamEvent); + break; + + case 'hitl_feedback_submitted': + case 'hitl_resume': + // Clear HITL state when feedback is submitted or execution resumes + if (dispatch) { + dispatch({ type: ActionType.CLEAR_HITL_STATE }); + // Clear userHandoffActive to restore normal ESC behavior + dispatch({ + type: ActionType.SET_USER_HANDOFF, + payload: false + }); + } + results.push(event as DisplayStreamEvent); + break; + case 'operation_init': // Reset dedup sets at operation start perToolOutputSeenRef.current.clear(); @@ -591,6 +690,10 @@ export const Terminal: React.FC = React.memo(({ if (typeof event.target === 'string') { targetRef.current = event.target; } + // Set HITL enabled status from operation init + if (dispatch && 'hitl_enabled' in event && event.hitl_enabled === true) { + dispatch({ type: ActionType.SET_HITL_ENABLED, payload: true }); + } // Reset counters at operation start stepCounterRef.current = 0; lastPushedTypeRef.current = null; diff --git a/src/modules/interfaces/react/src/hooks/useApplicationState.ts b/src/modules/interfaces/react/src/hooks/useApplicationState.ts index 7a85359d..3411212f 100644 --- a/src/modules/interfaces/react/src/hooks/useApplicationState.ts +++ b/src/modules/interfaces/react/src/hooks/useApplicationState.ts @@ -51,6 +51,22 @@ export interface ApplicationState { // Terminal dimensions terminalDisplayHeight: number; terminalDisplayWidth: number; + + // HITL (Human-in-the-Loop) state + hitlEnabled: boolean; + hitlPendingTool: { + toolName: string; + toolId: string; + parameters: Record; + reason?: string; + confidence?: number; + timeoutSeconds?: number; + } | null; + hitlInterpretation: { + toolId: string; + text: string; + modifiedParameters: Record; + } | null; } // Action types @@ -88,6 +104,12 @@ export enum ActionType { // Context usage UPDATE_CONTEXT_USAGE = 'UPDATE_CONTEXT_USAGE', + + // HITL actions + SET_HITL_ENABLED = 'SET_HITL_ENABLED', + SET_HITL_PENDING_TOOL = 'SET_HITL_PENDING_TOOL', + SET_HITL_INTERPRETATION = 'SET_HITL_INTERPRETATION', + CLEAR_HITL_STATE = 'CLEAR_HITL_STATE', } // Action definitions @@ -111,7 +133,11 @@ type Action = | { type: ActionType.INCREMENT_ERROR_COUNT } | { type: ActionType.RESET_ERROR_COUNT } | { type: ActionType.SET_DOCKER_AVAILABLE; payload: boolean } - | { type: ActionType.UPDATE_CONTEXT_USAGE; payload: number }; + | { type: ActionType.UPDATE_CONTEXT_USAGE; payload: number } + | { type: ActionType.SET_HITL_ENABLED; payload: boolean } + | { type: ActionType.SET_HITL_PENDING_TOOL; payload: { toolName: string; toolId: string; parameters: Record; reason?: string; confidence?: number; timeoutSeconds?: number } | null } + | { type: ActionType.SET_HITL_INTERPRETATION; payload: { toolId: string; text: string; modifiedParameters: Record } | null } + | { type: ActionType.CLEAR_HITL_STATE }; // Reducer function function applicationReducer(state: ApplicationState, action: Action): ApplicationState { @@ -199,7 +225,19 @@ function applicationReducer(state: ApplicationState, action: Action): Applicatio case ActionType.UPDATE_CONTEXT_USAGE: return { ...state, contextUsage: action.payload }; - + + case ActionType.SET_HITL_ENABLED: + return { ...state, hitlEnabled: action.payload }; + + case ActionType.SET_HITL_PENDING_TOOL: + return { ...state, hitlPendingTool: action.payload }; + + case ActionType.SET_HITL_INTERPRETATION: + return { ...state, hitlInterpretation: action.payload }; + + case ActionType.CLEAR_HITL_STATE: + return { ...state, hitlPendingTool: null, hitlInterpretation: null }; + default: return state; } @@ -233,6 +271,9 @@ function getInitialState(): ApplicationState { recentTargets: [], terminalDisplayHeight: process.stdout.rows || 24, terminalDisplayWidth: process.stdout.columns || 80, + hitlEnabled: false, + hitlPendingTool: null, + hitlInterpretation: null, }; } diff --git a/src/modules/interfaces/react/src/services/PythonExecutionService.ts b/src/modules/interfaces/react/src/services/PythonExecutionService.ts index 57b94a8f..36a2c7a0 100644 --- a/src/modules/interfaces/react/src/services/PythonExecutionService.ts +++ b/src/modules/interfaces/react/src/services/PythonExecutionService.ts @@ -352,16 +352,33 @@ export class PythonExecutionService extends EventEmitter { * Send user input to the active Python process (newline-terminated) */ public async sendUserInput(input: string): Promise { + const timestamp = new Date().toISOString(); + this.logger.debug(`[${timestamp}] [HITL-ExecService] sendUserInput called with ${input.length} chars`); + this.logger.debug(`[${timestamp}] [HITL-ExecService] Input preview: ${input.substring(0, 200)}`); + if (!this.activeProcess || !this.activeProcess.stdin) { + this.logger.error(`[${timestamp}] [HITL-ExecService] ERROR: No active Python process`); + this.logger.error(`[${timestamp}] [HITL-ExecService] activeProcess=${!!this.activeProcess}, stdin=${!!this.activeProcess?.stdin}`); throw new Error('No active Python process to receive input'); } + + this.logger.debug(`[${timestamp}] [HITL-ExecService] Writing to stdin...`); + return new Promise((resolve, reject) => { try { - this.activeProcess!.stdin!.write(input.endsWith('\n') ? input : input + '\n', (err?: Error) => { - if (err) return reject(err); + const finalInput = input.endsWith('\n') ? input : input + '\n'; + this.logger.debug(`[${timestamp}] [HITL-ExecService] Final input length: ${finalInput.length} chars`); + + this.activeProcess!.stdin!.write(finalInput, (err?: Error) => { + if (err) { + this.logger.error(`[${timestamp}] [HITL-ExecService] ERROR: Stdin write failed:`, err); + return reject(err); + } + this.logger.info(`[${timestamp}] [HITL-ExecService] ✓ Stdin write successful`); resolve(); }); } catch (err) { + this.logger.error(`[${timestamp}] [HITL-ExecService] ERROR: Exception during write:`, err); reject(err as Error); } }); diff --git a/src/modules/interfaces/react/src/types/events.ts b/src/modules/interfaces/react/src/types/events.ts index 0523f27a..c861833f 100644 --- a/src/modules/interfaces/react/src/types/events.ts +++ b/src/modules/interfaces/react/src/types/events.ts @@ -197,7 +197,17 @@ export enum EventType { AGENT_MESSAGE = 'agent_message', /** Security agent completed */ AGENT_COMPLETE = 'agent_complete', - + + // ============================================================================= + // HITL (Human-in-the-Loop) EVENTS - User intervention and feedback + // ============================================================================= + /** Tool execution paused for human review */ + HITL_PAUSE_REQUESTED = 'hitl_pause_requested', + /** User feedback submitted for pending tool */ + HITL_FEEDBACK_SUBMITTED = 'hitl_feedback_submitted', + /** Execution resumed after feedback processing */ + HITL_RESUME = 'hitl_resume', + } // ============================================================================= @@ -435,6 +445,26 @@ export interface AgentEvent extends BaseEvent { result?: any; } +// HITL (Human-in-the-Loop) events +export interface HITLEvent extends BaseEvent { + type: EventType.HITL_PAUSE_REQUESTED | EventType.HITL_FEEDBACK_SUBMITTED | EventType.HITL_RESUME; + /** Tool name being reviewed */ + tool_name?: string; + /** Unique tool invocation ID */ + tool_id?: string; + /** Tool parameters under review */ + parameters?: Record; + /** Confidence score (0-100) */ + confidence?: number; + /** Reason for pause (e.g., "destructive_operation") */ + reason?: string; + /** Feedback type (correction, suggestion, approval, rejection) */ + feedback_type?: string; + /** Feedback content from user */ + content?: string; + /** Timeout in seconds for pause duration */ + timeout_seconds?: number; +} // Python event system events export interface PythonSystemEvent extends BaseEvent { @@ -509,6 +539,7 @@ export type StreamEvent = | SystemEvent | ConnectionEvent | AgentEvent + | HITLEvent | PythonSystemEvent | ReportContentEvent | TerminationReasonEvent diff --git a/src/modules/interfaces/react/src/utils/hitlCommands.ts b/src/modules/interfaces/react/src/utils/hitlCommands.ts new file mode 100644 index 00000000..a3e07e07 --- /dev/null +++ b/src/modules/interfaces/react/src/utils/hitlCommands.ts @@ -0,0 +1,83 @@ +/** + * HITL Command Utilities + * + * Helper functions for sending Human-in-the-Loop feedback commands + * to the Python backend via stdin using the __HITL_COMMAND__ protocol. + */ + +import { ExecutionService } from '../services/ExecutionService.js'; + +// Global reference to execution service for HITL commands +let _executionService: ExecutionService | null = null; + +/** + * Set the execution service reference for HITL commands + */ +export const setExecutionServiceForHITL = (service: ExecutionService | null): void => { + _executionService = service; +}; + +/** + * Send a HITL command to the Python process via stdin + * + * Commands are wrapped in __HITL_COMMAND____HITL_COMMAND_END__ + * format for the Python FeedbackInputHandler to parse. + */ +const sendHITLCommand = async (command: Record): Promise => { + const timestamp = new Date().toISOString(); + console.log(`[${timestamp}] [HITL-UI] Preparing to send command:`, JSON.stringify(command, null, 2)); + + try { + const commandJson = JSON.stringify(command); + const formattedCommand = `__HITL_COMMAND__${commandJson}__HITL_COMMAND_END__`; + + console.log(`[${timestamp}] [HITL-UI] Formatted command length: ${formattedCommand.length} chars`); + console.log(`[${timestamp}] [HITL-UI] Formatted command:`, formattedCommand); + + // Send via execution service to Python process stdin + if (_executionService && 'sendUserInput' in _executionService) { + console.log(`[${timestamp}] [HITL-UI] Execution service available, calling sendUserInput`); + await (_executionService as any).sendUserInput(formattedCommand); + console.log(`[${timestamp}] [HITL-UI] sendUserInput completed successfully`); + } else { + console.error(`[${timestamp}] [HITL-UI] ERROR: No execution service available to send command`); + console.error(`[${timestamp}] [HITL-UI] _executionService:`, _executionService); + } + } catch (error) { + console.error(`[${timestamp}] [HITL-UI] ERROR: Failed to send HITL command:`, error); + } +}; + +/** + * Submit user feedback for a paused tool execution + */ +export const submitFeedback = ( + feedbackType: 'correction' | 'suggestion' | 'approval' | 'rejection', + content: string, + toolId: string +): void => { + const timestamp = new Date().toISOString(); + console.log(`[${timestamp}] [HITL-UI] submitFeedback() called:`, { + feedbackType, + contentLength: content.length, + toolId, + contentPreview: content.substring(0, 100) + }); + + sendHITLCommand({ + type: 'submit_feedback', + feedback_type: feedbackType, + content, + tool_id: toolId, + }); +}; + +/** + * Request manual intervention (pause agent for human review) + */ +export const requestManualIntervention = (): void => { + sendHITLCommand({ + type: 'request_pause', + is_manual: true, + }); +}; diff --git a/src/modules/operation_plugins/ctf/tools/__init__.py b/src/modules/operation_plugins/ctf/tools/__init__.py index 093ea540..140d7e50 100644 --- a/src/modules/operation_plugins/ctf/tools/__init__.py +++ b/src/modules/operation_plugins/ctf/tools/__init__.py @@ -1,5 +1,3 @@ -"""CTF-specific tools for generic capture-the-flag web challenges. - -""" +"""CTF-specific tools for generic capture-the-flag web challenges.""" __all__: list[str] = [] diff --git a/src/modules/operation_plugins/general/tools/advanced_payload_coordinator.py b/src/modules/operation_plugins/general/tools/advanced_payload_coordinator.py index 48fb1d16..f525fc4e 100644 --- a/src/modules/operation_plugins/general/tools/advanced_payload_coordinator.py +++ b/src/modules/operation_plugins/general/tools/advanced_payload_coordinator.py @@ -9,7 +9,9 @@ @tool -def advanced_payload_coordinator(target_url: str, test_type: str = "comprehensive", parameters: str = None) -> str: +def advanced_payload_coordinator( + target_url: str, test_type: str = "comprehensive", parameters: str = None +) -> str: """ Coordinates advanced payload testing using specialized external tools. @@ -76,11 +78,15 @@ def advanced_payload_coordinator(target_url: str, test_type: str = "comprehensiv output += "Phase 3: Advanced XSS Payload Testing\\n" output += "-" * 40 + "\\n" - xss_results = _coordinate_xss_testing(target_url, results.get("parameters_discovered", [])) + xss_results = _coordinate_xss_testing( + target_url, results.get("parameters_discovered", []) + ) results["payload_results"].extend(xss_results) xss_vulns = [r for r in xss_results if r.get("vulnerable", False)] - output += f"XSS testing completed: {len(xss_vulns)} potential vulnerabilities\\n" + output += ( + f"XSS testing completed: {len(xss_vulns)} potential vulnerabilities\\n" + ) for vuln in xss_vulns[:3]: output += f" • {vuln['parameter']}: {vuln['payload_type']}\\n" output += "\\n" @@ -104,10 +110,14 @@ def advanced_payload_coordinator(target_url: str, test_type: str = "comprehensiv output += "Phase 5: Advanced Injection Testing\\n" output += "-" * 40 + "\\n" - injection_results = _coordinate_injection_testing(target_url, results.get("parameters_discovered", [])) + injection_results = _coordinate_injection_testing( + target_url, results.get("parameters_discovered", []) + ) results["payload_results"].extend(injection_results) - injection_vulns = [r for r in injection_results if r.get("vulnerable", False)] + injection_vulns = [ + r for r in injection_results if r.get("vulnerable", False) + ] output += f"Injection testing: {len(injection_vulns)} potential vulnerabilities\\n" for vuln in injection_vulns[:3]: output += f" • {vuln['injection_type']}: {vuln['parameter']}\\n" @@ -120,9 +130,7 @@ def advanced_payload_coordinator(target_url: str, test_type: str = "comprehensiv intelligence = _analyze_payload_intelligence(results["payload_results"]) results["intelligence"] = intelligence - output += ( - f"Total vulnerabilities: {len([r for r in results['payload_results'] if r.get('vulnerable', False)])}\\n" - ) + output += f"Total vulnerabilities: {len([r for r in results['payload_results'] if r.get('vulnerable', False)])}\\n" output += f"Attack vectors identified: {len(intelligence['attack_vectors'])}\\n" output += f"Bypass techniques: {len(intelligence['bypass_techniques'])}\\n" @@ -177,10 +185,16 @@ def _setup_payload_tools() -> Dict[str, Any]: tools_status["failed"].append(tool_name) else: # Python tool - try pip install - pip_names = {"arjun": "arjun", "corsy": "corsy", "paramspider": "ParamSpider"} + pip_names = { + "arjun": "arjun", + "corsy": "corsy", + "paramspider": "ParamSpider", + } if tool_name in pip_names: install_cmd = ["pip3", "install", pip_names[tool_name]] - result = subprocess.run(install_cmd, capture_output=True, timeout=120) + result = subprocess.run( + install_cmd, capture_output=True, timeout=120 + ) if result.returncode == 0: tools_status["tools"].append(tool_name) else: @@ -193,7 +207,9 @@ def _setup_payload_tools() -> Dict[str, Any]: return tools_status -def _advanced_parameter_discovery(target_url: str, provided_params: str = None) -> List[str]: +def _advanced_parameter_discovery( + target_url: str, provided_params: str = None +) -> List[str]: """Advanced parameter discovery using multiple techniques""" discovered_params = set() @@ -302,7 +318,9 @@ def _advanced_parameter_discovery(target_url: str, provided_params: str = None) return sorted(list(discovered_params)) -def _coordinate_xss_testing(target_url: str, parameters: List[str]) -> List[Dict[str, Any]]: +def _coordinate_xss_testing( + target_url: str, parameters: List[str] +) -> List[Dict[str, Any]]: """Coordinate XSS testing using advanced payloads and techniques""" xss_results = [] @@ -345,7 +363,12 @@ def _coordinate_xss_testing(target_url: str, parameters: List[str]) -> List[Dict ) else: xss_results.append( - {"parameter": param, "vulnerable": False, "payload_type": "XSS tested", "tool": "dalfox"} + { + "parameter": param, + "vulnerable": False, + "payload_type": "XSS tested", + "tool": "dalfox", + } ) except Exception: @@ -360,7 +383,7 @@ def _coordinate_xss_testing(target_url: str, parameters: List[str]) -> List[Dict "", # Context-aware payloads "'\\\">", # Breaking out of attributes - "\\\";alert(1);//", # Breaking out of JavaScript strings + '\\";alert(1);//', # Breaking out of JavaScript strings "