LLM/pentestagent
1
0
Fork 0
mirror of https://github.com/GH05TCREW/pentestagent.git synced 2026-03-07 22:33:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: log and notify on critical exceptions (mcp manager, tui target persistence/display)

Browse Source
This commit is contained in:
giveen
2026-01-19 10:35:37 -07:00
parent 2c82a30b16
commit a186b62e8a
2 changed files with 50 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
pentestagent/interface/tui.py
View File

@@ -5,6 +5,7 @@ PentestAgent TUI - Terminal User Interface
import asyncio
import re
import textwrap
import logging
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
@@ -1573,16 +1574,28 @@ class PentestAgentTUI(App):
if active:
try:
wm.set_last_target(active, target)
except Exception:
pass
except Exception as e:
logging.getLogger(__name__).exception("Failed to persist last target for workspace %s: %s", active, e)
try:
from pentestagent.interface.notifier import notify
notify("warning", f"Failed to persist last target for workspace {active}: {e}")
except Exception:
logging.getLogger(__name__).exception("Failed to notify operator about target persist error")
except Exception:
pass
logging.getLogger(__name__).exception("Failed to access WorkspaceManager to persist last target")
# Update displayed Target in the UI
try:
self._apply_target_display(target)
except Exception:
pass
except Exception as e:
logging.getLogger(__name__).exception("Failed to apply target display: %s", e)
try:
from pentestagent.interface.notifier import notify
notify("warning", f"Failed to update target display: {e}")
except Exception:
logging.getLogger(__name__).exception("Failed to notify operator about target display error")
# Update the initial ready SystemMessage (if present) so Target appears under Runtime
try:
scroll = self.query_one("#chat-scroll", ScrollableContainer)
@@ -1595,8 +1608,6 @@ class PentestAgentTUI(App):
try:
if "Target:" in child.message_content:
# replace the first Target line
import re
child.message_content = re.sub(
r"(?m)^\s*Target:.*$",
f" Target: {target}",
@@ -1609,10 +1620,23 @@ class PentestAgentTUI(App):
)
try:
child.refresh()
except Exception as e:
logging.getLogger(__name__).exception("Failed to refresh child message after target update: %s", e)
try:
from pentestagent.interface.notifier import notify
notify("warning", f"Failed to refresh UI after target update: {e}")
except Exception:
logging.getLogger(__name__).exception("Failed to notify operator about UI refresh error")
except Exception as e:
# Fallback to append if regex replacement fails, and surface warning
logging.getLogger(__name__).exception("Failed to update SystemMessage target line: %s", e)
try:
from pentestagent.interface.notifier import notify
notify("warning", f"Failed to update target display: {e}")
except Exception:
pass
except Exception:
# Fallback to append if regex replacement fails
logging.getLogger(__name__).exception("Failed to notify operator about target update error")
child.message_content = (
child.message_content + f"\n Target: {target}"
)

29
pentestagent/mcp/manager.py
View File

@@ -78,8 +78,8 @@ class MCPManager:
# Ensure we attempt to clean up vendored servers on process exit
try:
atexit.register(self._atexit_cleanup)
except Exception:
pass
except Exception as e:
logging.getLogger(__name__).exception("Failed to register atexit cleanup: %s", e)
def _find_config(self) -> Path:
for path in self.DEFAULT_CONFIG_PATHS:
@@ -202,8 +202,10 @@ class MCPManager:
try:
stop()
continue
except Exception:
pass
except Exception as e:
logging.getLogger(__name__).exception(
"Error running adapter.stop(): %s", e
)
# Final fallback: kill underlying PID if available
pid = None
@@ -213,13 +215,14 @@ class MCPManager:
if pid:
try:
os.kill(pid, signal.SIGTERM)
except Exception:
except Exception as e:
logging.getLogger(__name__).exception("Failed to SIGTERM pid %s: %s", pid, e)
try:
os.kill(pid, signal.SIGKILL)
except Exception:
pass
except Exception:
pass
except Exception as e2:
logging.getLogger(__name__).exception("Failed to SIGKILL pid %s: %s", pid, e2)
except Exception as e:
logging.getLogger(__name__).exception("Error while attempting synchronous adapter stop: %s", e)
async def _stop_started_adapters_and_disconnect(self) -> None:
# Stop any adapters we started
@@ -233,15 +236,15 @@ class MCPManager:
# run blocking stop in executor
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, stop)
except Exception:
pass
except Exception as e:
logging.getLogger(__name__).exception("Error stopping adapter in async shutdown: %s", e)
self._started_adapters.clear()
# Disconnect any active MCP server connections
try:
await self.disconnect_all()
except Exception:
pass
except Exception as e:
logging.getLogger(__name__).exception("Error during disconnect_all in shutdown: %s", e)
def add_server(
self,