Multi-agent foundations (Phases 1+2): dynamic attack surface + revealed_targets

Phase 1 — schema additions (no behavior change yet, all 78 tests still pass):
- execute_tool() now returns 5-tuple (output, new_hosts, new_ports, reward, revealed_targets)
with backward-compat: 4-tuple handlers get padded to 5-tuple at the engine boundary
- env tracks _attack_surface (grows during episode), _revealed_targets log,
_active_subagents dict, _subagent_outcomes log
- New action_types accepted: spawn_subagent / return_to_parent (stubs, real impl in Phase 3)
- Tests updated: 3 test_generator.py unpack sites bumped to 5-tuple

Phase 2 — tools emit revealed_targets when they hit gateway vulns:
- New _revelations_for_vulns helper in tools_engine/testing.py: when a tool finds a
vulnerability whose ID gates a hidden host (via scenario hosts.hidden_until chain),
emit a {scope: host, target: <ip>, context: ...} revelation
- handle_test_injection / handle_test_xss / handle_test_auth surface revelations both
in their textual output ("[REVEALED] Sub-agent delegation candidates: ...") AND
via the 5th tuple element so the env can update attack_surface
- Verified end-to-end: test_injection on medium scenario's SSRF correctly reveals
the two hidden internal hosts 10.0.2.30 and 10.0.2.40

This is the foundation — Phases 3-8 wire actual sub-agent spawning, nested
inference loop, reward shaping, grader extension, and live deploy.

server/security_audit_env_environment.py

@@ -64,6 +64,15 @@ class SecurityAuditEnvironment(Environment):
         self._episode_reward: float = 0.0
         self._last_tool_call: tuple = ()
         self._rng: random.Random = random.Random()
+        # Multi-agent extension: dynamic attack surface + sub-agent registry.
+        # ``_attack_surface`` starts as the original scenario hosts and grows
+        # whenever tools reveal new targets (e.g. SSRF dumping internal IPs).
+        # ``_active_subagents`` tracks delegated investigation branches keyed
+        # by spawn_id; each entry carries scope/target/budget/findings/parent_step.
+        self._attack_surface: set = set()
+        self._revealed_targets: list = []
+        self._active_subagents: dict = {}
+        self._subagent_outcomes: list = []  # for grader's Delegation Score
 
     def reset(self, seed=None, episode_id=None, **kwargs) -> SecurityAuditObservation:
         """Reset the environment for a new audit engagement.
@@ -83,6 +92,12 @@ class SecurityAuditEnvironment(Environment):
         self._episode_reward = 0.0
         self._last_tool_call = ()
         self._rng = random.Random(seed) if seed is not None else random.Random()
+        # Reset multi-agent state. Attack surface seeds from initial scenario
+        # hosts; will grow as tools reveal targets during the episode.
+        self._attack_surface = set((self._scenario or {}).get("hosts", {}).keys())
+        self._revealed_targets = []
+        self._active_subagents = {}
+        self._subagent_outcomes = []
 
         eid = episode_id or str(uuid4())
         self._state = SecurityAuditState(
@@ -127,12 +142,16 @@ class SecurityAuditEnvironment(Environment):
             return self._handle_use_tool(action, steps_remaining)
         elif action.action_type == "submit_finding":
             return self._handle_submit_finding(action, steps_remaining)
+        elif action.action_type == "spawn_subagent":
+            return self._handle_spawn_subagent(action, steps_remaining)
+        elif action.action_type == "return_to_parent":
+            return self._handle_return_to_parent(action, steps_remaining)
         elif action.action_type == "generate_report":
             return self._finish_episode("Audit report generated.", truncated=False)
         else:
             return SecurityAuditObservation(
                 tool_output=f"Unknown action_type: {action.action_type}",
-                message="Use list_tools, use_tool, submit_finding, or generate_report.",
+                message="Use list_tools, use_tool, submit_finding, spawn_subagent, or generate_report.",
                 discovered_hosts=self._discovered_hosts,
                 discovered_services=self._discovered_services,
                 findings_submitted=len(self._submitted_findings),
@@ -142,6 +161,34 @@ class SecurityAuditEnvironment(Environment):
                 reward=-0.05,
             )
 
+    def _handle_spawn_subagent(self, action: SecurityAuditAction, steps_remaining: int) -> SecurityAuditObservation:
+        """Phase 1 stub. Real registration + budget tracking lands in Phase 3."""
+        return SecurityAuditObservation(
+            tool_output="spawn_subagent acknowledged (full implementation in Phase 3).",
+            message="Sub-agent infra is being wired up; this action is a no-op for now.",
+            discovered_hosts=self._discovered_hosts,
+            discovered_services=self._discovered_services,
+            findings_submitted=len(self._submitted_findings),
+            steps_remaining=steps_remaining,
+            current_phase=self._current_phase(),
+            done=False,
+            reward=0.0,
+        )
+
+    def _handle_return_to_parent(self, action: SecurityAuditAction, steps_remaining: int) -> SecurityAuditObservation:
+        """Phase 1 stub. Sub-agent termination lands in Phase 3."""
+        return SecurityAuditObservation(
+            tool_output="return_to_parent acknowledged (no active sub-agent context).",
+            message="Use this when finishing a sub-agent investigation.",
+            discovered_hosts=self._discovered_hosts,
+            discovered_services=self._discovered_services,
+            findings_submitted=len(self._submitted_findings),
+            steps_remaining=steps_remaining,
+            current_phase=self._current_phase(),
+            done=False,
+            reward=0.0,
+        )
+
     @property
     def state(self) -> SecurityAuditState:
         self._state.discovered_hosts = list(self._discovered_hosts)
@@ -195,11 +242,19 @@ class SecurityAuditEnvironment(Environment):
         redundancy_penalty = -0.01 if current_call == self._last_tool_call else 0.0
         self._last_tool_call = current_call
 
-        output, new_hosts, new_ports, tool_reward = execute_tool(
+        output, new_hosts, new_ports, tool_reward, revealed = execute_tool(
             action.tool_name, action.arguments, self._scenario,
             self._discovered_hosts, self._discovered_ports, self._discovered_vulns,
         )
 
+        # Phase 1: any new revelations expand the dynamic attack_surface so
+        # later spawn_subagent actions can validate their target.
+        for r in revealed:
+            tgt = r.get("target")
+            if tgt and tgt not in self._attack_surface:
+                self._attack_surface.add(tgt)
+                self._revealed_targets.append({**r, "revealed_at_step": self._state.step_count})
+
         # Difficulty multiplier on positive rewards
         difficulty = self._scenario.get("id", "easy")
         multiplier = self._DIFFICULTY_REWARD_MULTIPLIER.get(difficulty, 1.0)

server/tools_engine/engine.py

@@ -122,12 +122,28 @@ def execute_tool(
     discovered_hosts: List[str],
     discovered_ports: Dict[str, List[int]],
     discovered_vulns: Optional[Set[str]] = None,
-) -> Tuple[str, List[str], Dict[str, List[int]], float]:
-    """Execute a simulated tool and return (output, new_hosts, new_ports, reward).
+) -> Tuple[str, List[str], Dict[str, List[int]], float, List[Dict[str, Any]]]:
+    """Execute a simulated tool and return:
 
-    This is a drop-in replacement for the old ``tools.execute_tool``.
+    ``(output, new_hosts, new_ports, reward, revealed_targets)``
+
+    ``revealed_targets`` is a list of dicts shaped like
+    ``{"scope": "host|endpoint|cred", "target": "10.0.2.30", "context": "..."}``.
+    Tools emit revelations when their output discloses a follow-up surface
+    (e.g. SSRF dumping an internal IP) — the agent can use ``spawn_subagent``
+    to delegate investigation of those branches without losing the main thread.
+
+    Handlers that haven't yet been updated to return a 5-tuple will have an
+    empty ``revealed_targets`` list back-filled here, so this stays a drop-in
+    replacement for the old 4-tuple contract during Phase 1.
     """
     handler = TOOL_HANDLERS.get(tool_name)
     if not handler:
-        return (f"Error: Unknown tool '{tool_name}'. Use list_tools to see available tools.", [], {}, -0.05)
-    return handler(arguments, scenario, discovered_hosts, discovered_ports, discovered_vulns)
+        return (
+            f"Error: Unknown tool '{tool_name}'. Use list_tools to see available tools.",
+            [], {}, -0.05, [],
+        )
+    result = handler(arguments, scenario, discovered_hosts, discovered_ports, discovered_vulns)
+    if len(result) == 4:  # legacy 4-tuple — pad with empty revelations
+        return (*result, [])
+    return result

server/tools_engine/testing.py

@@ -47,6 +47,47 @@ def _normalize_difficulty(scenario: Dict[str, Any]) -> str:
     return "easy"
 
 
+def _revelations_for_vulns(
+    matched: List[Dict[str, Any]],
+    scenario: Dict[str, Any],
+    discovered_hosts: List[str],
+) -> List[Dict[str, Any]]:
+    """Emit `revealed_targets` for vulns that gate progressive-discovery hosts.
+
+    When a tool finds a vulnerability whose ID appears in some host's
+    `hidden_until` chain, that host becomes investigatable — but the agent
+    might not yet have submitted the finding (which is what officially adds
+    the host to discovered_hosts). Surfacing it via revealed_targets lets
+    the agent **delegate** the branch to a sub-agent without breaking its
+    primary focus.
+    """
+    if not matched:
+        return []
+    hosts = scenario.get("hosts", {})
+    matched_ids = {v.get("id") for v in matched if v.get("id")}
+    out: List[Dict[str, Any]] = []
+    for ip, info in hosts.items():
+        if ip in discovered_hosts:
+            continue
+        gates = info.get("hidden_until") or []
+        if not gates:
+            continue
+        triggers = [g for g in gates if g in matched_ids]
+        if not triggers:
+            continue
+        out.append({
+            "scope": "host",
+            "target": ip,
+            "context": (
+                f"Investigation gateway: vulnerability {triggers[0]} indicates "
+                f"reachable internal host {ip}. Consider spawning a sub-agent "
+                f"to recon/exploit it without losing the main thread."
+            ),
+            "trigger_vuln_ids": triggers,
+        })
+    return out
+
+
 def _build_context(
     vuln: Dict[str, Any],
     host: str,
@@ -105,7 +146,7 @@ def handle_test_injection(
             f"  Tested: SQLi (error-based, blind, time-based), command injection, SSTI, SSRF\n"
             f"  Payloads: 47 injection patterns tested\n"
             f"  Result: No injection vulnerabilities detected on this endpoint.",
-            [], {}, 0.01,
+            [], {}, 0.01, [],
         )
 
     output_parts = [f"Injection testing on {host}{endpoint}:", ""]
@@ -114,7 +155,14 @@ def handle_test_injection(
         output_parts.append(format_tool_output(v, difficulty, context))
         output_parts.append("")
 
-    return "\n".join(output_parts), [], {}, 0.08
+    revelations = _revelations_for_vulns(vulns, scenario, discovered_hosts)
+    if revelations:
+        output_parts.append("[REVEALED] Sub-agent delegation candidates:")
+        for r in revelations:
+            output_parts.append(f"  - scope={r['scope']} target={r['target']}  ({r['context']})")
+        output_parts.append("")
+
+    return "\n".join(output_parts), [], {}, 0.08, revelations
 
 
 # ---------------------------------------------------------------------------
@@ -160,7 +208,14 @@ def handle_test_xss(
         output_parts.append(format_tool_output(v, difficulty, context))
         output_parts.append("")
 
-    return "\n".join(output_parts), [], {}, 0.08
+    revelations = _revelations_for_vulns(vulns, scenario, discovered_hosts)
+    if revelations:
+        output_parts.append("[REVEALED] Sub-agent delegation candidates:")
+        for r in revelations:
+            output_parts.append(f"  - scope={r['scope']} target={r['target']}  ({r['context']})")
+        output_parts.append("")
+
+    return "\n".join(output_parts), [], {}, 0.08, revelations
 
 
 # ---------------------------------------------------------------------------
@@ -211,7 +266,14 @@ def handle_test_auth(
         output_parts.append(format_tool_output(v, difficulty, context))
         output_parts.append("")
 
-    return "\n".join(output_parts), [], {}, 0.08
+    revelations = _revelations_for_vulns(vulns, scenario, discovered_hosts)
+    if revelations:
+        output_parts.append("[REVEALED] Sub-agent delegation candidates:")
+        for r in revelations:
+            output_parts.append(f"  - scope={r['scope']} target={r['target']}  ({r['context']})")
+        output_parts.append("")
+
+    return "\n".join(output_parts), [], {}, 0.08, revelations
 
 
 # ---------------------------------------------------------------------------

tests/test_generator.py

@@ -158,7 +158,7 @@ class TestParameterTesting:
         from server.tools_engine import execute_tool
         scenario = get_scenario("easy")
         hosts = list(scenario["hosts"].keys())
-        output, _, _, _ = execute_tool(
+        output, _, _, _, _ = execute_tool(
             "web_crawl", {"host": hosts[0]},
             scenario, hosts, {}, set()
         )
@@ -170,7 +170,7 @@ class TestParameterTesting:
         from server.tools_engine import execute_tool
         scenario = get_scenario("easy")
         hosts = list(scenario["hosts"].keys())
-        output, _, _, reward = execute_tool(
+        output, _, _, reward, _ = execute_tool(
             "test_injection", {"host": hosts[0], "endpoint": "/api/login"},
             scenario, hosts, {}, set()
         )
@@ -182,7 +182,7 @@ class TestParameterTesting:
         from server.tools_engine import execute_tool
         scenario = get_scenario("easy")
         hosts = list(scenario["hosts"].keys())
-        output, _, _, reward = execute_tool(
+        output, _, _, reward, _ = execute_tool(
             "test_injection",
             {"host": hosts[0], "endpoint": "/api/login", "parameter": "username"},
             scenario, hosts, {}, set()