feat(tui+memory): add endpoint test memory, Activity & Plan TUI panels

- Add per-scan test_log tool (record_test/query_tests/test_log_summary)
  persisted to {state_dir}/test_log.json with merge logic, agent_history,
  tags, notes; survives --resume via hydrate on runner start
- Wire test_log tools into the base agent toolset and add an
  "ENDPOINT TEST MEMORY (REQUIRED)" block to the system prompt so every
  agent calls query_tests before testing and record_test after
- Propagate agent_name into child contexts so test_log can attribute
  entries to the spawning specialist
- Update scan_modes/{hunter,deep,standard,quick}.md to mandate the
  JS Analysis agent and call out the test memory workflow; add a
  Test-Coverage Memory section to coordination/root_agent.md
- TUI: add Activity panel (left) showing tool calls for the selected
  agent and a Plan/Todos panel (right) reading per-agent todos; wire
  refresh into the live-view tick and agent-selection change; rebalance
  layout widths in tui_styles.tcss

strix/agents/factory.py

@@ -42,6 +42,11 @@ from strix.tools.proxy.tools import (
     view_sitemap_entry,
 )
 from strix.tools.reporting.tool import create_vulnerability_report
+from strix.tools.test_log.tools import (
+    query_tests,
+    record_test,
+    test_log_summary,
+)
 from strix.tools.thinking.tool import think
 from strix.tools.todo.tools import (
     create_todo,
@@ -333,6 +338,9 @@ _BASE_TOOLS: tuple[Tool, ...] = (
     get_note,
     update_note,
     delete_note,
+    record_test,
+    query_tests,
+    test_log_summary,
     web_search,
     create_vulnerability_report,
     list_requests,

strix/agents/prompts/system_prompt.jinja

@@ -283,6 +283,13 @@ ROOT AGENT ROLE:
 8. **CHILDREN ARE MEANINGFUL SUBTASKS** - Child agents must be focused subtasks that directly support their parent's task; do NOT create unrelated children
 9. **UNIQUENESS** - Do not create two agents with the same task; ensure clear, non-overlapping responsibilities for every agent
 
+ENDPOINT TEST MEMORY (REQUIRED):
+- Before testing any endpoint for a vulnerability class, call `query_tests(endpoint=..., vuln_class=...)` to see what prior agents already covered. If another agent confirmed `not_vulnerable` for the same endpoint+class with payloads you would have tried, do NOT re-test — pick a different endpoint or a different class.
+- After each meaningful test attempt, call `record_test(endpoint, vuln_class, outcome, payloads=[short summaries], notes="")`. Use canonical `"METHOD /path/{id}"` form so entries de-duplicate across IDs.
+- Outcomes: `vulnerable` (confirmed with PoC), `not_vulnerable` (tested, nothing found), `needs_more_testing` (partial signal, hand-off), `inconclusive` (couldn't get a clean read), `blocked` (auth/WAF/rate-limit prevented testing).
+- The root agent should call `test_log_summary()` periodically to see coverage gaps and route new specialists toward uncovered surface.
+- This log persists across `--resume`; on resume, every agent must query the log first to avoid re-running work the prior session already did.
+
 WHEN TO CREATE NEW AGENTS:
 
 BLACK-BOX (domain/URL only):

strix/core/execution.py

@@ -506,6 +506,7 @@ async def _start_child_runner(
 
     child_ctx: dict[str, Any] = dict(parent_ctx)
     child_ctx["agent_id"] = child_id
+    child_ctx["agent_name"] = name
     child_ctx["parent_id"] = parent_id
     child_ctx["task"] = task
 

strix/core/runner.py

@@ -103,10 +103,12 @@ async def run_strix_scan(
     coordinator.set_snapshot_path(agents_path)
 
     from strix.tools.notes.tools import hydrate_notes_from_disk
+    from strix.tools.test_log.tools import hydrate_test_log_from_disk
     from strix.tools.todo.tools import hydrate_todos_from_disk
 
     hydrate_todos_from_disk(state_dir)
     hydrate_notes_from_disk(state_dir)
+    hydrate_test_log_from_disk(state_dir)
 
     root_id: str | None = None
     if is_resume:
@@ -211,6 +213,7 @@ async def run_strix_scan(
             "sandbox_session": bundle["session"],
             "caido_client": bundle["caido_client"],
             "agent_id": root_id,
+            "agent_name": "strix",
             "parent_id": None,
             "interactive": interactive,
             "spawn_child_agent": spawn_child_agent,

strix/interface/assets/tui_styles.tcss

@@ -205,7 +205,7 @@ VulnerabilityDetailScreen {
 }
 
 #chat_area_container {
-    width: 80%;
+    width: 44%;
     background: transparent;
 }
 
@@ -213,6 +213,66 @@ VulnerabilityDetailScreen {
     width: 100%;
 }
 
+#activity_panel {
+    width: 18%;
+    background: transparent;
+    margin-right: 1;
+}
+
+#activity_panel.-hidden {
+    display: none;
+}
+
+#activity_scroll {
+    height: 1fr;
+    background: transparent;
+    border: round #333333;
+    border-title-color: #a8a29e;
+    border-title-style: bold;
+    padding: 0 1;
+    scrollbar-background: #000000;
+    scrollbar-color: #1a1a1a;
+    scrollbar-corner-color: #000000;
+    scrollbar-size: 1 1;
+}
+
+#activity_display {
+    height: auto;
+    background: transparent;
+    padding: 0;
+    margin: 0;
+}
+
+#plan_panel {
+    width: 18%;
+    background: transparent;
+    margin-left: 1;
+}
+
+#plan_panel.-hidden {
+    display: none;
+}
+
+#plan_scroll {
+    height: 1fr;
+    background: transparent;
+    border: round #333333;
+    border-title-color: #a8a29e;
+    border-title-style: bold;
+    padding: 0 1;
+    scrollbar-background: #000000;
+    scrollbar-color: #1a1a1a;
+    scrollbar-corner-color: #000000;
+    scrollbar-size: 1 1;
+}
+
+#plan_display {
+    height: auto;
+    background: transparent;
+    padding: 0;
+    margin: 0;
+}
+
 #chat_history {
     height: 1fr;
     background: transparent;

strix/interface/tui/app.py

@@ -846,9 +846,19 @@ class StrixTUIApp(App):  # type: ignore[misc]
 
             vulnerabilities_panel = VulnerabilitiesPanel(id="vulnerabilities_panel")
 
+            activity_display = Static("", id="activity_display")
+            activity_scroll = VerticalScroll(activity_display, id="activity_scroll")
+            activity_panel = Vertical(activity_scroll, id="activity_panel")
+
+            plan_display = Static("", id="plan_display")
+            plan_scroll = VerticalScroll(plan_display, id="plan_scroll")
+            plan_panel = Vertical(plan_scroll, id="plan_panel")
+
             sidebar = Vertical(agents_tree, vulnerabilities_panel, stats_scroll, id="sidebar")
 
+            content_container.mount(activity_panel)
             content_container.mount(chat_area_container)
+            content_container.mount(plan_panel)
             content_container.mount(sidebar)
 
             chat_area_container.mount(chat_history)
@@ -937,6 +947,10 @@ class StrixTUIApp(App):  # type: ignore[misc]
 
         self._update_vulnerabilities_panel()
 
+        self._update_activity_panel()
+
+        self._update_plan_panel()
+
     def _sync_agent_graph(self) -> None:
         future = self._agent_graph_sync_future
         if future is not None:
@@ -1277,6 +1291,157 @@ class StrixTUIApp(App):  # type: ignore[misc]
         self._safe_widget_operation(vuln_panel.remove_class, "hidden")
         vuln_panel.update_vulnerabilities(enriched_vulns)
 
+    def _update_activity_panel(self) -> None:
+        try:
+            display = self.query_one("#activity_display", Static)
+        except (ValueError, Exception):
+            return
+        if not self._is_widget_safe(display):
+            return
+
+        agent_id = self.selected_agent_id
+        if not agent_id or agent_id not in self.live_view.agents:
+            agent_id = next(iter(self.live_view.agents), None)
+
+        text = Text()
+        agent_name = ""
+        if agent_id:
+            agent_name = self.live_view.agents.get(agent_id, {}).get("name", agent_id)
+        text.append("Activity", style="bold #a8a29e")
+        if agent_name:
+            text.append("  ")
+            text.append(agent_name, style="dim")
+        text.append("\n")
+
+        events = self._gather_agent_events(agent_id) if agent_id else []
+        tool_events = [e for e in events if e.get("type") == "tool"]
+
+        if not tool_events:
+            text.append("\n")
+            text.append("no tool calls yet", style="dim")
+            self._safe_widget_operation(display.update, text)
+            return
+
+        recent = tool_events[-200:]
+        status_color = {
+            "completed": "#22c55e",
+            "running": "#eab308",
+            "error": "#ef4444",
+            "failed": "#ef4444",
+        }
+        for event in recent:
+            data = event.get("data") or {}
+            tool_name = str(data.get("tool_name") or "?")
+            status = str(data.get("status") or "running").lower()
+            color = status_color.get(status, "#a8a29e")
+            timestamp = event.get("timestamp", "")
+            ts_suffix = timestamp[11:19] if len(timestamp) >= 19 else ""
+
+            marker = "●" if status == "running" else "✓" if status == "completed" else "✗"
+            text.append(f"{marker} ", style=color)
+            text.append(tool_name, style="white")
+            args = data.get("args")
+            if isinstance(args, dict) and args:
+                summary = self._summarize_tool_args(tool_name, args)
+                if summary:
+                    text.append("  ")
+                    text.append(summary, style="dim")
+            if ts_suffix:
+                text.append(f"  {ts_suffix}", style="#6b7280")
+            text.append("\n")
+
+        self._safe_widget_operation(display.update, text)
+
+    @staticmethod
+    def _summarize_tool_args(tool_name: str, args: dict[str, Any]) -> str:
+        priority_keys = (
+            "endpoint",
+            "url",
+            "path",
+            "target",
+            "command",
+            "vuln_class",
+            "skill",
+            "title",
+            "name",
+            "query",
+        )
+        for key in priority_keys:
+            value = args.get(key)
+            if isinstance(value, str) and value:
+                trimmed = value.strip().replace("\n", " ")
+                if len(trimmed) > 60:
+                    trimmed = trimmed[:57] + "..."
+                return trimmed
+        for value in args.values():
+            if isinstance(value, str) and value:
+                trimmed = value.strip().replace("\n", " ")
+                if len(trimmed) > 60:
+                    trimmed = trimmed[:57] + "..."
+                return trimmed
+        return ""
+
+    def _update_plan_panel(self) -> None:
+        try:
+            display = self.query_one("#plan_display", Static)
+        except (ValueError, Exception):
+            return
+        if not self._is_widget_safe(display):
+            return
+
+        from strix.tools.todo.tools import _sorted_todos, _todos_storage
+
+        agent_id = self.selected_agent_id
+        if not agent_id or agent_id not in self.live_view.agents:
+            agent_id = next(iter(self.live_view.agents), None)
+
+        text = Text()
+        text.append("Plan / Todos", style="bold #a8a29e")
+        if agent_id:
+            agent_name = self.live_view.agents.get(agent_id, {}).get("name", agent_id)
+            text.append("  ")
+            text.append(agent_name, style="dim")
+        text.append("\n")
+
+        if not agent_id:
+            text.append("\nno agent selected", style="dim")
+            self._safe_widget_operation(display.update, text)
+            return
+
+        todos = _sorted_todos(agent_id) if agent_id in _todos_storage else []
+        if not todos:
+            text.append("\nno todos yet", style="dim")
+            self._safe_widget_operation(display.update, text)
+            return
+
+        status_marker = {"done": "[x]", "in_progress": "[~]", "pending": "[ ]"}
+        status_color = {
+            "done": "#22c55e",
+            "in_progress": "#eab308",
+            "pending": "#a8a29e",
+        }
+        priority_color = {
+            "critical": "#ef4444",
+            "high": "#f97316",
+            "normal": "#a8a29e",
+            "low": "#6b7280",
+        }
+        for todo in todos:
+            status = str(todo.get("status") or "pending")
+            priority = str(todo.get("priority") or "normal")
+            marker = status_marker.get(status, "[?]")
+            text.append(marker, style=status_color.get(status, "#a8a29e"))
+            text.append(" ")
+            if priority in ("critical", "high"):
+                text.append(f"({priority[0].upper()}) ", style=priority_color.get(priority))
+            content = str(todo.get("content") or todo.get("title") or "")
+            if len(content) > 200:
+                content = content[:197] + "..."
+            text.append(content, style="white" if status != "done" else "dim")
+            text.append("\n")
+
+        self._safe_widget_operation(display.update, text)
+
     def _get_sweep_animation(self, color_palette: list[str]) -> Text:
         text = Text()
         num_squares = self._sweep_num_squares
@@ -1377,6 +1542,8 @@ class StrixTUIApp(App):  # type: ignore[misc]
         self._displayed_events.clear()
 
         self.call_later(self._update_chat_view)
+        self.call_later(self._update_activity_panel)
+        self.call_later(self._update_plan_panel)
         self._update_agent_status_display()
 
     def _start_scan_thread(self) -> None:

strix/skills/coordination/root_agent.md

@@ -90,3 +90,14 @@ When all agents report completion:
 2. Assess overall security posture
 3. Compile executive summary with prioritized recommendations
 4. Invoke finish tool with final report
+
+## Test-Coverage Memory
+
+A persistent endpoint test log lives in `{state_dir}/test_log.json` and is exposed via three tools: `record_test`, `query_tests`, and `test_log_summary`.
+
+As the coordinator:
+
+- Before spawning a new specialist, run `test_log_summary()` and `query_tests(vuln_class="<class>")` to see whether that surface is already covered.
+- Brief each spawned specialist with the coverage already done so it can `query_tests(endpoint=...)` to avoid duplicate work.
+- On `--resume`, your first action is `test_log_summary()` — it tells you what the prior session finished, what's partial (`needs_more_testing`), and what's untouched.
+- The Vulnerability panel shows confirmed `findings`; the test log records `negative` outcomes too, which the vulnerability panel never will.

strix/skills/scan_modes/deep.md

@@ -165,3 +165,7 @@ Before spawning vulnerability agents, spawn a single `JS Analysis Agent` with `s
 Relentless. Creative. Patient. Thorough. Persistent.
 
 This is about finding what others miss. Test every parameter, every endpoint, every edge case. If one approach fails, try ten more. Understand how components interact to find systemic issues.
+
+## Test Memory
+
+Every endpoint test must be recorded via `record_test` and every endpoint must be checked via `query_tests` before testing — this is how deep mode avoids burning iterations on already-covered ground, especially after `--resume`. The root agent uses `test_log_summary` to spot coverage gaps.

strix/skills/scan_modes/hunter.md

@@ -197,3 +197,4 @@ Before spawning vulnerability agents, spawn a single `JS Analysis Agent` with `s
 - Never test targets not explicitly authorized
 - Never stop mid-scan without a summary of findings
 - Zero hallucination: never write a response you didn't receive
+- **Test memory is mandatory**: call `query_tests` before testing any endpoint and `record_test` after every attempt. On `--resume`, this lets agents skip ground already covered.

strix/skills/scan_modes/quick.md

@@ -70,3 +70,7 @@ When a strong primitive is found (auth weakness, injection point, internal acces
 ## Mindset
 
 Think like a time-boxed bug bounty hunter going for quick wins. Prioritize breadth over depth on critical areas. If something looks exploitable, validate quickly and move on. Don't get stuck—if an attack vector isn't yielding results quickly, pivot.
+
+## Test Memory
+
+Even in quick mode, call `query_tests` before testing an endpoint and `record_test` after — this is the cheapest possible deduplication. On `--resume` it lets you skip directly to surface you haven't touched yet.

strix/skills/scan_modes/standard.md

@@ -101,3 +101,7 @@ When you discover a useful pivot (info leak, weak boundary, partial access), imm
 ## Mindset
 
 Methodical and systematic. Document as you go. Validate everything—no assumptions about exploitability. Think about business impact, not just technical severity.
+
+## Test Memory
+
+Before testing an endpoint, call `query_tests(endpoint=..., vuln_class=...)`. After every test attempt — including negatives — call `record_test(...)`. Other agents (and the resumed session) read this log to skip duplicate work.

strix/tools/test_log/tools.py

@@ -0,0 +1,455 @@
+"""Per-run endpoint test memory — mirrored to {state_dir}/test_log.json.
+
+Tracks which endpoints have been tested for which vulnerability classes,
+which payloads were tried, and what the outcome was. Persists across
+``--resume`` so spawned agents don't re-test ground other agents already
+covered.
+
+Shape on disk::
+
+    {
+      "entries": {
+        "<entry_id>": {
+          "endpoint": "GET /api/v2/orders/{id}",
+          "vuln_class": "idor",
+          "payloads": ["id=123 -> 200 victim data", ...],
+          "outcome": "vulnerable" | "not_vulnerable" | "needs_more_testing",
+          "agent_id": "<agent_id_that_recorded>",
+          "agent_name": "IDOR Agent",
+          "notes": "...",
+          "tags": ["auth", "production"],
+          "created_at": "...",
+          "updated_at": "..."
+        }
+      },
+      "by_endpoint": { "<endpoint>": ["<entry_id>", ...] }
+    }
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import tempfile
+import threading
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from agents import RunContextWrapper, function_tool
+
+
+logger = logging.getLogger(__name__)
+
+
+_VALID_OUTCOMES = {
+    "vulnerable",
+    "not_vulnerable",
+    "needs_more_testing",
+    "inconclusive",
+    "blocked",
+}
+
+_VALID_VULN_CLASSES = {
+    "idor",
+    "ssrf",
+    "xss",
+    "sqli",
+    "rce",
+    "lfi",
+    "rfi",
+    "xxe",
+    "ssti",
+    "auth_bypass",
+    "broken_access_control",
+    "csrf",
+    "open_redirect",
+    "file_upload",
+    "race_condition",
+    "business_logic",
+    "mass_assignment",
+    "nosql_injection",
+    "header_injection",
+    "http_smuggling",
+    "cache_poison",
+    "subdomain_takeover",
+    "info_disclosure",
+    "secrets_exposure",
+    "graphql",
+    "oauth",
+    "saml",
+    "mfa_bypass",
+    "prototype_pollution",
+    "other",
+}
+
+
+_storage: dict[str, dict[str, Any]] = {"entries": {}, "by_endpoint": {}}
+_lock = threading.RLock()
+_path: Path | None = None
+
+
+def hydrate_test_log_from_disk(state_dir: Path) -> None:
+    global _path  # noqa: PLW0603
+    _path = state_dir / "test_log.json"
+    with _lock:
+        _storage["entries"] = {}
+        _storage["by_endpoint"] = {}
+        if not _path.exists():
+            return
+        try:
+            data = json.loads(_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            logger.exception(
+                "test_log.json at %s is unreadable; starting empty",
+                _path,
+            )
+            return
+        if not isinstance(data, dict):
+            return
+        entries = data.get("entries")
+        if isinstance(entries, dict):
+            _storage["entries"] = {
+                eid: e
+                for eid, e in entries.items()
+                if isinstance(eid, str) and isinstance(e, dict)
+            }
+        by_endpoint = data.get("by_endpoint")
+        if isinstance(by_endpoint, dict):
+            _storage["by_endpoint"] = {
+                ep: list(ids)
+                for ep, ids in by_endpoint.items()
+                if isinstance(ep, str) and isinstance(ids, list)
+            }
+        else:
+            for eid, entry in _storage["entries"].items():
+                ep = entry.get("endpoint")
+                if isinstance(ep, str):
+                    _storage["by_endpoint"].setdefault(ep, []).append(eid)
+        logger.info(
+            "test_log hydrated from %s (%d entries)",
+            _path,
+            len(_storage["entries"]),
+        )
+
+
+def _persist() -> None:
+    path = _path
+    if path is None:
+        return
+    try:
+        payload = json.dumps(_storage, ensure_ascii=False, default=str)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with (
+            _lock,
+            tempfile.NamedTemporaryFile(
+                mode="w",
+                encoding="utf-8",
+                dir=str(path.parent),
+                prefix=f".{path.name}.",
+                suffix=".tmp",
+                delete=False,
+            ) as tmp,
+        ):
+            tmp.write(payload)
+            tmp_path = Path(tmp.name)
+        tmp_path.replace(path)
+    except Exception:
+        logger.exception("test_log persist to %s failed", path)
+
+
+def _agent_meta_from(ctx: RunContextWrapper) -> tuple[str, str]:
+    inner = ctx.context if isinstance(ctx.context, dict) else {}
+    agent_id = str(inner.get("agent_id") or "unknown")
+    agent_name = str(inner.get("agent_name") or agent_id)
+    return agent_id, agent_name
+
+
+def _normalize_endpoint(endpoint: str) -> str:
+    return endpoint.strip()
+
+
+def _record_test_impl(
+    endpoint: str,
+    vuln_class: str,
+    outcome: str,
+    payloads: list[str] | None,
+    notes: str,
+    tags: list[str] | None,
+    agent_id: str,
+    agent_name: str,
+) -> dict[str, Any]:
+    with _lock:
+        if not endpoint or not endpoint.strip():
+            return {"success": False, "error": "endpoint cannot be empty"}
+        vc = vuln_class.strip().lower()
+        if vc not in _VALID_VULN_CLASSES:
+            return {
+                "success": False,
+                "error": (
+                    f"unknown vuln_class '{vuln_class}'. valid: "
+                    f"{sorted(_VALID_VULN_CLASSES)}"
+                ),
+            }
+        oc = outcome.strip().lower()
+        if oc not in _VALID_OUTCOMES:
+            return {
+                "success": False,
+                "error": (
+                    f"unknown outcome '{outcome}'. valid: {sorted(_VALID_OUTCOMES)}"
+                ),
+            }
+
+        ep = _normalize_endpoint(endpoint)
+        timestamp = datetime.now(UTC).isoformat()
+
+        existing_id: str | None = None
+        for eid in _storage["by_endpoint"].get(ep, []):
+            entry = _storage["entries"].get(eid)
+            if entry and entry.get("vuln_class") == vc:
+                existing_id = eid
+                break
+
+        if existing_id is not None:
+            entry = _storage["entries"][existing_id]
+            if payloads:
+                merged = list(entry.get("payloads") or [])
+                for p in payloads:
+                    if p and p not in merged:
+                        merged.append(p)
+                entry["payloads"] = merged
+            if notes:
+                prior = entry.get("notes") or ""
+                if notes not in prior:
+                    entry["notes"] = (prior + "\n" + notes).strip() if prior else notes
+            if tags:
+                merged_tags = list(entry.get("tags") or [])
+                for t in tags:
+                    if t and t not in merged_tags:
+                        merged_tags.append(t)
+                entry["tags"] = merged_tags
+            entry["outcome"] = oc
+            entry["updated_at"] = timestamp
+            entry.setdefault("agent_history", []).append(
+                {"agent_id": agent_id, "agent_name": agent_name, "at": timestamp}
+            )
+            _persist()
+            return {
+                "success": True,
+                "entry_id": existing_id,
+                "merged": True,
+                "endpoint": ep,
+                "vuln_class": vc,
+                "outcome": oc,
+                "total_entries": len(_storage["entries"]),
+            }
+
+        entry_id = uuid.uuid4().hex[:8]
+        entry = {
+            "endpoint": ep,
+            "vuln_class": vc,
+            "payloads": list(payloads or []),
+            "outcome": oc,
+            "notes": notes or "",
+            "tags": list(tags or []),
+            "agent_id": agent_id,
+            "agent_name": agent_name,
+            "agent_history": [
+                {"agent_id": agent_id, "agent_name": agent_name, "at": timestamp}
+            ],
+            "created_at": timestamp,
+            "updated_at": timestamp,
+        }
+        _storage["entries"][entry_id] = entry
+        _storage["by_endpoint"].setdefault(ep, []).append(entry_id)
+        _persist()
+        return {
+            "success": True,
+            "entry_id": entry_id,
+            "merged": False,
+            "endpoint": ep,
+            "vuln_class": vc,
+            "outcome": oc,
+            "total_entries": len(_storage["entries"]),
+        }
+
+
+def _query_tests_impl(
+    endpoint: str | None,
+    vuln_class: str | None,
+    outcome: str | None,
+    agent_id: str | None,
+    tag: str | None,
+    limit: int,
+) -> dict[str, Any]:
+    with _lock:
+        results: list[dict[str, Any]] = []
+        ep_norm = _normalize_endpoint(endpoint) if endpoint else None
+        vc_norm = vuln_class.strip().lower() if vuln_class else None
+        oc_norm = outcome.strip().lower() if outcome else None
+
+        for eid, entry in _storage["entries"].items():
+            if ep_norm and ep_norm not in entry.get("endpoint", ""):
+                continue
+            if vc_norm and entry.get("vuln_class") != vc_norm:
+                continue
+            if oc_norm and entry.get("outcome") != oc_norm:
+                continue
+            if agent_id and entry.get("agent_id") != agent_id:
+                continue
+            if tag and tag not in (entry.get("tags") or []):
+                continue
+            results.append({"entry_id": eid, **entry})
+
+        results.sort(key=lambda r: r.get("updated_at", ""), reverse=True)
+        truncated = len(results) > limit
+        return {
+            "success": True,
+            "entries": results[:limit],
+            "returned": min(len(results), limit),
+            "matched": len(results),
+            "total_in_log": len(_storage["entries"]),
+            "truncated": truncated,
+        }
+
+
+def _summary_impl() -> dict[str, Any]:
+    with _lock:
+        by_class: dict[str, int] = {}
+        by_outcome: dict[str, int] = {}
+        for entry in _storage["entries"].values():
+            by_class[entry.get("vuln_class", "?")] = (
+                by_class.get(entry.get("vuln_class", "?"), 0) + 1
+            )
+            by_outcome[entry.get("outcome", "?")] = (
+                by_outcome.get(entry.get("outcome", "?"), 0) + 1
+            )
+        return {
+            "success": True,
+            "total_entries": len(_storage["entries"]),
+            "unique_endpoints": len(_storage["by_endpoint"]),
+            "by_vuln_class": by_class,
+            "by_outcome": by_outcome,
+        }
+
+
+@function_tool(timeout=30)
+async def record_test(
+    ctx: RunContextWrapper,
+    endpoint: str,
+    vuln_class: str,
+    outcome: str,
+    payloads: list[str] | None = None,
+    notes: str = "",
+    tags: list[str] | None = None,
+) -> str:
+    """Record that you tested an endpoint for a vulnerability class.
+
+    Call this **after every meaningful test attempt** so other agents
+    (and this same agent after a ``--resume``) can see what's already
+    been covered and skip duplicate work.
+
+    If an entry for the same ``endpoint`` + ``vuln_class`` already
+    exists, this **merges** into it: payloads and tags are
+    de-duplicated and appended, notes are concatenated, and ``outcome``
+    is overwritten with the new value (so re-testing can promote
+    ``needs_more_testing`` to ``vulnerable``).
+
+    Args:
+        endpoint: Canonical request identifier. Prefer
+            ``"METHOD /path"`` (e.g. ``"GET /api/v2/orders/{id}"``).
+            Keep path-param shapes (``{id}``) instead of concrete
+            values so the entry de-duplicates across IDs.
+        vuln_class: One of the supported classes (``idor``, ``ssrf``,
+            ``xss``, ``sqli``, ``rce``, ``lfi``, ``rfi``, ``xxe``,
+            ``ssti``, ``auth_bypass``, ``broken_access_control``,
+            ``csrf``, ``open_redirect``, ``file_upload``,
+            ``race_condition``, ``business_logic``,
+            ``mass_assignment``, ``nosql_injection``,
+            ``header_injection``, ``http_smuggling``,
+            ``cache_poison``, ``subdomain_takeover``,
+            ``info_disclosure``, ``secrets_exposure``, ``graphql``,
+            ``oauth``, ``saml``, ``mfa_bypass``,
+            ``prototype_pollution``, ``other``).
+        outcome: One of ``vulnerable``, ``not_vulnerable``,
+            ``needs_more_testing``, ``inconclusive``, ``blocked``.
+        payloads: Short list of payload summaries you actually sent,
+            each ideally one line including the response signal
+            (``"id=2 -> 200 victim PII returned"``). Skip raw
+            multi-line HTTP — keep it scannable.
+        notes: Free-form notes that the next agent would benefit from
+            (auth required, WAF behavior, rate limit observed, etc.).
+        tags: Free-form tags for later filtering
+            (e.g. ``["auth", "production", "v2-api"]``).
+    """
+    agent_id, agent_name = _agent_meta_from(ctx)
+    return json.dumps(
+        await asyncio.to_thread(
+            _record_test_impl,
+            endpoint,
+            vuln_class,
+            outcome,
+            payloads,
+            notes,
+            tags,
+            agent_id,
+            agent_name,
+        ),
+        ensure_ascii=False,
+        default=str,
+    )
+
+
+@function_tool(timeout=30)
+async def query_tests(
+    ctx: RunContextWrapper,
+    endpoint: str | None = None,
+    vuln_class: str | None = None,
+    outcome: str | None = None,
+    agent_id: str | None = None,
+    tag: str | None = None,
+    limit: int = 50,
+) -> str:
+    """Look up prior test attempts before spending time re-testing.
+
+    Call this **before** any vulnerability testing on an endpoint —
+    another agent (or yourself in a prior run) may have already covered
+    the same surface. Filters compose with AND semantics.
+
+    Args:
+        endpoint: Substring match against recorded endpoint strings
+            (e.g. ``"/api/v2/orders"`` matches
+            ``"GET /api/v2/orders/{id}"``).
+        vuln_class: Exact match against vuln class.
+        outcome: Exact match against outcome.
+        agent_id: Filter to a specific recording agent.
+        tag: Match entries that include this tag.
+        limit: Cap on returned entries (default 50).
+    """
+    return json.dumps(
+        await asyncio.to_thread(
+            _query_tests_impl,
+            endpoint,
+            vuln_class,
+            outcome,
+            agent_id,
+            tag,
+            max(1, min(limit, 500)),
+        ),
+        ensure_ascii=False,
+        default=str,
+    )
+
+
+@function_tool(timeout=30)
+async def test_log_summary(ctx: RunContextWrapper) -> str:
+    """Coverage snapshot: how many endpoints have been tested, broken
+    down by vuln class and outcome. Useful for the root agent to decide
+    where to spawn the next specialist."""
+    return json.dumps(
+        await asyncio.to_thread(_summary_impl),
+        ensure_ascii=False,
+        default=str,
+    )