feat: Middleware Sync Integration for RunAgent

[RadeenXALNW]

Overview

This PR introduces comprehensive middleware synchronization capabilities to the RunAgent system, enabling seamless integration between local agent execution and remote middleware services. The enhancement provides real-time synchronization of agent lifecycle events, invocation tracking, and basic logging across distributed deployments.

Key Features Implemented

Middleware Sync Service - Implemented a robust middleware synchronization service that handles agent startup sync, invocation lifecycle tracking, and log aggregation. The service automatically detects API key configuration and gracefully falls back to local-only mode when middleware is unavailable.

Enhanced Invocation Tracking - Extended the database service with comprehensive invocation tracking capabilities, including start/completion timestamps, execution metrics, and error handling. Added new local AgentInvocation table with detailed metadata capture and middleware correlation support.

Local Server Integration - Enhanced the FastAPI local server with automatic middleware sync on startup, real-time invocation synchronization, and comprehensive logging integration. The server now creates middleware projects automatically when agents start and synchronizes all execution events.

Advanced Logging System - Implemented a sophisticated database log handler that buffers logs intelligently and syncs to middleware when agents are properly registered. The system includes execution context tracking and smart buffering based on log levels.

WebSocket Streaming Sync - Extended WebSocket streaming capabilities with full middleware synchronization, enabling real-time tracking of streaming invocations across multiple chunks with comprehensive metadata preservation.

CLI Integration - Enhanced CLI commands with middleware sync status reporting and sync configuration management. Added new local-sync command for managing middleware connectivity and sync preferences with detailed status reporting.

Technical Implementation

The implementation uses proper async/await patterns for middleware communication while maintaining backward compatibility. The entire system gracefully handles middleware unavailability, continuing local operations without interruption. Created comprehensive invocation tracking that captures complete execution metadata with millisecond-precision timing. Implemented smart caching for API limit validation and robust error handling throughout the sync pipeline.

Database Enhancements

Added new AgentInvocation table with UUID-based invocation tracking, input/output data serialization, execution timing, status lifecycle management, entrypoint tag tracking, SDK type identification, and client metadata preservation with optimized indexes.

Graceful Degradation and Error Handling

Automatic Fallback - When middleware sync fails due to network issues, invalid API keys, or server unavailability, the system automatically falls back to local-only mode without disrupting agent operations. All invocation tracking, logging, and core functionality continues to work normally using the local database.

Clear Status Reporting - The system provides clear status messages during startup indicating sync status. When sync fails, users see "Agent sync failed - running in local-only mode" with full invocation tracking still enabled locally. Invalid or missing API keys result in graceful degradation with informative error messages.

Robust Operation Modes - The system operates effectively in three modes: full middleware sync (when API key is valid and middleware is available), local-only mode (when sync fails or is disabled), and mixed mode (when some sync operations succeed while others fail). All modes maintain full local functionality and data integrity.

Benefits

Provides unified observability across local development and remote deployments, enabling comprehensive monitoring of agent performance. Enables detailed analytics on agent usage and performance metrics through centralized middleware aggregation. Significantly improves debugging capabilities for complex agent interactions. The enhancement is completely backward compatible, requiring no changes to existing agent code while providing immediate value to teams using middleware services.

[sawradip]

@CodiumAI-Agent /improve

[QodoAI-Agent]

PR Code Suggestions ✨

Latest suggestions up to 63f9240

Category	Suggestion	Impact
Possible issue	Guard middleware start sync with try/except Wrap the middleware sync invocation start in a try/except to prevent middleware failures from prematurely terminating the WebSocket handling. Log any sync errors at a warning level and continue with a None middleware_invocation_id if the sync fails. This ensures robustness against middleware downtime. runagent/sdk/server/socket_utils.py [109-123] if self.middleware_sync and self.middleware_sync.sync_enabled: - middleware_invocation_id = await self.middleware_sync.sync_invocation_start({ - "agent_id": agent_id, - "input_data": { - "input_args": ws_request.input_data.input_args, - "input_kwargs": ws_request.input_data.input_kwargs - }, - "entrypoint_tag": entrypoint_tag, - "sdk_type": "websocket_stream", - "client_info": { - "connection_id": connection_id, - "websocket_action": ws_request.action.value, - "stream_config": ws_request.stream_config - } - }) + try: + middleware_invocation_id = await self.middleware_sync.sync_invocation_start({ + "agent_id": agent_id, + "input_data": { + "input_args": ws_request.input_data.input_args, + "input_kwargs": ws_request.input_data.input_kwargs + }, + "entrypoint_tag": entrypoint_tag, + "sdk_type": "websocket_stream", + "client_info": { + "connection_id": connection_id, + "websocket_action": ws_request.action.value, + "stream_config": ws_request.stream_config + } + }) + except Exception as sync_error: + console.print(f"[yellow]Middleware sync start failed: {sync_error}[/yellow]") + middleware_invocation_id = None Suggestion importance[1-10]: 8 __ Why: Wrapping sync_invocation_start in try/except prevents middleware failures from crashing the handler, improving resilience.	Medium
General	Use create_task instead of run_until_complete Avoid blocking the event loop by replacing run_until_complete with an asynchronous task when possible. Always schedule log syncing via create_task if inside a running loop, and fall back to asyncio.run only when no loop is available. This prevents deadlocks in async contexts. runagent/utils/logs.py [104-113] -if loop.is_running(): - # If we're in an async context, schedule for later - asyncio.create_task( - self.middleware_sync.sync_agent_logs(logs_to_sync) - ) -else: - # Run in the current loop - loop.run_until_complete( - self.middleware_sync.sync_agent_logs(logs_to_sync) - ) +try: + loop = asyncio.get_running_loop() + asyncio.create_task(self.middleware_sync.sync_agent_logs(logs_to_sync)) +except RuntimeError: + asyncio.run(self.middleware_sync.sync_agent_logs(logs_to_sync)) Suggestion importance[1-10]: 6 __ Why: Switching to get_running_loop and always using create_task avoids blocking and simplifies log syncing logic.	Low
	Fix console color tag The color tag in the f-string is missing its leading bracket, so the message won’t render as yellow in the console. Add the missing "[" before "yellow" to restore proper markup. runagent/sdk/deployment/middleware_sync.py [317] -console.print(f"yellow]Could not initialize middleware sync: {e}[/yellow]") +console.print(f"[yellow]Could not initialize middleware sync: {e}[/yellow]") Suggestion importance[1-10]: 4 __ Why: The missing [ before yellow breaks the rich markup in console.print, so adding it restores the intended color rendering.	Low

---

Previous suggestions

Suggestions up to commit 63f9240

Category	Suggestion	Impact
Possible issue	Fix method signature mismatch The signature of _handle_stream_start_with_tracking does not match its invocation and is missing the websocket and request parameters. This mismatch will cause runtime errors when the method is called. Adjust the signature to align with its usage and include all needed parameters. runagent/sdk/server/socket_utils.py [273-278] async def _handle_stream_start_with_tracking( - connection_id: str, + websocket: WebSocket, + request: WebSocketAgentRequest, + connection_id: str, agent_execution_streamer: Callable, invocation_id: str, db_service, - middleware_invocation_id: str = None # NEW: Add middleware invocation ID + middleware_invocation_id: str = None ): Suggestion importance[1-10]: 9 __ Why: The _handle_stream_start_with_tracking signature omits the websocket and request parameters used at its call site, causing a runtime error; aligning them is critical.	High
	Use correct timestamp key for log sync The code references log["timestamp"] but the log objects from get_agent_logs use "created_at" as the timestamp key. Update the field to use created_at so timestamps are correctly serialized. runagent/sdk/deployment/middleware_sync.py [121-139] async def sync_agent_logs(self, logs_data: List[Dict[str, Any]]) -> bool: # CRITICAL: Don't try to sync if not enabled if not self.is_sync_enabled() or not logs_data: return False try: console.print(f"[dim]Syncing {len(logs_data)} logs to middleware...[/dim]") # Prepare logs for middleware middleware_logs = [] for log in logs_data: middleware_log = { "agent_id": log["agent_id"], "log_level": log["log_level"], "message": log["message"], "execution_id": log.get("execution_id"), - "timestamp": log["timestamp"].isoformat() if hasattr(log["timestamp"], 'isoformat') else str(log["timestamp"]), + "timestamp": log["created_at"].isoformat() if hasattr(log["created_at"], 'isoformat') else str(log["created_at"]), "source": "local_server" } middleware_logs.append(middleware_log) Suggestion importance[1-10]: 7 __ Why: The code references log["timestamp"] but get_agent_logs populates "created_at", so switching to log["created_at"] prevents missing or incorrect timestamps during sync.	Medium
	Ensure logs_to_sync is always defined If an exception occurs before logs_to_sync is defined, the except block will reference an undefined variable. Define logs_to_sync before the try block to ensure it always exists. runagent/utils/logs.py [82-125] def _flush_logs_to_middleware(self): """Send buffered logs to middleware""" if not self.log_buffer: return # Double-check sync conditions if not self._should_sync_to_middleware(): print(f"Skipping middleware sync - agent not ready (buffer: {len(self.log_buffer)} logs)") - # Keep logs in buffer for later return + logs_to_sync = [] try: # Create a copy of buffer and clear it immediately logs_to_sync = self.log_buffer.copy() self.log_buffer.clear() print(f"🔍 Flushing {len(logs_to_sync)} logs to middleware for agent_id: {self.agent_id}") ... except Exception as e: print(f"Error flushing logs to middleware: {e}") # Put logs back in buffer for retry self.log_buffer.extend(logs_to_sync) Suggestion importance[1-10]: 5 __ Why: Predefining logs_to_sync avoids a potential UnboundLocalError in the except block, making error handling safer.	Low
General	Guard middleware start sync If sync_invocation_start raises an exception, it will bubble out and disrupt the stream. Wrap the middleware sync call in a try/except to log failures without halting execution. runagent/sdk/server/socket_utils.py [109-123] -if self.middleware_sync and self.middleware_sync.sync_enabled: - middleware_invocation_id = await self.middleware_sync.sync_invocation_start({ - "agent_id": agent_id, - "input_data": { - "input_args": ws_request.input_data.input_args, - "input_kwargs": ws_request.input_data.input_kwargs - }, - "entrypoint_tag": entrypoint_tag, - "sdk_type": "websocket_stream", - "client_info": { - "connection_id": connection_id, - "websocket_action": ws_request.action.value, - "stream_config": ws_request.stream_config - } - }) +if self.middleware_sync and getattr(self.middleware_sync, "sync_enabled", False): + try: + middleware_invocation_id = await self.middleware_sync.sync_invocation_start({ + "agent_id": agent_id, + "input_data": { + "input_args": ws_request.input_data.input_args, + "input_kwargs": ws_request.input_data.input_kwargs + }, + "entrypoint_tag": entrypoint_tag, + "sdk_type": "websocket_stream", + "client_info": { + "connection_id": connection_id, + "websocket_action": ws_request.action.value, + "stream_config": ws_request.stream_config + } + }) + except Exception as sync_start_error: + console.print(f"[yellow]Failed to sync middleware start: {sync_start_error}[/yellow]") Suggestion importance[1-10]: 6 __ Why: Wrapping the sync_invocation_start call in try/except prevents a middleware sync failure from interrupting the streaming logic, improving reliability.	Low
	Remove duplicate shutdown_logging There are two identical shutdown_logging methods defined back-to-back, causing one to override the other. Remove the duplicate definition to avoid confusion and ensure a single implementation is used. runagent/sdk/server/local_server.py [1072-1094] def shutdown_logging(self): """Properly shutdown logging and flush remaining logs""" try: if hasattr(self, 'db_handler'): self.db_handler.close() if hasattr(self, 'agent_logger'): for handler in self.agent_logger.handlers[:]: handler.close() self.agent_logger.removeHandler(handler) except Exception as e: console.print(f"⚠️ [yellow]Error during logging shutdown: {e}[/yellow]") -def shutdown_logging(self): - """Properly shutdown logging and flush remaining logs""" - try: - if hasattr(self, 'db_handler'): - self.db_handler.close() - if hasattr(self, 'agent_logger'): - for handler in self.agent_logger.handlers[:]: - handler.close() - self.agent_logger.removeHandler(handler) - except Exception as e: - console.print(f"⚠️ [yellow]Error during logging shutdown: {e}[/yellow]") - Suggestion importance[1-10]: 4 __ Why: There are two identical shutdown_logging methods back-to-back, which overrides one of them and clutters the class. Removing the duplicate improves maintainability without changing behavior.	Low
	Fix missing bracket in color tag The error message tag is missing an opening bracket, so the color code is malformed. Add the missing "[" before "yellow" to properly format the console output. This ensures the message appears in yellow as intended. runagent/sdk/deployment/middleware_sync.py [308-320] def get_middleware_sync() -> Optional[MiddlewareSyncService]: """Get the global middleware sync instance""" if _global_middleware_sync is None: try: config = SDKConfig() _global_middleware_sync = MiddlewareSyncService(config) except Exception as e: - console.print(f"yellow]Could not initialize middleware sync: {e}[/yellow]") + console.print(f"[yellow]Could not initialize middleware sync: {e}[/yellow]") _global_middleware_sync = None return _global_middleware_sync Suggestion importance[1-10]: 3 __ Why: The malformed f-string missing the "[" before "yellow" breaks the intended color formatting in console.print. Fixing it corrects log output but is a minor stylistic issue.	Low