fix: prevent premature subagent termination, add attack planning framework, and fix TUI lag

Subagent stop bug:
- Remove stop_agent as suggested option in finish_scan docstring and rejection message
- Rewrite root_agent.md Completion section to require waiting for all agents
- Add warning to stop_agent docstring against using it to clear active agents

Attack planning:
- Add Attack Planning section to root_agent.md with think + load_skill framework
- Add ATTACK PLANNING METHODOLOGY to system_prompt.jinja vulnerability_focus
- Replace brute-force decomposition with technology-aware agent spawning

TUI lag:
- Batch SDK events via 150ms flush timer instead of per-event call_from_thread
- Index events by agent_id for O(1) lookup instead of O(n) list scan
- Cap events per agent at 500 with automatic pruning of oldest events
- Window event rendering to last 200 events instead of full re-render
- Reduce animation timer from 60ms to 150ms
This commit is contained in:
Hafzan Ishigaki 2026-07-06 14:38:55 +08:00
parent 302efedca6
commit b801c3d208
6 changed files with 111 additions and 19 deletions

View file

@ -217,6 +217,15 @@ VALIDATION APPROACH:
- Chain vulnerabilities when needed to demonstrate maximum impact
- Focus on demonstrating real business impact
ATTACK PLANNING METHODOLOGY:
Before testing a vulnerability type, reason deeply about the attack — do not spray payloads blindly.
1. **Load the skill first**: Call `load_skill` with the relevant vulnerability name (e.g., `["sql_injection"]`, `["xss"]`, `["ssrf"]`, `["business_logic"]`) to get the full methodology — attack surfaces, detection channels, bypass techniques, technology-specific primitives, and testing workflow. Do not plan attacks from memory alone.
2. **Use `think` to plan**: Before acting, reason through: target technology and how it handles input/sessions/auth → which endpoints and parameters are user-controlled → which techniques are most likely to succeed given the technology → what defenses (WAF, validation, encoding) are in place and how to bypass them → what encoding tricks or syntax variations could work → whether findings can be chained for greater impact
3. **Match techniques to technology**: Do not test irrelevant attack vectors. NoSQL injection is pointless against PostgreSQL. XXE is irrelevant if no XML parsing occurs. SSRF requires server-side URL fetching. Select techniques based on what the target actually does.
4. **Plan bypass strategies**: When standard payloads are filtered, reason about alternatives — encoding variations (double URL, unicode, NFKC), syntax tricks (comment splitting, case folding), alternative injection points (headers, cookies, path params), and protocol-level bypasses.
5. **Chain for impact**: Think about how low-severity findings combine into high-impact attacks. IDOR + business logic = financial fraud. SSRF + cloud metadata = credential theft. XSS + CSRF = account takeover. Stored XSS + admin viewing = privilege escalation.
VULNERABILITY KNOWLEDGE BASE:
You have access to comprehensive guides for each vulnerability type above. Use these references for:
- Discovery techniques and automation
@ -259,8 +268,10 @@ WHITE-BOX TESTING - PHASE 1 (CODE UNDERSTANDING):
- ONLY AFTER full code comprehension → proceed to vulnerability testing
PHASE 2 - SYSTEMATIC VULNERABILITY TESTING:
- CREATE SPECIALIZED SUBAGENT for EACH vulnerability type × EACH component
- Each agent focuses on ONE vulnerability type in ONE specific location
- ANALYZE the target's technology stack and attack surface to determine which vulnerability types are relevant
- LOAD the relevant vulnerability skill via `load_skill` and use `think` to plan the attack strategy before spawning agents
- CREATE SPECIALIZED SUBAGENTS for vulnerability types that match the target's technology and attack surface — not blindly for every category
- Each agent focuses on ONE vulnerability type in ONE specific location with a task description informed by your analysis
- EVERY detected vulnerability MUST spawn its own validation subagent
SIMPLE WORKFLOW RULES:

View file

@ -730,6 +730,9 @@ class StrixTUIApp(App): # type: ignore[misc]
]
self._dot_animation_timer: Any | None = None
self._event_buffer: list[tuple[str, Any]] = []
self._event_flush_timer: Any | None = None
self._setup_cleanup_handlers()
def _build_scan_config(self, args: argparse.Namespace) -> dict[str, Any]:
@ -1292,7 +1295,7 @@ class StrixTUIApp(App): # type: ignore[misc]
def _start_dot_animation(self) -> None:
if self._dot_animation_timer is None:
self._dot_animation_timer = self.set_interval(0.06, self._animate_dots)
self._dot_animation_timer = self.set_interval(0.15, self._animate_dots)
def _stop_dot_animation(self) -> None:
if self._dot_animation_timer is not None:
@ -1335,9 +1338,13 @@ class StrixTUIApp(App): # type: ignore[misc]
if vuln.get("agent_id") == agent_id
)
MAX_RENDER_EVENTS = 200
def _gather_agent_events(self, agent_id: str) -> list[dict[str, Any]]:
events = self.live_view.events_for_agent(agent_id)
events.sort(key=lambda e: (e["timestamp"], e["id"]))
if len(events) > self.MAX_RENDER_EVENTS:
events = events[-self.MAX_RENDER_EVENTS:]
return events
def watch_selected_agent_id(self, _agent_id: str | None) -> None:
@ -1407,10 +1414,25 @@ class StrixTUIApp(App): # type: ignore[misc]
self._scan_thread.start()
def _capture_sdk_event(self, agent_id: str, event: Any) -> None:
self._event_buffer.append((agent_id, event))
if self._event_flush_timer is None:
self._event_flush_timer = self.set_timer(0.15, self._flush_event_buffer)
def _flush_event_buffer(self) -> None:
self._event_flush_timer = None
if not self._event_buffer:
return
batch = self._event_buffer[:]
self._event_buffer.clear()
try:
self.call_from_thread(self._record_sdk_event, agent_id, event)
self.call_from_thread(self._record_sdk_events_batch, batch)
except RuntimeError:
self._record_sdk_event(agent_id, event)
for agent_id, event in batch:
self._record_sdk_event(agent_id, event)
def _record_sdk_events_batch(self, batch: list[tuple[str, Any]]) -> None:
for agent_id, event in batch:
self.live_view.ingest_sdk_event(agent_id, event)
def _record_sdk_event(self, agent_id: str, event: Any) -> None:
self.live_view.ingest_sdk_event(agent_id, event)

View file

@ -14,10 +14,14 @@ from strix.core.paths import runtime_state_dir
from strix.interface.tui.history import load_session_history
MAX_EVENTS_PER_AGENT = 500
class TuiLiveView:
def __init__(self) -> None:
self.agents: dict[str, dict[str, Any]] = {}
self.events: list[dict[str, Any]] = []
self._events_by_agent: dict[str, list[dict[str, Any]]] = {}
self._next_event_id = 1
self._open_assistant_event_by_agent: dict[str, dict[str, Any]] = {}
self._tool_event_by_call_id: dict[str, dict[str, Any]] = {}
@ -115,10 +119,10 @@ class TuiLiveView:
self._record_tool_output(agent_id, item)
def events_for_agent(self, agent_id: str) -> list[dict[str, Any]]:
return [event for event in self.events if event.get("agent_id") == agent_id]
return list(self._events_by_agent.get(agent_id, []))
def has_events_for_agent(self, agent_id: str) -> bool:
return any(event.get("agent_id") == agent_id for event in self.events)
return bool(self._events_by_agent.get(agent_id))
def _ingest_raw_response_event(self, agent_id: str, data: Any) -> None:
data_type = getattr(data, "type", "")
@ -277,6 +281,16 @@ class TuiLiveView:
}
self._next_event_id += 1
self.events.append(event)
agent_events = self._events_by_agent.setdefault(agent_id, [])
agent_events.append(event)
if len(agent_events) > MAX_EVENTS_PER_AGENT:
dropped = agent_events[:len(agent_events) - MAX_EVENTS_PER_AGENT]
for d in dropped:
try:
self.events.remove(d)
except ValueError:
pass
del agent_events[:len(agent_events) - MAX_EVENTS_PER_AGENT]
return event
@staticmethod

View file

@ -25,6 +25,30 @@ Before spawning agents, analyze the target:
3. **Determine approach** - blackbox, greybox, or whitebox assessment
4. **Prioritize by risk** - critical assets and high-value targets first
## Attack Planning
Before spawning subagents for a vulnerability type, **think deeply about the attack strategy**. Do not blindly spawn agents for every vulnerability category — reason about what is actually relevant to the target.
### Step 1: Load Vulnerability Methodology
Before planning an attack against a specific vulnerability type, call `load_skill` with the relevant skill name (e.g., `["sql_injection"]`, `["xss"]`, `["business_logic"]`, `["ssrf"]`, `["xxe"]`). This gives you the full methodology — attack surfaces, detection channels, bypass techniques, DBMS-specific primitives, testing workflow, and validation steps. **Do not plan attacks from memory alone — load the skill first.**
### Step 2: Reason About the Attack
Use `think` to reason through the attack plan before spawning agents. Structure your thinking:
- **Target technology**: What framework, language, database, server is in use? How does it handle input, auth, sessions?
- **Attack surface mapping**: Which endpoints, parameters, headers, cookies are user-controlled? Where does data flow to (database, filesystem, OS, other services)?
- **Technique selection**: Given the technology, which attack techniques are most likely to succeed? Which are irrelevant? (e.g., NoSQL injection is irrelevant for a PostgreSQL backend; XXE is irrelevant if no XML parsing occurs)
- **Defense analysis**: What WAF, input validation, encoding, or framework protections are in place? How might they be bypassed?
- **Bypass strategy**: If standard payloads are filtered, what encoding tricks, syntax variations, or alternative injection points could work?
- **Chaining opportunities**: Can findings be combined for greater impact? (e.g., IDOR + business logic = financial fraud; SSRF + metadata = cloud compromise; XSS + CSRF = account takeover)
- **Priority ordering**: Which attacks should be attempted first based on likelihood of success and potential impact?
### Step 3: Spawn Targeted Agents
Only after loading the skill and reasoning through the attack, spawn subagents with specific, informed task descriptions. The task description should reflect your analysis — tell the subagent what technology to target, what techniques to prioritize, what bypasses to try, and what the success criteria are. A well-planned task description produces far better results than a generic "test for SQLi."
## Agent Architecture
Structure agents by function:
@ -84,9 +108,17 @@ Complex findings warrant specialized subagents:
## Completion
When all agents report completion:
**WAIT for all agents to self-terminate.** Do NOT call `stop_agent` on active children to clear them before finishing — this orphans their work and produces an incomplete report.
1. Collect and deduplicate findings across agents
2. Assess overall security posture
3. Compile executive summary with prioritized recommendations
4. Invoke finish tool with final report
When you believe all work is done:
1. Call `view_agent_graph` to check every agent's status
2. If ANY agent is still `running` or `waiting`, call `wait_for_message` to block until their completion reports arrive — do NOT stop them
3. Only proceed when ALL agents show `completed` or `crashed` (these are the only safe terminal states)
4. `stopped` agents were forcibly cancelled — their results are lost. If agents were stopped, their work must be re-done by spawning replacement agents
5. Collect and deduplicate findings across all completion reports
6. Assess overall security posture
7. Compile executive summary with prioritized recommendations
8. Invoke `finish_scan` with the final report
**Never use `stop_agent` as a shortcut to bypass the active-agent check in `finish_scan`.** The check exists to prevent incomplete reports. If a child is taking too long, send it a message asking for status or telling it to wrap up via `agent_finish` — don't stop it.

View file

@ -595,6 +595,15 @@ async def stop_agent(
to wrap up) for soft-stop scenarios. Reach for ``stop_agent`` when
a child has gone off-track and won't self-correct.
**NEVER use ``stop_agent`` to clear active children so you can call
``finish_scan``.** Stopped agents lose their in-progress work and
their completion reports never arrive. Instead, call
``wait_for_message`` to block until children self-terminate via
``agent_finish``, or ``send_message_to_agent`` asking them to wrap
up. The only valid use of ``stop_agent`` is for agents that are
genuinely off-track, stuck, or producing errors — not for agents
that are productively working but haven't finished yet.
Args:
target_agent_id: The 8-char id from ``view_agent_graph`` /
``create_agent``. Cannot stop yourself.

View file

@ -101,12 +101,13 @@ async def finish_scan(
1. **Call ``view_agent_graph`` first.** Inspect every entry in the
summary. If ANY agent is in ``running`` / ``waiting`` state,
you MUST NOT call ``finish_scan`` yet —
wrap them up first via ``send_message_to_agent`` (ask them to
finish), ``wait_for_message`` (block until their report
arrives), or ``stop_agent`` (graceful cancel). Only ``completed``
/ ``crashed`` / ``stopped`` agents are safe to leave behind.
Calling ``finish_scan`` while children are alive orphans their
work and produces an incomplete report.
call ``wait_for_message`` to block until their completion reports
arrive, or ``send_message_to_agent`` asking them to call
``agent_finish``. **NEVER use ``stop_agent`` to clear active
agents so you can finish — this orphans their work and produces
an incomplete report.** Only ``completed`` / ``crashed`` agents
are safe to leave behind. ``stopped`` agents were forcibly
cancelled and their results are lost.
2. All vulnerabilities you found are filed via
``create_vulnerability_report`` (un-reported findings are not
tracked and not credited).
@ -162,7 +163,10 @@ async def finish_scan(
"scan_completed": False,
"error": (
"Cannot finish scan while child agents are still active. "
"Wait for completion, send them finish instructions, or stop them first"
"Call wait_for_message to block until their completion reports "
"arrive, or send_message_to_agent asking them to call agent_finish. "
"Do NOT use stop_agent to clear active agents — their work will "
"be lost and the report will be incomplete."
),
"active_agents": active_agents,
},