Captures the current system global state for debugging context. This function gathers critical system metrics, application state (locks, queues), and configuration details to help diagnose issues like client disconnects or headless mode failures. Args: req_id: Request
(
req_id: str = "unknown", error_name: str = "unknown"
)
| 155 | |
| 156 | |
| 157 | async def capture_system_context( |
| 158 | req_id: str = "unknown", error_name: str = "unknown" |
| 159 | ) -> Dict[str, Any]: |
| 160 | """ |
| 161 | Captures the current system global state for debugging context. |
| 162 | |
| 163 | This function gathers critical system metrics, application state (locks, queues), |
| 164 | and configuration details to help diagnose issues like client disconnects |
| 165 | or headless mode failures. |
| 166 | |
| 167 | Args: |
| 168 | req_id: Request ID associated with the context |
| 169 | error_name: Name of the error triggering the capture |
| 170 | |
| 171 | Returns: |
| 172 | Dict containing comprehensive system context |
| 173 | """ |
| 174 | # Import server locally to avoid circular dependency |
| 175 | import platform |
| 176 | import sys |
| 177 | |
| 178 | from api_utils.server_state import state |
| 179 | |
| 180 | iso_time, texas_time = get_texas_timestamp() |
| 181 | |
| 182 | # Helper to safely get queue size |
| 183 | def get_qsize(q: Optional[Union[Queue[Any], SupportsSizeQuery]]) -> int: |
| 184 | try: |
| 185 | return q.qsize() if q else -1 |
| 186 | except (NotImplementedError, AttributeError): |
| 187 | return -1 # Some queue types (like multiprocessing.Queue on macOS) might not support qsize |
| 188 | |
| 189 | # Helper to safely check lock state |
| 190 | def is_locked(lock: Optional[Union[Lock, SupportsLockQuery]]) -> bool: |
| 191 | try: |
| 192 | return lock.locked() if lock else False |
| 193 | except AttributeError: |
| 194 | return False |
| 195 | |
| 196 | # Helper to sanitize proxy settings |
| 197 | def _sanitize_proxy(settings: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: |
| 198 | if not settings: |
| 199 | return None |
| 200 | safe_settings = settings.copy() |
| 201 | server_val = safe_settings.get("server") |
| 202 | if isinstance(server_val, str) and "@" in server_val: |
| 203 | # Redact credentials in http://user:pass@host format |
| 204 | try: |
| 205 | parts = server_val.split("@") |
| 206 | scheme_creds = parts[0].split("://") |
| 207 | if len(scheme_creds) == 2: |
| 208 | safe_settings["server"] = f"{scheme_creds[0]}://***:***@{parts[1]}" |
| 209 | except Exception: |
| 210 | safe_settings["server"] = "REDACTED" |
| 211 | return safe_settings |
| 212 | |
| 213 | context: Dict[str, Any] = { |
| 214 | "meta": { |