Wire vulnerability amendment callbacks

README.md

@@ -320,30 +320,6 @@ strix auth status             # show the active sign-in
 strix auth logout             # forget the sign-in
 ```
 
-#### Connect your own MCP servers
-
-Strix can connect to Model Context Protocol (MCP) servers you list and expose their tools to the agent during a run. Create `~/.strix/mcp-servers.json` with a JSON list of servers. Each entry is either a local `stdio` server that Strix launches as a subprocess, or a remote `http` server:
-
-```json
-[
-  {
-    "name": "local_fs",
-    "transport": "stdio",
-    "command": "npx",
-    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"]
-  },
-  {
-    "name": "github",
-    "transport": "http",
-    "url": "https://api.githubcopilot.com/mcp/",
-    "auth": { "kind": "bearer", "token": "your-token" },
-    "allowed_tools": ["list_issues"]
-  }
-]
-```
-
-Each server's tools are namespaced by `name` (for example `local_fs.read_file`). Omit `allowed_tools` to expose every tool the server offers, or set it to a list to restrict which tools the agent can call. The file is optional, and a server that fails to connect is skipped without failing the run. You can point Strix at a different file with `STRIX_MCP_CONFIG`.
-
 **Recommended models for best results:**
 
 - [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4`

docs/docs.json

@@ -47,8 +47,7 @@
             "pages": [
               "integrations/github-actions",
               "integrations/ci-cd",
-              "integrations/coding-agents",
-              "integrations/mcp"
+              "integrations/coding-agents"
             ]
           },
           {

docs/integrations/mcp.mdx

@@ -1,113 +0,0 @@
----
-title: "MCP Servers"
-description: "Connect your own MCP servers and expose their tools to the agent"
----
-
-Strix can connect to [Model Context Protocol (MCP)](https://modelcontextprotocol.io) servers you list and expose their tools to the agent during a run. Use this to give the agent extra capabilities — reading files, querying an issue tracker, or any other tool a server offers.
-
-## Setup
-
-Create the file `~/.strix/mcp-servers.json`. It holds a JSON list of the servers you want the agent to reach. Each entry is either a local `stdio` server that Strix launches as a subprocess, or a remote `http` server.
-
-Create the directory if it does not exist, then write the file:
-
-```bash
-mkdir -p ~/.strix
-```
-
-Paste the servers you want into `~/.strix/mcp-servers.json`. The example below shows one of each transport — a local filesystem server over `stdio` and a remote GitHub server over `http` with a bearer token:
-
-```json
-[
-  {
-    "name": "local_fs",
-    "transport": "stdio",
-    "command": "npx",
-    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/project"]
-  },
-  {
-    "name": "github",
-    "transport": "http",
-    "url": "https://api.githubcopilot.com/mcp/",
-    "auth": { "kind": "bearer", "token": "your-token" },
-    "allowed_tools": ["list_issues"]
-  }
-]
-```
-
-Strix reads this file at the start of each run. There is no default file, so no MCP tools are loaded until you create it. Edit `command`, `args`, `url`, and `token` to match your own servers.
-
-## Fields
-
-<ParamField path="name" type="string" required>
-  A short label for the connection. Each server's tools are namespaced by
-  `name` (for example `local_fs.read_file`), so two servers can offer the same
-  tool name without colliding.
-</ParamField>
-
-<ParamField path="transport" type="string">
-  `stdio` for a local subprocess server, or `http` for a remote server.
-</ParamField>
-
-<ParamField path="command" type="string">
-  For `stdio` servers: the executable Strix launches (for example `npx`).
-</ParamField>
-
-<ParamField path="args" type="array">
-  For `stdio` servers: the arguments passed to `command`.
-</ParamField>
-
-<ParamField path="url" type="string">
-  For `http` servers: the server endpoint URL.
-</ParamField>
-
-<ParamField path="auth" type="object">
-  For `http` servers that need a bearer token:
-  `{ "kind": "bearer", "token": "your-token" }`.
-</ParamField>
-
-<ParamField path="allowed_tools" type="array">
-  Restrict which tools the agent can call. Omit it to expose every tool the
-  server offers, or set it to a list of tool names to allow only those.
-</ParamField>
-
-<ParamField path="notes" type="string">
-  Free-text notes for the agent about what this connection is and how you want
-  it used, for example "Staging analytics database, read-only, prefer aggregate
-  queries." When set, the notes are given to the agent at the start of the run
-  as a description of the connection.
-</ParamField>
-
-## Choosing connections per run
-
-By default every connection in the file is used on each run. To narrow it for a
-single run without editing the file, use either flag (both repeatable):
-
-```bash
-strix --mcp-server github -t ...        # use only the named connection(s)
-strix --mcp-exclude staging-db -t ...   # use everything except the named one(s)
-```
-
-`--mcp-server` keeps only the connections you name; `--mcp-exclude` drops the
-ones you name. Connection names must be unique in the file; if two entries share
-a name, the first is kept and the rest are ignored.
-
-## Pointing at a different file
-
-To read the config from another path instead of `~/.strix/mcp-servers.json`, either pass `--mcp-config <path>` on the command line:
-
-```bash
-strix --mcp-config ./mcp-servers.json -t ...
-```
-
-or set the `STRIX_MCP_CONFIG` environment variable to that path. The flag takes precedence when both are given.
-
-## Startup confirmation
-
-When servers are configured, Strix prints a one-line summary at scan startup, for example `MCP: connected 1 server (14 tools): local_fs`, so you can confirm your servers connected.
-
-## Behavior
-
-- The config file is optional. Without it, a run simply gets no MCP tools.
-- A server that fails to connect is skipped and logged, and the run continues without it.
-- A single malformed entry is skipped without blocking the valid ones.

pyproject.toml

@@ -241,8 +241,6 @@ ignore = [
 "tests/test_stream_idle_timeout.py" = ["N802", "SLF001"]
 "tests/test_unknown_tool_recovery.py" = ["N802"]
 "tests/test_report_pdf.py" = ["S105", "S106"]
-# Fake MCP server matches the SDK's MCPServer signature; its args are unused.
-"tests/test_mcp_client.py" = ["S105", "S106", "ARG002"]
 # Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a
 # circular dependency with strix.telemetry / strix.interface.viewer.report_pdf.
 "strix/interface/viewer/server.py" = ["N802", "PLC0415"]

strix/agents/factory.py

@@ -48,6 +48,7 @@ from strix.tools.reporting.tool import (
     create_vulnerability_report,
     get_report,
     list_reports,
+    update_vulnerability_report,
 )
 from strix.tools.respond.tool import respond_to_user
 from strix.tools.thinking.tool import think
@@ -500,6 +501,7 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     delete_note,
     web_search,
     create_vulnerability_report,
+    update_vulnerability_report,
     create_dependency_report,
     list_reports,
     get_report,

strix/agents/prompts/system_prompt.jinja

@@ -214,11 +214,12 @@ VALIDATION REQUIREMENTS:
 - Treat public metadata, internal-looking identifiers, source maps without secrets, and transport/configuration hygiene as observations unless validation proves unauthorized restricted-data access, modification, or service disruption
 - Every non-None Confidentiality, Integrity, or Availability metric must map to explicit evidence in the report; use Scope Changed only for a demonstrated crossing of security authorities
 - Independent verification through subagent
-- Document complete attack chain
+- Document complete attack chains. Amend an existing finding when new evidence increases or decreases its impact on the same asset and root cause
 - Keep going until you find something that matters
-- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
+- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report or update_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
 - Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
-- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent
+- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, do not re-submit the same vulnerability. If new validated evidence shows greater or lower impact for the same finding, use update_vulnerability_report with the returned duplicate_of id. Otherwise, move on to testing other areas. The vulnerability has already been reported
+- DISTINCT ISSUE VS AMENDMENT: File a distinct issue with create_vulnerability_report. Use update_vulnerability_report when new evidence proves greater or lower impact for the same root cause on the same asset. Supply an evidence-backed update_reason and recalculate the rating from cvss_breakdown
 - REVIEWING FILED FINDINGS (orchestrator/root agent): use list_reports to see every vulnerability filed so far in this scan (by any agent, root or child) — metadata-first with per-severity counts — and get_report to read one finding in full by its id. These are read-only orchestration tools: the root agent uses them to track coverage, avoid dispatching work on already-covered ground, assemble the finish_scan executive summary, and reason about attack-chaining across confirmed findings. Leaf/specialist agents should NOT call them — just do your assigned testing and file findings. Each entry shows which agent filed it (agent_name), and your own entries are flagged by_you. list_notes/get_note do the same for notes.
 </execution_guidelines>
 
@@ -372,7 +373,7 @@ CRITICAL RULES:
 - **REALISTIC OUTCOMES** - Some tests find nothing, some validations fail
 - **ONE AGENT = ONE TASK** - Don't let agents do multiple unrelated jobs
 - **SPAWN REACTIVELY** - Create new agents based on what you discover
-- **ONLY REPORTING AGENTS** can use create_vulnerability_report tool
+- **ONLY REPORTING AGENTS** can use create_vulnerability_report or update_vulnerability_report tools
 - **AGENT SPECIALIZATION MANDATORY** - Each agent must be highly specialized; prefer 1–3 skills, up to 5 for complex contexts
 - **NO GENERIC AGENTS** - Avoid creating broad, multi-purpose agents that dilute focus
 

strix/core/runner.py

@@ -51,12 +51,10 @@ from strix.tools.output_store import (
 
 
 if TYPE_CHECKING:
-    from agents.mcp import MCPServer
     from agents.memory import SQLiteSession
     from agents.result import RunResultBase
 
     from strix.runtime.status import StatusSink
-    from strix.tools.mcp import ConnectedMcpServer
 
 
 logger = logging.getLogger(__name__)
@@ -64,33 +62,6 @@ logger = logging.getLogger(__name__)
 StreamEventSink = Callable[[str, Any], None]
 
 
-def _mcp_startup_summary(connections: list[ConnectedMcpServer]) -> str:
-    """One user-facing line summarizing the MCP servers that connected."""
-    server_count = len(connections)
-    tool_count = sum(c.tool_count for c in connections)
-    servers_word = "server" if server_count == 1 else "servers"
-    tools_word = "tool" if tool_count == 1 else "tools"
-    names = ", ".join(c.name for c in connections)
-    return f"MCP: connected {server_count} {servers_word} ({tool_count} {tools_word}): {names}"
-
-
-def _mcp_connection_notes(connections: list[ConnectedMcpServer]) -> str | None:
-    """A block describing the connections the user left notes on, for the agent.
-
-    Only connections with notes are listed, so the note describes the connection
-    once rather than being repeated onto every tool. Returns ``None`` when no
-    connection has notes.
-    """
-    noted = [(c.name, c.notes) for c in connections if c.notes]
-    if not noted:
-        return None
-    lines = "\n".join(f"- `{name}.*` tools: {notes}" for name, notes in noted)
-    return (
-        "The user connected these MCP servers for this run and left notes on how "
-        f"to use each:\n{lines}"
-    )
-
-
 def _merge_root_prompt_context(
     scope_context: dict[str, Any],
     extra_system_prompt_context: dict[str, Any] | None,
@@ -282,7 +253,6 @@ async def run_strix_scan(
     configure_spill_writer(_spill_to_workspace)
 
     sessions_to_close: list[SQLiteSession] = []
-    mcp_servers: list[MCPServer] = []
 
     try:
         targets = scan_config.get("targets") or []
@@ -328,24 +298,6 @@ async def run_strix_scan(
             system_prompt_context=root_context,
         )
 
-        # Connect any MCP servers the user listed in ~/.strix/mcp-servers.json and
-        # register their tools before the agent is built. Fail-open: a missing
-        # config, or a server that will not connect, must never break a run.
-        from strix.tools.mcp import connect_mcp_servers, load_user_mcp_configs
-
-        try:
-            user_mcp_configs = load_user_mcp_configs()
-            if user_mcp_configs:
-                connections = await connect_mcp_servers(user_mcp_configs)
-                mcp_servers = [c.server for c in connections]
-                if connections:
-                    report(_mcp_startup_summary(connections))
-                    notes_block = _mcp_connection_notes(connections)
-                    if notes_block:
-                        root_task = f"{root_task}\n\n{notes_block}"
-        except Exception:
-            logger.exception("Failed to connect user MCP servers; continuing without them")
-
         root_agent = build_strix_agent(
             name="Root Agent",
             skills=skills,
@@ -520,9 +472,6 @@ async def run_strix_scan(
         for s in sessions_to_close:
             with contextlib.suppress(Exception):
                 s.close()
-        for mcp_server in mcp_servers:
-            with contextlib.suppress(Exception):
-                await mcp_server.cleanup()  # type: ignore[no-untyped-call]
         with contextlib.suppress(Exception):
             await coordinator._maybe_snapshot()
         if cleanup_on_exit:

strix/interface/cli.py

@@ -38,6 +38,35 @@ def _resolve_sandbox_image() -> str:
     return image
 
 
+def _configure_report_callbacks(report_state: ReportState, console: Console) -> None:
+    def display_vulnerability(report: dict[str, Any], *, updated: bool = False) -> None:
+        report_id = report.get("id", "unknown")
+
+        vuln_text = format_vulnerability_report(report)
+
+        title = (
+            f"[bold yellow]{report_id.upper()} — UPDATED FINDING"
+            if updated
+            else f"[bold red]{report_id.upper()}"
+        )
+        vuln_panel = Panel(
+            vuln_text,
+            title=title,
+            title_align="left",
+            border_style="yellow" if updated else "red",
+            padding=(1, 2),
+        )
+
+        console.print(vuln_panel)
+        console.print()
+
+    report_state.vulnerability_found_callback = display_vulnerability
+    report_state.vulnerability_updated_callback = lambda report: display_vulnerability(
+        report,
+        updated=True,
+    )
+
+
 async def run_cli(args: Any) -> None:  # noqa: PLR0915
     console = Console()
 
@@ -105,23 +134,7 @@ async def run_cli(args: Any) -> None:  # noqa: PLR0915
     report_state.set_scan_config(scan_config)
     report_state.save_run_data()
 
-    def display_vulnerability(report: dict[str, Any]) -> None:
-        report_id = report.get("id", "unknown")
-
-        vuln_text = format_vulnerability_report(report)
-
-        vuln_panel = Panel(
-            vuln_text,
-            title=f"[bold red]{report_id.upper()}",
-            title_align="left",
-            border_style="red",
-            padding=(1, 2),
-        )
-
-        console.print(vuln_panel)
-        console.print()
-
-    report_state.vulnerability_found_callback = display_vulnerability
+    _configure_report_callbacks(report_state, console)
 
     def cleanup_on_exit() -> None:
         report_state.cleanup()

strix/interface/cli_args.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import argparse
-import os
 import sys
 from pathlib import Path
 
@@ -220,30 +219,6 @@ Examples:
         help="Path to a custom config file (JSON) to use instead of ~/.strix/cli-config.json",
     )
 
-    parser.add_argument(
-        "--mcp-config",
-        type=str,
-        metavar="PATH",
-        help="Path to an MCP servers JSON file to use instead of ~/.strix/mcp-servers.json.",
-    )
-
-    parser.add_argument(
-        "--mcp-server",
-        dest="mcp_server",
-        action="append",
-        metavar="NAME",
-        help="Use only this MCP connection for the run, by its config name "
-        "(repeatable). Every other configured connection is skipped.",
-    )
-
-    parser.add_argument(
-        "--mcp-exclude",
-        dest="mcp_exclude",
-        action="append",
-        metavar="NAME",
-        help="Skip this MCP connection for the run, by its config name (repeatable).",
-    )
-
     parser.add_argument(
         "--max-budget",
         "--max-budget-usd",
@@ -292,20 +267,6 @@ Examples:
     if args.config:
         apply_config_override(validate_config_file(args.config))
 
-    if args.mcp_config:
-        mcp_config_path = Path(args.mcp_config).expanduser()
-        if not mcp_config_path.is_file():
-            parser.error(f"--mcp-config file not found: {args.mcp_config}")
-        # The MCP loader reads this env var as its config-path override, so
-        # setting it here makes the flag win over the default location.
-        os.environ["STRIX_MCP_CONFIG"] = str(mcp_config_path)
-
-    # The MCP loader reads these as its per-run include/exclude selection.
-    if args.mcp_server:
-        os.environ["STRIX_MCP_ONLY"] = ",".join(args.mcp_server)
-    if args.mcp_exclude:
-        os.environ["STRIX_MCP_EXCLUDE"] = ",".join(args.mcp_exclude)
-
     if args.update:
         sys.exit(0 if self_update() else 1)
 

strix/interface/tui/internal/render/registry.go

@@ -22,20 +22,19 @@ func statusIcon(status string) (string, lipgloss.Style) {
 	return "○ Unknown", Dim()
 }
 
-// renderGenericTool ports registry._render_default_tool_widget. It shows the
-// tool name, its arguments, and a status line only. The raw result is
-// deliberately not rendered: a generic/MCP result (e.g. a multi-kilobyte JSON
-// payload from a database query tool) is noise on screen, and the agent narrates
-// what it got in its next message. The full result still lives in the event
-// data, the run log, and the `strix view` viewer.
-func renderGenericTool(name string, args map[string]any, status string) string {
+// renderGenericTool ports registry._render_default_tool_widget.
+func renderGenericTool(name string, args map[string]any, result any, status string) string {
 	var b strings.Builder
 	b.WriteString(Dim().Render("→ Using tool ") + Bold(Blue).Render(name) + "\n")
 	for _, k := range SortedKeys(args) {
 		b.WriteString("  " + Dim().Render(k) + ": " + StringValue(args[k]) + "\n")
 	}
-	icon, style := statusIcon(status)
-	b.WriteString(style.Render(icon))
+	if (status == "completed" || status == "failed" || status == "error") && result != nil {
+		b.WriteString(lipgloss.NewStyle().Bold(true).Render("Result: ") + StringValue(result))
+	} else {
+		icon, style := statusIcon(status)
+		b.WriteString(style.Render(icon))
+	}
 	return b.String()
 }
 
@@ -88,7 +87,7 @@ func Tool(data map[string]any) string {
 	case "list_requests", "view_request", "repeat_request", "list_sitemap", "view_sitemap_entry", "scope_rules":
 		return renderProxyTool(name, args, result, status)
 	}
-	return renderGenericTool(name, args, status)
+	return renderGenericTool(name, args, result, status)
 }
 
 // ---------------------------------------------------------------------------

strix/interface/tui/internal/render/render_test.go

@@ -203,7 +203,7 @@ func TestToolDispatchCoversKnownTools(t *testing.T) {
 		{
 			"unknown tool falls back to generic",
 			tool("brand_new_tool", map[string]any{"alpha": "1"}, "done", "completed"),
-			[]string{"brand_new_tool", "alpha", "Done"},
+			[]string{"brand_new_tool", "alpha", "Result:", "done"},
 		},
 	}
 
@@ -214,18 +214,6 @@ func TestToolDispatchCoversKnownTools(t *testing.T) {
 	}
 }
 
-func TestGenericToolOmitsRawResult(t *testing.T) {
-	// The generic/MCP renderer shows tool name, args, and a status line only —
-	// never the raw result payload.
-	long := strings.Repeat("x", 5000)
-	out := ansi.Strip(Tool(tool("db_query", map[string]any{"query": "select 1"}, long, "completed")))
-
-	requireContains(t, out, "db_query", "query", "Done")
-	if strings.Contains(out, "Result:") || strings.Contains(out, strings.Repeat("x", 20)) {
-		t.Fatalf("generic result body must not be rendered:\n%s", out)
-	}
-}
-
 func TestCollapseToolShellPreviewAndExpand(t *testing.T) {
 	lines := make([]string, 16)
 	for i := range lines {

strix/interface/tui/runtime.py

@@ -102,6 +102,9 @@ class GoTuiRuntime:
         self.report_state.vulnerability_found_callback = lambda _report: (
             self.controller.notify_changed()
         )
+        self.report_state.vulnerability_updated_callback = lambda _report: (
+            self.controller.notify_changed()
+        )
         self.controller.notify_changed()
 
     async def start_from_setup(self, verify: bool = True) -> None:

strix/report/state.py

@@ -149,6 +149,7 @@ class ReportState:
 
         self.caido_url: str | None = None
         self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None
+        self.vulnerability_updated_callback: Callable[[dict[str, Any]], None] | None = None
 
         self._sarif_repo_ctx: dict[str, Any] | None = None
         self._sarif_repo_ctx_ready: bool = False
@@ -315,6 +316,53 @@ class ReportState:
         self.save_run_data()
         return report_id
 
+    def update_vulnerability_report(
+        self,
+        report_id: str,
+        update_reason: str,
+        **updates: Any,
+    ) -> dict[str, Any] | None:
+        """Amend a filed vulnerability report and persist the updated state."""
+        report = next(
+            (item for item in self.vulnerability_reports if item.get("id") == report_id),
+            None,
+        )
+        if report is None:
+            return None
+
+        timestamp = datetime.now(UTC).isoformat()
+        previous_severity = report.get("severity")
+        previous_cvss = report.get("cvss")
+        changed_fields: list[str] = []
+        for field, value in updates.items():
+            if report.get(field) != value:
+                changed_fields.append(field)
+            report[field] = value
+
+        history_entry: dict[str, Any] = {
+            "timestamp": timestamp,
+            "update_reason": update_reason.strip(),
+            "fields_changed": changed_fields,
+        }
+        if ("severity" in updates and updates.get("severity") != previous_severity) or (
+            "cvss" in updates and updates.get("cvss") != previous_cvss
+        ):
+            history_entry["previous_severity"] = previous_severity
+            history_entry["previous_cvss_score"] = previous_cvss
+
+        history = report.setdefault("update_history", [])
+        if not isinstance(history, list):
+            history = []
+            report["update_history"] = history
+        history.append(history_entry)
+        report["updated_at"] = timestamp
+
+        self._saved_vuln_ids.discard(report_id)
+        if self.vulnerability_updated_callback:
+            self.vulnerability_updated_callback(report)
+        self.save_run_data()
+        return report
+
     def get_existing_vulnerabilities(self) -> list[dict[str, Any]]:
         return list(self.vulnerability_reports)
 

strix/report/writer.py

@@ -197,6 +197,8 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
         f"**Severity:** {report.get('severity', 'unknown').upper()}",
         f"**Found:** {report.get('timestamp', 'unknown')}",
     ]
+    if report.get("updated_at"):
+        lines.append(f"**Updated:** {report['updated_at']}")
 
     dep_meta = report.get("dependency_metadata") or {}
     metadata: list[tuple[str, Any]] = [
@@ -304,4 +306,23 @@ def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PL
         lines.append(str(report["assumptions"]))
         lines.append("")
 
+    update_history = report.get("update_history")
+    if isinstance(update_history, list) and update_history:
+        lines.append("## Amendment History\n")
+        for entry in update_history:
+            if not isinstance(entry, dict):
+                continue
+            timestamp = entry.get("timestamp", "unknown")
+            reason = entry.get("update_reason", "")
+            fields = ", ".join(str(field) for field in entry.get("fields_changed", []))
+            lines.append(f"- **{timestamp}:** {reason}")
+            lines.append(f"  Changed fields: {fields or 'none'}")
+            if "previous_severity" in entry or "previous_cvss_score" in entry:
+                lines.append(
+                    "  Previous rating: "
+                    f"{str(entry.get('previous_severity', 'unknown')).upper()} "
+                    f"(CVSS {entry.get('previous_cvss_score', 'unknown')})"
+                )
+        lines.append("")
+
     return "\n".join(lines)

strix/tools/finish/tool.py

@@ -133,14 +133,15 @@ async def finish_scan(
        combination. You may rule out combinations you can confidently
        call unrelated — note why instead of padding chains. Any
        validated chain must already be filed via
-       ``create_vulnerability_report`` — a demonstrated end-to-end chain
-       is a PoC-backed vulnerability, so it uses that tool even when one
-       link is a dependency CVE (the standalone CVE stays in its own
-       ``create_dependency_report``) — and surfaced prominently in
+       ``create_vulnerability_report`` — or update the existing finding with
+       ``update_vulnerability_report`` when the chain amplifies that finding
+       on the same asset and root cause. A demonstrated new chain is a
+       PoC-backed vulnerability, so file it even when one link is a
+       dependency CVE. Keep the standalone CVE in its own
+       ``create_dependency_report``. Surface the result prominently in
        ``executive_summary`` / ``technical_analysis``. Finding no real
-       chain after a serious attempt is acceptable; skipping the
-       chaining reasoning, or ignoring a plausibly-related combination,
-       is not.
+       chain after a serious attempt is acceptable. Skipping the chaining
+       reasoning, or ignoring a plausibly-related combination, is not.
 
     **Calling this multiple times overwrites the previous report.**
     Make the single call comprehensive.

strix/tools/mcp/__init__.py

@@ -1,21 +0,0 @@
-"""Generic MCP client: connect MCP servers and expose their tools."""
-
-from __future__ import annotations
-
-from strix.tools.mcp.client import ConnectedMcpServer, connect_mcp_servers
-from strix.tools.mcp.config import (
-    BearerAuth,
-    McpAuth,
-    McpConnectionConfig,
-)
-from strix.tools.mcp.loader import load_user_mcp_configs
-
-
-__all__ = [
-    "BearerAuth",
-    "ConnectedMcpServer",
-    "McpAuth",
-    "McpConnectionConfig",
-    "connect_mcp_servers",
-    "load_user_mcp_configs",
-]

strix/tools/mcp/client.py

@@ -1,348 +0,0 @@
-"""Connect to MCP servers and expose their tools to the agent.
-
-Given one :class:`McpConnectionConfig` per server, :func:`connect_mcp_servers`
-lists each server's tools, keeps the ones on the connection's allowlist (or all
-of them when none is set), prefixes each with the connection name so servers do
-not collide, and registers them through the agent factory. The factory applies
-output bounding, per-call timeouts, and structured errors to every registered
-tool, so this layer does not reimplement them.
-
-A server that cannot connect, or a tool set that cannot be registered, is logged
-and skipped, so one bad connection never fails the run.
-"""
-
-from __future__ import annotations
-
-import contextlib
-import json
-import logging
-from typing import TYPE_CHECKING, Any, NamedTuple, cast
-
-from agents.exceptions import ModelBehaviorError
-from agents.mcp import (
-    MCPServer,
-    MCPServerStdio,
-    MCPServerStdioParams,
-    MCPServerStreamableHttp,
-    MCPServerStreamableHttpParams,
-    MCPUtil,
-    create_static_tool_filter,
-)
-
-from strix.agents.factory import register_agent_tools
-
-
-if TYPE_CHECKING:
-    from collections.abc import Callable
-
-    from agents.tool import FunctionTool, Tool
-    from mcp.types import Tool as MCPTool
-
-    from strix.tools.mcp.config import McpConnectionConfig
-
-    # Runs on each tool's structured result before it reaches the agent. Called
-    # ``result_transform(namespaced_tool_name, structured_result)`` and its return
-    # value becomes the tool's output. ``structured_result`` is the parsed
-    # ``CallToolResult`` as a dict (not a serialized string), so the transform can
-    # project or drop individual fields.
-    ResultTransform = Callable[[str, Any], Any]
-
-
-logger = logging.getLogger(__name__)
-
-
-class ConnectedMcpServer(NamedTuple):
-    """One successfully connected MCP server and how many tools it registered.
-
-    ``server`` is kept so the caller can clean it up when the run ends;
-    ``name`` and ``tool_count`` let the caller show the user a startup summary;
-    ``notes`` carries the connection's optional free-text description so the
-    caller can surface it to the agent as context about the connection.
-    """
-
-    server: MCPServer
-    name: str
-    tool_count: int
-    notes: str | None = None
-
-
-def _auth_headers(config: McpConnectionConfig) -> dict[str, str]:
-    """Build the per-server request headers from the connection's auth."""
-    auth = config.auth
-    if auth is None:
-        return {}
-    return {"Authorization": f"Bearer {auth.token}"}
-
-
-def _build_server(config: McpConnectionConfig) -> MCPServer:
-    """Construct (but do not connect) the SDK server for one connection.
-
-    When ``allowed_tools`` is a list the static filter means the server will not
-    even list tools outside it; :func:`_register_server_tools` re-applies the
-    same allowlist as the authoritative gate on what gets registered. When it is
-    ``None`` no filter is applied and every listed tool is registered.
-    """
-    tool_filter = (
-        create_static_tool_filter(allowed_tool_names=config.allowed_tools)
-        if config.allowed_tools is not None
-        else None
-    )
-
-    if config.transport == "stdio":
-        stdio_params: MCPServerStdioParams = {
-            "command": cast("str", config.command),
-            "args": config.args,
-            "env": config.env,
-        }
-        return MCPServerStdio(
-            params=stdio_params,
-            name=config.name,
-            tool_filter=tool_filter,
-            cache_tools_list=True,
-        )
-
-    http_params: MCPServerStreamableHttpParams = {
-        "url": cast("str", config.url),
-        "headers": _auth_headers(config),
-    }
-    return MCPServerStreamableHttp(
-        params=http_params,
-        name=config.name,
-        tool_filter=tool_filter,
-        cache_tools_list=True,
-    )
-
-
-def _build_tool(
-    config: McpConnectionConfig,
-    server: MCPServer,
-    mcp_tool: MCPTool,
-    result_transform: ResultTransform | None,
-) -> FunctionTool:
-    """Build one namespaced FunctionTool from a listed MCP tool.
-
-    The SDK builds the tool (so name override, input schema, approval policy,
-    error-as-result handling, and tool-origin metadata are unchanged). With a
-    ``result_transform`` we route the underlying MCP call through
-    :func:`_install_result_transform` so the transform sees the structured result
-    and decides the tool's output. Without one (the stock path), we still route
-    the call, through :func:`_install_error_status_capture`, so an errored result
-    reads as failed in the TUI while the agent's content is unchanged.
-    """
-    namespaced_name = f"{config.name}.{mcp_tool.name}"
-    tool = MCPUtil.to_function_tool(
-        mcp_tool,
-        server,
-        convert_schemas_to_strict=False,
-        tool_name_override=namespaced_name,
-    )
-    if result_transform is not None:
-        _install_result_transform(tool, server, mcp_tool.name, namespaced_name, result_transform)
-    else:
-        _install_error_status_capture(tool, server, mcp_tool.name, namespaced_name)
-    return tool
-
-
-def _install_result_transform(
-    tool: FunctionTool,
-    server: MCPServer,
-    base_tool_name: str,
-    namespaced_name: str,
-    result_transform: ResultTransform,
-) -> None:
-    """Route a tool's MCP call through ``result_transform``, innermost.
-
-    ``MCPUtil.to_function_tool`` serializes the result inside its own invoke, so
-    the structured result cannot be intercepted through it. Instead we call
-    ``server.call_tool`` ourselves, hand the parsed :class:`CallToolResult` to the
-    transform, and return the transform's output as the tool result.
-
-    This runs INSIDE the tool's invoke. The agent factory wraps a registered
-    tool's ``on_invoke_tool`` with output bounding, disk spill, and tracing at
-    agent-build time, which is OUTSIDE this invoke, so the transform is genuinely
-    the innermost step: nothing sees the raw result before the transform does.
-
-    ``to_function_tool`` wraps the real invoke in the SDK's failure-handling
-    invoker, which stores the inner coroutine on ``_invoke_tool_impl`` and calls
-    it inside its try/except. Swapping that inner impl keeps the SDK's
-    error-as-result handling and all tool metadata while inserting the transform.
-    If the SDK ever renames that attribute we fail loudly rather than silently
-    skip the transform.
-    """
-
-    async def _invoke(_ctx: Any, input_json: str) -> Any:
-        parsed: Any = json.loads(input_json) if input_json else {}
-        if not isinstance(parsed, dict):
-            raise ModelBehaviorError(
-                f"Invalid JSON input for tool {namespaced_name}: expected a JSON object"
-            )
-        args = cast("dict[str, Any]", parsed)
-        result = await server.call_tool(base_tool_name, args)
-        structured_result = result.model_dump(mode="json")
-        return result_transform(namespaced_name, structured_result)
-
-    _replace_tool_invoke(tool, _invoke)
-
-
-def _replace_tool_invoke(tool: FunctionTool, invoke: Callable[[Any, str], Any]) -> None:
-    """Swap a FunctionTool's inner invoke, failing loudly if the SDK shape changed.
-
-    ``to_function_tool`` wraps the real invoke in the SDK's failure-handling
-    invoker, which stores the inner coroutine on ``_invoke_tool_impl`` and calls
-    it inside its own try/except. Swapping that inner impl keeps the SDK's
-    error-as-result handling and every piece of tool metadata intact. It is a
-    plain object with the coroutine as an attribute, not a function, so we treat
-    it as untyped to swap it. If the SDK ever renames that attribute we raise
-    rather than silently leave the swap un-applied.
-    """
-    invoker = cast("Any", tool.on_invoke_tool)
-    if not hasattr(invoker, "_invoke_tool_impl"):
-        raise RuntimeError(
-            "agents SDK FunctionTool invoker shape changed: cannot swap the tool "
-            "invoke without risking it being silently skipped."
-        )
-    invoker._invoke_tool_impl = invoke
-
-
-def _mcp_result_to_tool_output(server: MCPServer, result: Any) -> Any:
-    """Serialize a ``CallToolResult`` to a tool output, mirroring the agents SDK.
-
-    This reproduces the serialization in ``agents.mcp.util.MCPUtil.invoke_mcp_tool``
-    (structured-content JSON when the server asks for it, otherwise text/image
-    content blocks, unwrapping a single block). Because the stock path now routes
-    its own call, this is what makes the agent see byte-identical content to what
-    the SDK would have produced on its own.
-    """
-    if getattr(server, "use_structured_content", False) and result.structuredContent:
-        return json.dumps(result.structuredContent)
-
-    outputs: list[dict[str, Any]] = []
-    for item in result.content:
-        if item.type == "text":
-            outputs.append({"type": "text", "text": item.text})
-        elif item.type == "image":
-            outputs.append(
-                {"type": "image", "image_url": f"data:{item.mimeType};base64,{item.data}"}
-            )
-        else:
-            outputs.append({"type": "text", "text": str(item.model_dump(mode="json"))})
-    if len(outputs) == 1:
-        return outputs[0]
-    return outputs
-
-
-def _install_error_status_capture(
-    tool: FunctionTool,
-    server: MCPServer,
-    base_tool_name: str,
-    namespaced_name: str,
-) -> None:
-    """Make an errored MCP result read as failed in the TUI, agent content unchanged.
-
-    The stock SDK invoke returns only the text/image tool output and drops the
-    ``CallToolResult.isError`` flag, so the TUI cannot tell an errored MCP call
-    (which it renders as a green "done") from a successful one. We route the call
-    the same way :func:`_install_result_transform` does, read ``isError`` off the
-    full result, and on an error tag the returned output dict with
-    ``success: False``.
-
-    That tag reaches the human-facing status but not the agent. The SDK stores the
-    raw return value on the run item's ``output`` (which the TUI reads to derive a
-    tool's status), but hands the agent the value re-projected through its
-    ToolOutput schema, which keeps only the known ``type``/``text`` fields and
-    drops the extra ``success`` key. So the status flips to failed while the agent
-    still receives exactly the same error content it does today. Non-error calls
-    return the stock output unchanged and keep rendering as done.
-    """
-
-    async def _invoke(_ctx: Any, input_json: str) -> Any:
-        parsed: Any = json.loads(input_json) if input_json else {}
-        if not isinstance(parsed, dict):
-            raise ModelBehaviorError(
-                f"Invalid JSON input for tool {namespaced_name}: expected a JSON object"
-            )
-        args = cast("dict[str, Any]", parsed)
-        result = await server.call_tool(base_tool_name, args)
-        tool_output = _mcp_result_to_tool_output(server, result)
-        if getattr(result, "isError", False) and isinstance(tool_output, dict):
-            return {**tool_output, "success": False}
-        return tool_output
-
-    _replace_tool_invoke(tool, _invoke)
-
-
-async def _register_server_tools(
-    config: McpConnectionConfig,
-    server: MCPServer,
-    result_transform: ResultTransform | None = None,
-) -> list[Tool]:
-    """List a connected server's tools, prefix + filter them, and register them.
-
-    ``allowed_tools`` of ``None`` registers every listed tool; a list restricts
-    to exactly those names.
-    """
-    allowed = config.allowed_tools
-    mcp_tools = await server.list_tools()
-
-    tools: list[Tool] = [
-        _build_tool(config, server, mcp_tool, result_transform)
-        for mcp_tool in mcp_tools
-        if allowed is None or mcp_tool.name in allowed
-    ]
-
-    register_agent_tools(*tools)
-    return tools
-
-
-async def connect_mcp_servers(
-    configs: list[McpConnectionConfig],
-    result_transform: ResultTransform | None = None,
-) -> list[ConnectedMcpServer]:
-    """Connect to each MCP server and register its tools.
-
-    When ``result_transform`` is given, every registered tool routes its result
-    through it before the result reaches the agent (see
-    :func:`_install_result_transform`). When it is ``None`` the tools behave
-    exactly as the SDK builds them.
-
-    Returns one :class:`ConnectedMcpServer` per server that connected, carrying
-    the SDK server (so the caller can clean it up when the run ends) plus the
-    server name and how many tools it registered (so the caller can show the
-    user a startup summary). Connections that fail are skipped rather than
-    raised.
-    """
-    connected: list[ConnectedMcpServer] = []
-    for config in configs:
-        server: MCPServer | None = None
-        try:
-            server = _build_server(config)
-            await server.connect()  # type: ignore[no-untyped-call]
-            tools = await _register_server_tools(config, server, result_transform)
-        except Exception:
-            logger.exception("Skipping MCP connection %r", config.name)
-            if server is not None:
-                with contextlib.suppress(Exception):
-                    await server.cleanup()  # type: ignore[no-untyped-call]
-            continue
-        except BaseException:
-            # A cancellation (or other non-Exception failure) mid-connect must not
-            # orphan MCP subprocesses or HTTP sessions. Clean up the server being
-            # connected and every server already connected, then re-raise so the
-            # caller still stops. The runner only receives the list on a clean
-            # return, so on an abnormal exit this function owns the cleanup.
-            if server is not None:
-                with contextlib.suppress(Exception):
-                    await server.cleanup()  # type: ignore[no-untyped-call]
-            for established in connected:
-                with contextlib.suppress(Exception):
-                    await established.server.cleanup()  # type: ignore[no-untyped-call]
-            raise
-
-        logger.info("Connected MCP server %r (%d tools)", config.name, len(tools))
-        connected.append(
-            ConnectedMcpServer(
-                server=server, name=config.name, tool_count=len(tools), notes=config.notes
-            )
-        )
-
-    return connected

strix/tools/mcp/config.py

@@ -1,73 +0,0 @@
-"""The connection-config contract for the MCP client.
-
-Describes one MCP server the client can connect to: its transport, endpoint or
-launch command, optional auth, and an optional tool allowlist. Field names are
-stable; callers build against them.
-"""
-
-from __future__ import annotations
-
-from typing import Annotated, Literal
-
-from pydantic import BaseModel, ConfigDict, Field, model_validator
-
-
-class BearerAuth(BaseModel):
-    """Header-token auth, sent as ``Authorization: Bearer <token>``."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    kind: Literal["bearer"] = "bearer"
-    token: str = Field(min_length=1, repr=False)
-
-
-McpAuth = Annotated[BearerAuth, Field(discriminator="kind")]
-
-
-class McpConnectionConfig(BaseModel):
-    """One MCP server the client can connect to.
-
-    Two transports are supported: streamable ``http`` (a remote endpoint) and
-    ``stdio`` (a local server launched as a subprocess).
-    """
-
-    model_config = ConfigDict(extra="forbid")
-
-    name: str = Field(min_length=1)
-    """Namespaced tool prefix, unique per run (e.g. ``github``)."""
-
-    transport: Literal["http", "stdio"] = "http"
-    """``http`` for a streamable HTTP endpoint, ``stdio`` for a local subprocess."""
-
-    url: str | None = Field(default=None, min_length=1)
-    """The MCP server endpoint. Required for ``http``."""
-
-    auth: McpAuth | None = None
-    """Bearer token for the server. Optional; a local stdio server usually
-    needs none."""
-
-    command: str | None = Field(default=None, min_length=1)
-    """The executable to launch for ``stdio``. Required for ``stdio``."""
-
-    args: list[str] = Field(default_factory=list)
-    """Arguments passed to ``command`` (stdio only)."""
-
-    env: dict[str, str] = Field(default_factory=dict)
-    """Extra environment variables for the stdio subprocess."""
-
-    allowed_tools: list[str] | None = None
-    """Tool allowlist, applied after the server lists its tools. ``None`` (the
-    default) exposes every tool the server lists; a list restricts to it."""
-
-    notes: str | None = None
-    """Free-text notes for the agent describing what this connection is and how
-    to use it. When set, they are prefixed onto each of the connection's tool
-    descriptions so the agent sees them."""
-
-    @model_validator(mode="after")
-    def _check_transport_fields(self) -> McpConnectionConfig:
-        if self.transport == "http" and not self.url:
-            raise ValueError("an http MCP connection requires 'url'")
-        if self.transport == "stdio" and not self.command:
-            raise ValueError("a stdio MCP connection requires 'command'")
-        return self

strix/tools/mcp/loader.py

@@ -1,132 +0,0 @@
-"""Read the open-source user's MCP servers from ``~/.strix/mcp-servers.json``.
-
-An open-source user lists the MCP servers they want the agent to reach in a
-small JSON file. Strix reads it at the start of a run, connects to each server,
-and registers its tools. The file is optional; without it the run simply gets
-no MCP tools.
-
-Parsing is fail-open. A single malformed entry is logged and skipped rather than
-raising, so one bad row never blocks the servers that are valid, and a missing
-or unreadable file yields an empty list.
-"""
-
-from __future__ import annotations
-
-import json
-import logging
-import os
-from pathlib import Path
-from typing import cast
-
-from pydantic import ValidationError
-
-from strix.tools.mcp.config import McpConnectionConfig
-
-
-logger = logging.getLogger(__name__)
-
-
-_DEFAULT_PATH: Path = Path.home() / ".strix" / "mcp-servers.json"
-_PATH_ENV_VAR = "STRIX_MCP_CONFIG"
-# Per-run selection, set by the --mcp-server / --mcp-exclude CLI flags. Each is a
-# comma-separated list of connection names.
-_ONLY_ENV_VAR = "STRIX_MCP_ONLY"
-_EXCLUDE_ENV_VAR = "STRIX_MCP_EXCLUDE"
-
-
-def _resolve_path(path: Path | None) -> Path:
-    if path is not None:
-        return path
-    override = os.environ.get(_PATH_ENV_VAR)
-    if override:
-        return Path(override)
-    return _DEFAULT_PATH
-
-
-def _dedupe_by_name(configs: list[McpConnectionConfig]) -> list[McpConnectionConfig]:
-    """Keep the first connection of each name, dropping later duplicates.
-
-    Names namespace a server's tools (``<name>.<tool>``), so two connections
-    sharing a name would collide and the second's tools would be silently
-    rejected at registration. Drop the duplicate here, with a warning, instead.
-    """
-    seen: set[str] = set()
-    unique: list[McpConnectionConfig] = []
-    for config in configs:
-        if config.name in seen:
-            logger.warning(
-                "Ignoring MCP server %r: another connection already uses that name "
-                "(names must be unique because they namespace the server's tools).",
-                config.name,
-            )
-            continue
-        seen.add(config.name)
-        unique.append(config)
-    return unique
-
-
-def _parse_names(env_var: str) -> set[str]:
-    return {name.strip() for name in os.environ.get(env_var, "").split(",") if name.strip()}
-
-
-def _apply_run_selection(configs: list[McpConnectionConfig]) -> list[McpConnectionConfig]:
-    """Restrict this run's connections to an optional include/exclude selection.
-
-    ``STRIX_MCP_ONLY`` (if set) keeps only the named connections; then
-    ``STRIX_MCP_EXCLUDE`` drops any named connection. With neither set, every
-    connection is kept.
-    """
-    only = _parse_names(_ONLY_ENV_VAR)
-    exclude = _parse_names(_EXCLUDE_ENV_VAR)
-    if not only and not exclude:
-        return configs
-
-    available = {config.name for config in configs}
-    for name in sorted((only | exclude) - available):
-        logger.warning(
-            "MCP connection selection named %r, which is not configured; ignoring it", name
-        )
-
-    selected: list[McpConnectionConfig] = []
-    for config in configs:
-        if only and config.name not in only:
-            continue
-        if config.name in exclude:
-            continue
-        selected.append(config)
-    return selected
-
-
-def load_user_mcp_configs(path: Path | None = None) -> list[McpConnectionConfig]:
-    """Load MCP connection configs from the user's JSON file.
-
-    The path is ``path`` if given, else ``$STRIX_MCP_CONFIG``, else
-    ``~/.strix/mcp-servers.json``. The file is a JSON list of server entries.
-    A missing file returns ``[]``; an unreadable or non-list file is logged and
-    returns ``[]``; individual entries that fail validation are logged and
-    skipped. Connections sharing a name are de-duplicated (first wins), and an
-    optional per-run include/exclude selection is applied last.
-    """
-    source = _resolve_path(path)
-    if not source.exists():
-        return []
-
-    try:
-        raw = json.loads(source.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError):
-        logger.exception("Could not read MCP config at %s; ignoring it", source)
-        return []
-
-    if not isinstance(raw, list):
-        logger.warning("MCP config at %s is not a JSON list; ignoring it", source)
-        return []
-
-    entries = cast("list[object]", raw)
-    configs: list[McpConnectionConfig] = []
-    for index, entry in enumerate(entries):
-        try:
-            configs.append(McpConnectionConfig.model_validate(entry))
-        except ValidationError as exc:
-            logger.warning("Skipping invalid MCP server entry #%d in %s: %s", index, source, exc)
-
-    return _apply_run_selection(_dedupe_by_name(configs))

strix/tools/reporting/tool.py

@@ -1,8 +1,8 @@
-"""Reporting tools — file vuln findings (with dedup + CVSS) and read them back.
+"""Reporting tools — file findings (with dedup + CVSS) and read them back.
 
 ``create_vulnerability_report`` / ``create_dependency_report`` file findings;
-``list_reports`` / ``get_report`` let any agent (notably the root orchestrator)
-review what's been filed so far across the whole scan.
+``update_vulnerability_report`` amends a known dynamic finding;
+``list_reports`` / ``get_report`` let the root orchestrator review findings.
 """
 
 from __future__ import annotations
@@ -162,6 +162,24 @@ _REQUIRED_FIELDS = {
 
 _VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"})
 
+_AMENDABLE_FIELDS = (
+    "title",
+    "description",
+    "impact",
+    "technical_analysis",
+    "poc_description",
+    "poc_script_code",
+    "remediation_steps",
+    "evidence",
+    "assumptions",
+    "fix_effort",
+    "cvss_breakdown",
+    "endpoint",
+    "method",
+    "cwe",
+    "code_locations",
+)
+
 
 async def _do_create(  # noqa: PLR0912
     *,
@@ -277,7 +295,10 @@ async def _do_create(  # noqa: PLR0912
                 "success": False,
                 "error": (
                     f"Potential duplicate of '{duplicate_title}' "
-                    f"(id={duplicate_id[:8]}...) — do not re-report the same vulnerability"
+                    f"(id={duplicate_id[:8]}...) — do not re-report the same vulnerability. "
+                    f"If new validated evidence shows greater or lower impact than filed, "
+                    f"amend this finding with update_vulnerability_report using id "
+                    f"'{duplicate_id}' instead."
                 ),
                 "duplicate_of": duplicate_id,
                 "duplicate_title": duplicate_title,
@@ -329,6 +350,189 @@ async def _do_create(  # noqa: PLR0912
         }
 
 
+async def _do_update(  # noqa: PLR0911, PLR0912, PLR0915
+    *,
+    report_id: str,
+    update_reason: str,
+    title: str | None = None,
+    description: str | None = None,
+    impact: str | None = None,
+    technical_analysis: str | None = None,
+    poc_description: str | None = None,
+    poc_script_code: str | None = None,
+    remediation_steps: str | None = None,
+    evidence: str | None = None,
+    assumptions: str | None = None,
+    fix_effort: str | None = None,
+    cvss_breakdown: dict[str, str] | None = None,
+    endpoint: str | None = None,
+    method: str | None = None,
+    cwe: str | None = None,
+    code_locations: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Validate and amend one known dynamic vulnerability report."""
+    if not report_id.strip():
+        return {"success": False, "error": "report_id cannot be empty"}
+    if not update_reason.strip():
+        return {"success": False, "error": "update_reason cannot be empty"}
+
+    raw_updates = {
+        name: value
+        for name, value in {
+            "title": title,
+            "description": description,
+            "impact": impact,
+            "technical_analysis": technical_analysis,
+            "poc_description": poc_description,
+            "poc_script_code": poc_script_code,
+            "remediation_steps": remediation_steps,
+            "evidence": evidence,
+            "assumptions": assumptions,
+            "fix_effort": fix_effort,
+            "cvss_breakdown": cvss_breakdown,
+            "endpoint": endpoint,
+            "method": method,
+            "cwe": cwe,
+            "code_locations": code_locations,
+        }.items()
+        if value is not None and name in _AMENDABLE_FIELDS
+    }
+    if not raw_updates:
+        return {
+            "success": False,
+            "error": "At least one amendable field must be supplied",
+        }
+
+    try:
+        from strix.report.state import get_global_report_state
+
+        report_state = get_global_report_state()
+        if report_state is None:
+            return {
+                "success": False,
+                "error": "Report state is unavailable",
+            }
+
+        existing = report_state.get_existing_vulnerabilities()
+        report = next((item for item in existing if item.get("id") == report_id), None)
+        if report is None:
+            valid_ids = [str(item["id"]) for item in existing if item.get("id")]
+            result: dict[str, Any] = {
+                "success": False,
+                "error": f"Report with id '{report_id}' was not found",
+            }
+            if valid_ids:
+                result["valid_report_ids"] = valid_ids
+            return result
+
+        errors: list[str] = []
+        updates: dict[str, Any] = {}
+        for field, value in raw_updates.items():
+            if field in {
+                "title",
+                "description",
+                "impact",
+                "technical_analysis",
+                "poc_description",
+                "poc_script_code",
+                "remediation_steps",
+                "evidence",
+                "assumptions",
+                "endpoint",
+                "method",
+            }:
+                normalized_value = str(value).strip()
+                if not normalized_value:
+                    errors.append(
+                        _REQUIRED_FIELDS.get(
+                            field,
+                            f"{field.replace('_', ' ').capitalize()} cannot be empty",
+                        ),
+                    )
+                else:
+                    updates[field] = normalized_value
+
+        if "fix_effort" in raw_updates:
+            normalized_fix_effort = str(fix_effort).strip().lower()
+            if normalized_fix_effort not in _VALID_FIX_EFFORT:
+                errors.append(
+                    f"Invalid fix_effort: {normalized_fix_effort!r}. "
+                    f"Must be one of: {sorted(_VALID_FIX_EFFORT)}"
+                )
+            else:
+                updates["fix_effort"] = normalized_fix_effort
+
+        if "cvss_breakdown" in raw_updates:
+            if not isinstance(cvss_breakdown, dict) or not cvss_breakdown:
+                errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics")
+            else:
+                for name, valid in _CVSS_VALID.items():
+                    metric_value = cvss_breakdown.get(name)
+                    if metric_value not in valid:
+                        errors.append(
+                            f"Invalid {name}: {metric_value}. Must be one of: {valid}",
+                        )
+                updates["cvss_breakdown"] = cvss_breakdown
+
+        if "cwe" in raw_updates:
+            parsed_cwe = _extract_cwe(str(cwe))
+            cwe_err = _validate_cwe(parsed_cwe)
+            if cwe_err:
+                errors.append(cwe_err)
+            else:
+                updates["cwe"] = parsed_cwe
+
+        if "code_locations" in raw_updates:
+            if not isinstance(code_locations, list) or not code_locations:
+                errors.append("code_locations must contain at least one location")
+            else:
+                parsed_locations = _normalize_code_locations(code_locations)
+                if not parsed_locations:
+                    errors.append("code_locations must contain at least one valid location")
+                else:
+                    errors.extend(_validate_code_locations(parsed_locations))
+                    updates["code_locations"] = parsed_locations
+
+        if errors:
+            return {"success": False, "error": "Validation failed", "errors": errors}
+
+        if "cvss_breakdown" in updates:
+            try:
+                cvss_score, severity, _vector = _calculate_cvss(updates["cvss_breakdown"])
+            except ValueError as exc:
+                return {"success": False, "error": "Validation failed", "errors": [str(exc)]}
+            updates["cvss"] = cvss_score
+            updates["severity"] = severity
+
+        updated = report_state.update_vulnerability_report(
+            report_id,
+            update_reason,
+            **updates,
+        )
+        if updated is None:
+            return {
+                "success": False,
+                "error": f"Report with id '{report_id}' was not found",
+            }
+    except (AttributeError, KeyError, TypeError, ValueError) as exc:
+        logger.exception("update_vulnerability_report persistence failed")
+        return {"success": False, "error": f"Failed to update vulnerability report: {exc!s}"}
+
+    logger.info(
+        "Vulnerability report updated: id=%s fields=%s",
+        report_id,
+        sorted(raw_updates),
+    )
+    return {
+        "success": True,
+        "message": f"Vulnerability report '{report_id}' updated successfully",
+        "report_id": report_id,
+        "severity": updated.get("severity"),
+        "cvss_score": updated.get("cvss"),
+        "updated_at": updated.get("updated_at"),
+    }
+
+
 def _caller_identity(ctx: RunContextWrapper) -> tuple[str | None, str | None]:
     """Return the (agent_id, agent_name) of the agent invoking this tool."""
     inner = ctx.context if isinstance(ctx.context, dict) else {}
@@ -377,6 +581,8 @@ async def create_vulnerability_report(
     - Suspicions you haven't confirmed with a PoC.
     - Tracking multiple vulnerabilities at once — one report per vuln.
     - Re-reporting something you (or another agent) already filed.
+    - A chain that only amplifies an existing finding's impact on the same
+      asset and root cause. Use ``update_vulnerability_report`` instead.
     - Known-CVE dependency / supply-chain findings that can't be
       dynamically PoC'd — a vulnerable dependency version pinned in a
       lockfile/manifest that matches a published advisory. File those
@@ -408,8 +614,10 @@ async def create_vulnerability_report(
 
     Automatic LLM-based **deduplication** rejects reports that describe
     the same root cause on the same asset as an existing report. If you
-    get a ``duplicate_of`` response, do NOT retry — move on to other
-    areas.
+    get a ``duplicate_of`` response, do not re-submit the same vulnerability.
+    If new validated evidence shows greater or lower impact for that finding,
+    amend it with ``update_vulnerability_report`` using the returned id.
+    Otherwise, move on to other areas.
 
     **Report output rules** (this content may be rendered into generated
     reports):
@@ -701,6 +909,69 @@ async def create_vulnerability_report(
     return json.dumps(result, ensure_ascii=False, default=str)
 
 
+@function_tool(timeout=180, strict_mode=False)
+async def update_vulnerability_report(
+    ctx: RunContextWrapper,
+    report_id: str,
+    update_reason: str,
+    title: str | None = None,
+    description: str | None = None,
+    impact: str | None = None,
+    technical_analysis: str | None = None,
+    poc_description: str | None = None,
+    poc_script_code: str | None = None,
+    remediation_steps: str | None = None,
+    evidence: str | None = None,
+    assumptions: str | None = None,
+    fix_effort: str | None = None,
+    cvss_breakdown: dict[str, str] | None = None,
+    endpoint: str | None = None,
+    method: str | None = None,
+    cwe: str | None = None,
+    code_locations: list[dict[str, Any]] | None = None,
+) -> str:
+    """Amend a known vulnerability report when new evidence changes its impact.
+
+    Use ``create_vulnerability_report`` when the chain proves a distinct issue.
+    Use this tool when the same root cause on the same asset has greater impact.
+    The tool recomputes severity and score from ``cvss_breakdown``.
+
+    Apply the same report output rules and CVSS calibration discipline as the
+    create tool. An amendment must use evidence from a validated result. Do
+    not use this tool for a speculative upgrade.
+
+    Pass at least one amendable field:
+
+    - ``title``, ``description``, ``impact``, ``technical_analysis``,
+      ``poc_description``, ``poc_script_code``, ``remediation_steps``,
+      ``evidence``, ``assumptions``, ``fix_effort``, ``cvss_breakdown``,
+      ``endpoint``, ``method``, ``cwe``, or ``code_locations``.
+
+    Do not change ``target``, ``cve``, or dependency metadata. File a new
+    report instead when those values must change.
+    """
+    result = await _do_update(
+        report_id=report_id,
+        update_reason=update_reason,
+        title=title,
+        description=description,
+        impact=impact,
+        technical_analysis=technical_analysis,
+        poc_description=poc_description,
+        poc_script_code=poc_script_code,
+        remediation_steps=remediation_steps,
+        evidence=evidence,
+        assumptions=assumptions,
+        fix_effort=fix_effort,
+        cvss_breakdown=cvss_breakdown,
+        endpoint=endpoint,
+        method=method,
+        cwe=cwe,
+        code_locations=code_locations,
+    )
+    return json.dumps(result, ensure_ascii=False, default=str)
+
+
 _DEP_SEVERITY_FROM_CVSS = {
     (9.0, 10.0): "critical",
     (7.0, 9.0): "high",

tests/test_cli_callbacks.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import Mock
+
+from rich.console import Console
+from rich.panel import Panel
+
+from strix.interface import cli
+
+
+def test_cli_report_callbacks_render_new_and_updated_findings() -> None:
+    report_state = SimpleNamespace(
+        vulnerability_found_callback=None,
+        vulnerability_updated_callback=None,
+    )
+    console = Mock(spec=Console)
+
+    cli._configure_report_callbacks(cast("Any", report_state), console)
+
+    report = {"id": "vuln-0001", "title": "Unsafe redirect"}
+    report_state.vulnerability_found_callback(report)
+    report_state.vulnerability_updated_callback(report)
+
+    panels = [call.args[0] for call in console.print.call_args_list if call.args]
+    assert all(isinstance(panel, Panel) for panel in panels)
+    assert panels[0].title == "[bold red]VULN-0001"
+    assert panels[1].title == "[bold yellow]VULN-0001 — UPDATED FINDING"

tests/test_cli_mcp_config.py

@@ -1,88 +0,0 @@
-"""Tests for the --mcp-config CLI flag."""
-
-from __future__ import annotations
-
-import importlib
-import os
-import sys
-from types import SimpleNamespace
-from typing import TYPE_CHECKING, Any
-
-import pytest
-
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
-
-cli_main: Any = importlib.import_module("strix.interface.main")
-
-
-def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None:
-    monkeypatch.setattr(
-        cli_main,
-        "load_settings",
-        lambda: SimpleNamespace(runtime=SimpleNamespace(max_local_copy_mb=1024)),
-    )
-
-
-def test_mcp_config_flag_sets_loader_override(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    config = tmp_path / "servers.json"
-    config.write_text("[]", encoding="utf-8")
-    _stub_settings(monkeypatch)
-    # delenv records "originally absent" so monkeypatch removes whatever the
-    # parser sets, keeping the override from leaking into other tests.
-    monkeypatch.delenv("STRIX_MCP_CONFIG", raising=False)
-    monkeypatch.setattr(
-        sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--mcp-config", str(config)]
-    )
-
-    args = cli_main.parse_arguments()
-
-    assert args.mcp_config == str(config)
-    assert os.environ["STRIX_MCP_CONFIG"] == str(config)
-
-
-def test_mcp_config_flag_rejects_missing_file(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
-) -> None:
-    _stub_settings(monkeypatch)
-    monkeypatch.delenv("STRIX_MCP_CONFIG", raising=False)
-    missing = tmp_path / "nope.json"
-    monkeypatch.setattr(
-        sys, "argv", ["strix", "-t", "https://test.com/", "-n", "--mcp-config", str(missing)]
-    )
-
-    with pytest.raises(SystemExit):
-        cli_main.parse_arguments()
-
-    assert "--mcp-config file not found" in capsys.readouterr().err
-
-
-def test_mcp_server_flags_set_selection_env(monkeypatch: pytest.MonkeyPatch) -> None:
-    _stub_settings(monkeypatch)
-    monkeypatch.delenv("STRIX_MCP_ONLY", raising=False)
-    monkeypatch.delenv("STRIX_MCP_EXCLUDE", raising=False)
-    monkeypatch.setattr(
-        sys,
-        "argv",
-        [
-            "strix",
-            "-t",
-            "https://test.com/",
-            "-n",
-            "--mcp-server",
-            "a",
-            "--mcp-server",
-            "b",
-            "--mcp-exclude",
-            "c",
-        ],
-    )
-
-    cli_main.parse_arguments()
-
-    assert os.environ["STRIX_MCP_ONLY"] == "a,b"
-    assert os.environ["STRIX_MCP_EXCLUDE"] == "c"

tests/test_go_tui_runtime.py

@@ -12,6 +12,7 @@ import threading
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any, cast
+from unittest.mock import Mock
 
 import pytest
 
@@ -95,6 +96,25 @@ def test_binary_command_ignores_unconstrained_path_sidecar(
         GoTuiRuntime.binary_command()
 
 
+@pytest.mark.asyncio
+async def test_init_run_state_wires_updated_report_callback(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    runtime = GoTuiRuntime(args())
+    notify_changed = Mock()
+    monkeypatch.setattr(runtime.controller, "notify_changed", notify_changed)
+
+    runtime.init_run_state()
+
+    assert runtime.report_state is not None
+    assert runtime.report_state.vulnerability_updated_callback is not None
+    notify_changed.reset_mock()
+    runtime.report_state.vulnerability_updated_callback({"id": "vuln-0001"})
+    notify_changed.assert_called_once_with()
+
+
 def test_child_environment_excludes_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("OPENAI_API_KEY", "openai-secret")
     monkeypatch.setenv("AWS_ACCESS_KEY_ID", "aws-id")

tests/test_mcp_client.py

@@ -1,631 +0,0 @@
-"""Tests for the generic MCP client: config contract, namespacing, and filtering."""
-
-from __future__ import annotations
-
-import asyncio
-import json
-from typing import TYPE_CHECKING, Any
-
-import pytest
-from agents.mcp import MCPServer, MCPServerStdio, MCPServerStreamableHttp
-from mcp.types import CallToolResult, TextContent
-from mcp.types import Tool as MCPTool
-from pydantic import ValidationError
-
-from strix.agents import factory
-from strix.core.runner import _mcp_connection_notes
-from strix.interface.tui.live_view import _tool_status_from_result
-from strix.tools.mcp import (
-    BearerAuth,
-    ConnectedMcpServer,
-    McpConnectionConfig,
-    load_user_mcp_configs,
-)
-from strix.tools.mcp import client as mcp_client
-from strix.tools.mcp.client import _auth_headers, _build_server, _register_server_tools
-
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
-    from agents.tool import Tool
-
-
-class FakeMCPServer(MCPServer):
-    """A connected MCP server stand-in, so tests never touch the network."""
-
-    def __init__(self, name: str, tools: list[MCPTool]) -> None:
-        super().__init__()
-        self._name = name
-        self._tools = tools
-        self.calls: list[tuple[str, dict[str, Any] | None]] = []
-
-    @property
-    def name(self) -> str:
-        return self._name
-
-    async def connect(self) -> None:
-        return None
-
-    async def cleanup(self) -> None:
-        return None
-
-    async def list_tools(
-        self,
-        run_context: Any = None,
-        agent: Any = None,
-    ) -> list[MCPTool]:
-        return list(self._tools)
-
-    async def call_tool(
-        self,
-        tool_name: str,
-        arguments: dict[str, Any] | None,
-        meta: dict[str, Any] | None = None,
-    ) -> CallToolResult:
-        self.calls.append((tool_name, arguments))
-        return CallToolResult(content=[TextContent(type="text", text=f"routed:{tool_name}")])
-
-    async def list_prompts(self) -> Any:
-        raise NotImplementedError
-
-    async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> Any:
-        raise NotImplementedError
-
-
-def _mcp_tool(name: str) -> MCPTool:
-    return MCPTool(
-        name=name,
-        description=f"remote tool {name}",
-        inputSchema={"type": "object", "properties": {}},
-    )
-
-
-def _config(name: str, allowed_tools: list[str]) -> McpConnectionConfig:
-    return McpConnectionConfig(
-        name=name,
-        url="https://mcp.example.com",
-        auth=BearerAuth(token="run-token"),
-        allowed_tools=allowed_tools,
-    )
-
-
-@pytest.fixture(autouse=True)
-def _reset_registry() -> Any:
-    saved = list(factory._EXTRA_TOOLS)
-    factory._EXTRA_TOOLS.clear()
-    try:
-        yield
-    finally:
-        factory._EXTRA_TOOLS[:] = saved
-
-
-# --- config contract ---------------------------------------------------------
-
-
-def test_bearer_config_parses_from_dict() -> None:
-    config = McpConnectionConfig.model_validate(
-        {
-            "name": "files_main",
-            "transport": "http",
-            "url": "https://mcp.example.com",
-            "auth": {"kind": "bearer", "token": "abc"},
-            "allowed_tools": ["list_files"],
-        }
-    )
-
-    assert isinstance(config.auth, BearerAuth)
-    assert config.auth.token == "abc"
-    assert config.allowed_tools == ["list_files"]
-
-
-def test_unknown_auth_kind_is_rejected() -> None:
-    with pytest.raises(ValidationError):
-        McpConnectionConfig.model_validate(
-            {
-                "name": "x",
-                "url": "https://mcp.example.com",
-                "auth": {"kind": "oauth", "token": "abc"},
-            }
-        )
-
-
-def test_stdio_config_parses_from_dict() -> None:
-    config = McpConnectionConfig.model_validate(
-        {
-            "name": "local_fs",
-            "transport": "stdio",
-            "command": "npx",
-            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/srv/data"],
-            "env": {"FOO": "bar"},
-        }
-    )
-
-    assert config.transport == "stdio"
-    assert config.command == "npx"
-    assert config.args == ["-y", "@modelcontextprotocol/server-filesystem", "/srv/data"]
-    assert config.env == {"FOO": "bar"}
-    # A local stdio server needs no auth, and omitting allowed_tools means "all".
-    assert config.auth is None
-    assert config.allowed_tools is None
-
-
-def test_http_config_without_url_is_rejected() -> None:
-    with pytest.raises(ValidationError):
-        McpConnectionConfig.model_validate(
-            {
-                "name": "x",
-                "transport": "http",
-                "auth": {"kind": "bearer", "token": "abc"},
-            }
-        )
-
-
-def test_stdio_config_without_command_is_rejected() -> None:
-    with pytest.raises(ValidationError):
-        McpConnectionConfig.model_validate(
-            {
-                "name": "x",
-                "transport": "stdio",
-            }
-        )
-
-
-def test_empty_name_is_rejected() -> None:
-    with pytest.raises(ValidationError):
-        McpConnectionConfig.model_validate(
-            {
-                "name": "",
-                "url": "https://mcp.example.com",
-                "auth": {"kind": "bearer", "token": "abc"},
-            }
-        )
-
-
-def test_unknown_field_is_rejected() -> None:
-    with pytest.raises(ValidationError):
-        McpConnectionConfig.model_validate(
-            {
-                "name": "x",
-                "url": "https://mcp.example.com",
-                "auth": {"kind": "bearer", "token": "abc"},
-                "surprise": True,
-            }
-        )
-
-
-# --- auth headers ------------------------------------------------------------
-
-
-def test_bearer_auth_builds_authorization_header() -> None:
-    headers = _auth_headers(_config("files_main", []))
-
-    assert headers == {"Authorization": "Bearer run-token"}
-
-
-# --- namespacing and filtering -----------------------------------------------
-
-
-def _registered_names() -> list[str]:
-    return [tool.name for tool in factory.registered_agent_tools()]
-
-
-@pytest.mark.asyncio
-async def test_tools_are_namespaced_per_connection() -> None:
-    server_a = FakeMCPServer("conn_a", [_mcp_tool("describe")])
-    server_b = FakeMCPServer("conn_b", [_mcp_tool("describe")])
-
-    await _register_server_tools(_config("conn_a", ["describe"]), server_a)
-    await _register_server_tools(_config("conn_b", ["describe"]), server_b)
-
-    # Same remote tool name on two connections does not collide.
-    assert _registered_names() == ["conn_a.describe", "conn_b.describe"]
-
-
-@pytest.mark.asyncio
-async def test_disallowed_tool_is_not_registered() -> None:
-    server = FakeMCPServer(
-        "files_main",
-        [_mcp_tool("list_files"), _mcp_tool("search")],
-    )
-
-    await _register_server_tools(_config("files_main", ["list_files"]), server)
-
-    names = _registered_names()
-    assert "files_main.list_files" in names
-    assert "files_main.search" not in names
-
-
-@pytest.mark.asyncio
-async def test_allowed_tools_none_registers_every_listed_tool() -> None:
-    server = FakeMCPServer(
-        "local_fs",
-        [_mcp_tool("read_file"), _mcp_tool("write_file")],
-    )
-    config = McpConnectionConfig(name="local_fs", url="https://mcp.example.com", allowed_tools=None)
-
-    await _register_server_tools(config, server)
-
-    names = _registered_names()
-    assert "local_fs.read_file" in names
-    assert "local_fs.write_file" in names
-
-
-@pytest.mark.asyncio
-async def test_allowed_tools_list_restricts_registration() -> None:
-    server = FakeMCPServer(
-        "local_fs",
-        [_mcp_tool("read_file"), _mcp_tool("write_file")],
-    )
-
-    await _register_server_tools(_config("local_fs", ["read_file"]), server)
-
-    names = _registered_names()
-    assert names == ["local_fs.read_file"]
-
-
-@pytest.mark.asyncio
-async def test_registered_tool_routes_to_its_server_with_the_original_name() -> None:
-    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
-
-    tools: list[Tool] = await _register_server_tools(
-        _config("files_main", ["list_files"]), server
-    )
-    tool = tools[0]
-
-    output = await tool.on_invoke_tool(None, "{}")  # type: ignore[union-attr]
-
-    # The call reaches the right server, addressed by the unprefixed remote name.
-    assert server.calls == [("list_files", {})]
-    assert output == {"type": "text", "text": "routed:list_files"}
-
-
-# --- result transform --------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_result_transform_receives_namespaced_name_and_structured_result() -> None:
-    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
-    seen: list[tuple[str, Any]] = []
-
-    def transform(name: str, structured: Any) -> Any:
-        seen.append((name, structured))
-        return {"kept": structured["content"][0]["text"]}
-
-    tools: list[Tool] = await _register_server_tools(
-        _config("files_main", ["list_files"]), server, result_transform=transform
-    )
-
-    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
-
-    # The underlying MCP call still routes by the unprefixed remote name.
-    assert server.calls == [("list_files", {})]
-
-    # The transform is called with the namespaced name and the parsed result.
-    assert len(seen) == 1
-    name, structured = seen[0]
-    assert name == "files_main.list_files"
-    # A parsed CallToolResult (dict/list), not a pre-serialized string.
-    assert structured["content"][0]["text"] == "routed:list_files"
-    assert structured["isError"] is False
-
-    # The transform's return value is exactly what the tool yields.
-    assert output == {"kept": "routed:list_files"}
-
-
-@pytest.mark.asyncio
-async def test_result_transform_can_rewrite_the_tool_output() -> None:
-    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
-
-    def transform(_name: str, structured: Any) -> Any:
-        # Keep only a truncated view of the text field.
-        return structured["content"][0]["text"][:6]
-
-    tools: list[Tool] = await _register_server_tools(
-        _config("files_main", ["list_files"]), server, result_transform=transform
-    )
-
-    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
-
-    assert output == "routed"
-
-
-@pytest.mark.asyncio
-async def test_without_result_transform_output_is_unchanged() -> None:
-    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
-
-    tools: list[Tool] = await _register_server_tools(
-        _config("files_main", ["list_files"]), server, result_transform=None
-    )
-
-    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
-
-    # Same shape the SDK produces today: no transform in the path.
-    assert server.calls == [("list_files", {})]
-    assert output == {"type": "text", "text": "routed:list_files"}
-
-
-# --- error status capture ----------------------------------------------------
-
-
-class ErroringMCPServer(FakeMCPServer):
-    """A connected server whose calls come back as MCP errors (isError=True)."""
-
-    async def call_tool(
-        self,
-        tool_name: str,
-        arguments: dict[str, Any] | None,
-        meta: dict[str, Any] | None = None,
-    ) -> CallToolResult:
-        self.calls.append((tool_name, arguments))
-        return CallToolResult(
-            content=[TextContent(type="text", text=f"boom:{tool_name}")],
-            isError=True,
-        )
-
-
-@pytest.mark.asyncio
-async def test_errored_mcp_result_is_flagged_failed_for_the_tui() -> None:
-    server = ErroringMCPServer("files_main", [_mcp_tool("list_files")])
-
-    tools: list[Tool] = await _register_server_tools(
-        _config("files_main", ["list_files"]), server
-    )
-    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
-
-    # The error text stays exactly what the agent gets today; a success:False tag
-    # rides alongside it purely so the TUI can tell the call apart from a success.
-    assert output == {"type": "text", "text": "boom:list_files", "success": False}
-    assert _tool_status_from_result(output) == "failed"
-
-
-@pytest.mark.asyncio
-async def test_successful_mcp_result_stays_completed_for_the_tui() -> None:
-    server = FakeMCPServer("files_main", [_mcp_tool("list_files")])
-
-    tools: list[Tool] = await _register_server_tools(
-        _config("files_main", ["list_files"]), server
-    )
-    output = await tools[0].on_invoke_tool(None, "{}")  # type: ignore[union-attr]
-
-    # A non-error result is untouched and keeps rendering as done.
-    assert output == {"type": "text", "text": "routed:list_files"}
-    assert _tool_status_from_result(output) == "completed"
-
-
-# --- server build branch -----------------------------------------------------
-
-
-def test_build_server_stdio_branch() -> None:
-    config = McpConnectionConfig(
-        name="local_fs",
-        transport="stdio",
-        command="my-server",
-        args=["--flag", "value"],
-        env={"TOKEN": "x"},
-    )
-
-    server = _build_server(config)
-
-    # Built, not connected: no subprocess is launched here.
-    assert isinstance(server, MCPServerStdio)
-    assert server.name == "local_fs"
-    assert server.params.command == "my-server"
-    assert server.params.args == ["--flag", "value"]
-    assert server.params.env == {"TOKEN": "x"}
-
-
-def test_build_server_http_branch() -> None:
-    server = _build_server(_config("files_main", ["list_files"]))
-
-    assert isinstance(server, MCPServerStreamableHttp)
-    assert server.name == "files_main"
-
-
-# --- loader ------------------------------------------------------------------
-
-
-def test_loader_parses_stdio_and_http_entries(tmp_path: Path) -> None:
-    config_file = tmp_path / "mcp-servers.json"
-    config_file.write_text(
-        json.dumps(
-            [
-                {
-                    "name": "local_fs",
-                    "transport": "stdio",
-                    "command": "npx",
-                    "args": ["-y", "server-filesystem"],
-                },
-                {
-                    "name": "files_main",
-                    "transport": "http",
-                    "url": "https://mcp.example.com",
-                    "auth": {"kind": "bearer", "token": "abc"},
-                    "allowed_tools": ["list_files"],
-                },
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    configs = load_user_mcp_configs(config_file)
-
-    assert [c.name for c in configs] == ["local_fs", "files_main"]
-    assert configs[0].transport == "stdio"
-    assert configs[1].allowed_tools == ["list_files"]
-
-
-def test_loader_skips_bad_entry_but_keeps_good_ones(tmp_path: Path) -> None:
-    config_file = tmp_path / "mcp-servers.json"
-    config_file.write_text(
-        json.dumps(
-            [
-                {"name": "broken", "transport": "http"},  # missing url
-                {
-                    "name": "local_fs",
-                    "transport": "stdio",
-                    "command": "npx",
-                },
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    configs = load_user_mcp_configs(config_file)
-
-    assert [c.name for c in configs] == ["local_fs"]
-
-
-def test_loader_returns_empty_when_file_absent(tmp_path: Path) -> None:
-    assert load_user_mcp_configs(tmp_path / "does-not-exist.json") == []
-
-
-def test_loader_reads_env_var_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
-    config_file = tmp_path / "from-env.json"
-    config_file.write_text(
-        json.dumps([{"name": "local_fs", "transport": "stdio", "command": "npx"}]),
-        encoding="utf-8",
-    )
-    monkeypatch.setenv("STRIX_MCP_CONFIG", str(config_file))
-
-    configs = load_user_mcp_configs()
-
-    assert [c.name for c in configs] == ["local_fs"]
-
-
-# --- connection notes --------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_connection_notes_are_carried_on_the_connection(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    server = FakeMCPServer("db", [_mcp_tool("query")])
-    monkeypatch.setattr(mcp_client, "_build_server", lambda _config: server)
-    config = McpConnectionConfig(
-        name="db",
-        url="https://mcp.example.com",
-        notes="Staging analytics DB; read-only.",
-        allowed_tools=["query"],
-    )
-
-    connections = await mcp_client.connect_mcp_servers([config])
-
-    # Notes ride on the connection (surfaced once), not stapled onto each tool.
-    assert connections[0].notes == "Staging analytics DB; read-only."
-
-
-def test_connection_notes_block_lists_only_noted_connections() -> None:
-    connections = [
-        ConnectedMcpServer(
-            server=FakeMCPServer("db", []), name="db", tool_count=2, notes="staging, read-only"
-        ),
-        ConnectedMcpServer(server=FakeMCPServer("fs", []), name="fs", tool_count=1, notes=None),
-    ]
-
-    block = _mcp_connection_notes(connections)
-
-    assert block is not None
-    assert "db" in block
-    assert "staging, read-only" in block
-    # A connection without notes is not listed.
-    assert "fs" not in block
-
-
-def test_connection_notes_block_is_none_without_notes() -> None:
-    connections = [
-        ConnectedMcpServer(server=FakeMCPServer("db", []), name="db", tool_count=1, notes=None)
-    ]
-
-    assert _mcp_connection_notes(connections) is None
-
-
-# --- cancellation cleanup ----------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_connect_cleans_up_when_cancelled_mid_connect(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    cleaned: list[str] = []
-
-    class _Tracking(FakeMCPServer):
-        def __init__(self, name: str, *, fail_connect: bool = False) -> None:
-            super().__init__(name, [_mcp_tool("t")])
-            self._fail_connect = fail_connect
-
-        async def connect(self) -> None:
-            if self._fail_connect:
-                raise asyncio.CancelledError
-
-        async def cleanup(self) -> None:
-            cleaned.append(self._name)
-
-    servers = {"good": _Tracking("good"), "bad": _Tracking("bad", fail_connect=True)}
-    monkeypatch.setattr(mcp_client, "_build_server", lambda config: servers[config.name])
-
-    configs = [
-        McpConnectionConfig(name="good", url="https://mcp.example.com", allowed_tools=["t"]),
-        McpConnectionConfig(name="bad", url="https://mcp.example.com", allowed_tools=["t"]),
-    ]
-
-    with pytest.raises(asyncio.CancelledError):
-        await mcp_client.connect_mcp_servers(configs)
-
-    # The server being connected when cancelled, and the one already connected,
-    # are both cleaned up rather than orphaned.
-    assert cleaned == ["bad", "good"]
-
-
-# --- duplicate names and run selection ---------------------------------------
-
-
-def _names_file(tmp_path: Path, *names: str) -> Path:
-    config_file = tmp_path / "mcp-servers.json"
-    config_file.write_text(
-        json.dumps([{"name": n, "transport": "stdio", "command": "npx"} for n in names]),
-        encoding="utf-8",
-    )
-    return config_file
-
-
-def test_loader_drops_duplicate_named_connections(tmp_path: Path) -> None:
-    config_file = tmp_path / "mcp-servers.json"
-    config_file.write_text(
-        json.dumps(
-            [
-                {"name": "dup", "transport": "stdio", "command": "first"},
-                {"name": "dup", "transport": "stdio", "command": "second"},
-                {"name": "other", "transport": "stdio", "command": "npx"},
-            ]
-        ),
-        encoding="utf-8",
-    )
-
-    configs = load_user_mcp_configs(config_file)
-
-    # Duplicate name is dropped; the first entry wins.
-    assert [c.name for c in configs] == ["dup", "other"]
-    assert configs[0].command == "first"
-
-
-def test_loader_include_selection_keeps_only_named(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    config_file = _names_file(tmp_path, "a", "b", "c")
-    monkeypatch.setenv("STRIX_MCP_ONLY", "a,c")
-
-    configs = load_user_mcp_configs(config_file)
-
-    assert [c.name for c in configs] == ["a", "c"]
-
-
-def test_loader_exclude_selection_drops_named(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    config_file = _names_file(tmp_path, "a", "b", "c")
-    monkeypatch.setenv("STRIX_MCP_EXCLUDE", "b")
-
-    configs = load_user_mcp_configs(config_file)
-
-    assert [c.name for c in configs] == ["a", "c"]

tests/test_vulnerability_update_tool.py

@@ -0,0 +1,262 @@
+"""Tests for amending filed vulnerability reports."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any, cast
+
+import pytest
+
+from strix.report.state import ReportState, set_global_report_state
+from strix.tools.reporting.tool import _do_update
+
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+_LOW_CVSS = {
+    "attack_vector": "N",
+    "attack_complexity": "H",
+    "privileges_required": "H",
+    "user_interaction": "R",
+    "scope": "U",
+    "confidentiality": "L",
+    "integrity": "N",
+    "availability": "N",
+}
+
+_CRITICAL_CVSS = {
+    "attack_vector": "N",
+    "attack_complexity": "L",
+    "privileges_required": "N",
+    "user_interaction": "N",
+    "scope": "U",
+    "confidentiality": "H",
+    "integrity": "H",
+    "availability": "H",
+}
+
+
+@pytest.fixture(autouse=True)
+def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
+    monkeypatch.chdir(tmp_path)
+    state = ReportState(run_name="test-run")
+    set_global_report_state(state)
+    state.add_vulnerability_report(
+        title="Unsafe redirect",
+        severity="low",
+        description="The redirect accepts attacker input.",
+        impact="Limited redirect manipulation.",
+        target="https://app.example.com",
+        technical_analysis="The handler does not validate the destination.",
+        poc_description="Send a crafted redirect value.",
+        poc_script_code="GET /redirect?url=https://example.net",
+        remediation_steps="Validate redirect destinations.",
+        evidence="The response contains the attacker-controlled destination.",
+        assumptions="Assumes a victim follows the link.",
+        fix_effort="medium",
+        cvss=3.1,
+        cvss_breakdown=_LOW_CVSS,
+        endpoint="/redirect",
+        method="GET",
+        cwe="CWE-601",
+        code_locations=[
+            {
+                "file": "src/redirect.py",
+                "start_line": 10,
+                "end_line": 12,
+                "snippet": "return redirect(url)",
+            },
+        ],
+    )
+    return state
+
+
+async def test_update_rejects_unknown_report_id() -> None:
+    result = await _do_update(
+        report_id="vuln-9999",
+        update_reason="The validation pass proved broader impact.",
+        impact="Broader impact.",
+    )
+
+    assert result["success"] is False
+    assert "vuln-9999" in result["error"]
+    assert result["valid_report_ids"] == ["vuln-0001"]
+
+
+async def test_update_requires_an_amendable_field() -> None:
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="The validation pass found no new field to amend.",
+    )
+
+    assert result == {
+        "success": False,
+        "error": "At least one amendable field must be supplied",
+    }
+
+
+async def test_update_requires_nonempty_reason() -> None:
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="  ",
+        impact="The impact is broader.",
+    )
+
+    assert result == {"success": False, "error": "update_reason cannot be empty"}
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "expected_error"),
+    [
+        ("impact", "", "Impact cannot be empty"),
+        ("impact", "   ", "Impact cannot be empty"),
+        ("endpoint", "", "Endpoint cannot be empty"),
+        ("endpoint", "   ", "Endpoint cannot be empty"),
+    ],
+)
+async def test_update_rejects_blank_text_fields(
+    report_state: ReportState,
+    field: str,
+    value: str,
+    expected_error: str,
+) -> None:
+    original_value = report_state.vulnerability_reports[0][field]
+    update = cast("dict[str, Any]", {field: value})
+
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="The source review supplied no content for this field.",
+        **update,
+    )
+
+    assert result["success"] is False
+    assert expected_error in result["errors"]
+    assert report_state.vulnerability_reports[0][field] == original_value
+
+
+async def test_update_changes_impact_only(report_state: ReportState) -> None:
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="The confirmed chain exposes account data.",
+        impact="The chain exposes account data.",
+    )
+
+    assert result["success"] is True
+    report = report_state.vulnerability_reports[0]
+    assert report["impact"] == "The chain exposes account data."
+    assert report["severity"] == "low"
+    assert report["cvss"] == 3.1
+    assert report["update_history"][0]["fields_changed"] == ["impact"]
+    assert "previous_severity" not in report["update_history"][0]
+
+
+async def test_cvss_update_recomputes_score_and_severity(report_state: ReportState) -> None:
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="The exploit chain proves full account compromise.",
+        cvss_breakdown=_CRITICAL_CVSS,
+    )
+
+    assert result["success"] is True
+    report = report_state.vulnerability_reports[0]
+    assert report["severity"] == "critical"
+    assert report["cvss"] == 9.8
+    history = report["update_history"][0]
+    assert history["fields_changed"] == ["cvss_breakdown", "cvss", "severity"]
+    assert history["previous_severity"] == "low"
+    assert history["previous_cvss_score"] == 3.1
+
+
+async def test_update_history_is_append_only(report_state: ReportState) -> None:
+    await _do_update(
+        report_id="vuln-0001",
+        update_reason="The chain proves account access.",
+        impact="Account access is possible.",
+    )
+    await _do_update(
+        report_id="vuln-0001",
+        update_reason="The second proof confirms persistent access.",
+        evidence="The second proof confirms persistent access.",
+    )
+
+    history = report_state.vulnerability_reports[0]["update_history"]
+    assert len(history) == 2
+    assert history[0]["update_reason"] == "The chain proves account access."
+    assert history[1]["update_reason"] == "The second proof confirms persistent access."
+    assert all("description" not in entry for entry in history)
+
+
+async def test_update_callback_fires(report_state: ReportState) -> None:
+    updated: list[dict[str, Any]] = []
+    report_state.vulnerability_updated_callback = updated.append
+
+    await _do_update(
+        report_id="vuln-0001",
+        update_reason="The new proof confirms data exposure.",
+        evidence="The new proof confirms data exposure.",
+    )
+
+    assert len(updated) == 1
+    assert updated[0] is report_state.vulnerability_reports[0]
+
+
+async def test_update_persists_all_report_artifacts(report_state: ReportState) -> None:
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="The chain proves account takeover.",
+        description="The redirect reaches the account takeover flow.",
+        impact="An attacker can take over an account.",
+        cvss_breakdown=_CRITICAL_CVSS,
+    )
+
+    assert result["success"] is True
+    run_dir = report_state.get_run_dir()
+    finding_md = (run_dir / "vulnerabilities" / "vuln-0001.md").read_text(encoding="utf-8")
+    findings = json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8"))
+    sarif = json.loads((run_dir / "findings.sarif").read_text(encoding="utf-8"))
+    sarif_finding = sarif["runs"][0]["results"][0]
+
+    assert "An attacker can take over an account." in finding_md
+    assert findings[0]["impact"] == "An attacker can take over an account."
+    assert findings[0]["severity"] == "critical"
+    assert sarif_finding["properties"]["strix"]["impact"] == (
+        "An attacker can take over an account."
+    )
+    assert sarif_finding["properties"]["strix"]["severity"] == "critical"
+
+
+@pytest.mark.parametrize(
+    "code_locations",
+    [[], [{"file": "../invalid.py", "start_line": 1}]],
+)
+async def test_update_rejects_empty_code_locations(
+    report_state: ReportState,
+    code_locations: list[dict[str, Any]],
+) -> None:
+    original_locations = report_state.vulnerability_reports[0]["code_locations"]
+
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="The source review did not provide a valid location.",
+        code_locations=code_locations,
+    )
+
+    assert result["success"] is False
+    assert any("code_locations" in error for error in result["errors"])
+    assert report_state.vulnerability_reports[0]["code_locations"] == original_locations
+
+
+async def test_update_fails_without_global_report_state(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr("strix.report.state._global_report_state", None)
+
+    result = await _do_update(
+        report_id="vuln-0001",
+        update_reason="The new proof confirms broader impact.",
+        impact="Broader impact.",
+    )
+
+    assert result == {"success": False, "error": "Report state is unavailable"}