Exploit Agent — Runner (tools/exploit_agent/runner/)

Round loop, outcome pipeline, advisory hooks, and context assembly for the single-target loop. Canonical implementation: tools/exploit_agent/runner/_impl.py (~2.6K lines); runner/loop.py is a plain from ._impl import ... re-export, and runner/context.py / runner/phase.py are re-export shims over the package-level context.py / phase_tracker.py. (Implementation note: the task brief named runner/tool_catalog.py, but the file lives at tools/exploit_agent/tool_catalog.py — there is no runner/tool_catalog.py.)

Entry point run_exploit_agent (_impl.py:391); companionship doc loop.md covers init sequence, phase tracking, and budgets in the same file.

Architecture

run_exploit_agent (_impl.py:391)
 ├─ init: target lock → attacker OS → context profile → stores/mutator →
 │        planner/session → memory → briefings → system prompt → seed messages
 ├─ per round: refresh advisories → budgets → compaction → model call →
 │             validate → dispatch tools → outcome pipeline → hooks → phase/summary
 └─ return final_result dict (target_ip, total_actions, audit_path, workspace,
    records, messages, outcome_summary, plan, session_state, ...)

async def run_exploit_agent(
    *,
    client: object,
    model: str,
    session: object,
    exploit_tools: list[dict[str, object]],
    policy: ExploitPolicy,
    target_ip: str,
    target_cve: str = "",
    target_os: str | None = None,
    known_cves: list[str] | None = None,
    service_context: str = "",
    reports_dir: Path = Path("reports"),
    experience_store: object | None = None,
    semantic_memory: object | None = None,
    scope_gate: "ScopeGate | None" = None,
    skill_registry: SkillRegistry | None = None,
    skills_cfg: dict[str, object] | None = None,
    skill_embedder: object | None = None,
    config: dict[str, object] | None = None,
    heartbeat: object = None,
    original_target: str | None = None,
    resolved_ip: str | None = None,
    event_sink: object = None,
    checkpoint_hook: CheckpointHook | None = None,
    clock: Callable[[], float] | None = None,
    sleep: Any | None = None,
) -> dict[str, object]:

Round loop lifecycle (_impl.py:1084)

for _round in range(max_rounds) (max_rounds = policy.settings.effective_max_rounds):

1. Heartbeat update + round header (ui.round_header, skipped round 0).
2. Refresh advisory messages: _refresh_attack_memory_message, _refresh_reasoning_advisory_message (single in-flight each).
3. Budget gates: time (_budget_exceeded) and policy.can_proceed — either streams a final summary via _stream_model and breaks.
4. Compaction: _should_compact_context → _build_compacted_messages + _checkpoint_messages (long-session resume).
5. response = await _call_model_with_retry(client, model, messages, _round_tools(), ...) where _round_tools() narrows schemas to the current phase (_impl.py:1065).
6. tool_calls, invalid = _filter_and_validate_tool_calls(raw, all_tools); malformed calls get a RECOVERABLE_ERROR tool message (+ replan prompt); non-recoverable ones count toward the terminal-constraint limit.
7. No-tool turn: phase-minima enforcement (can_terminate, finish-early push-back), goal-complete break on verified compromise/cred-dump, no_path checkpoint hook, else break.
8. Per tool call: availability check → research-assistant branch → repeated policy.approve_action → optional pre-destructive snapshot → heartbeat + tool_start emit → session.call_tool (one auto-retry via _attempt_retry_correction) → parse content blocks → blocked/success handling → outcome pipeline → policy.record + session_mgr.record_action → tool_result emit (secret-masked) → hooks → banner parsing/phase/plan.

# Example: minimal standalone invocation (all args keyword-only)
result = await run_exploit_agent(
    client=model_client,
    model="glm",
    session=mcp_session,
    exploit_tools=tool_schemas,
    policy=policy,
    target_ip="10.0.0.50",
)
assert result["total_actions"] >= 0

Outcome pipeline: normalize → classify → judge

One ActionResult per tool result, threaded through every downstream consumer (loop, swarm, report, eval). Authority flows strictly downward: the tracker reads the normalized verdict, never the judge alone.

In-loop wiring (_impl.py:1844-2057):

_action_result = _normalize_result(tool_name=name, result_text=result_text, mcp_result=result)
exit_code = _action_result.exit_code  # None preserved; never fabricated as 0
success = (_action_result.operational_status == _OpStatus.COMPLETED
           and not _action_result.is_error
           and exit_code in (0, None)
           and _action_result.exploit_outcome != _ExploitOutcome.FAILURE)
if _action_result.is_compromise:
    outcome_tracker.record_compromise(shell_type=..., privilege_level=...)
elif _action_result.is_cred_dump:
    outcome_tracker.record_cred_dump()
_flowa = await _judge_flow_a(..., action_result=_action_result)

_ToolOutcomeTracker (tool_calls.py:95): threshold=3 consecutive blocked/unavailable → terminal_constraint_reached; record_compromise / record_cred_dump / record_partial monotonic counters; record_exploit_success / record_exploit_failure drive the peer-consult trigger; should_escalate / should_pivot signals for the orchestrator. Only _EXPLOIT_VALIDATION_TOOLS (outcome_truth.py:67) can yield a compromise — recon/install/workspace output classifies as none.

Tool catalog (tool_catalog.py)

_round_tools() intersects the full schema list with the current phase family (_impl.py:1065-1070), always keeping _UNIVERSAL (shell, workspace, skills, advisory consults, state/capability tools) and hiding _HIDDEN (nested control-plane + package/clone noise). Unknown phase or empty selection falls back to the full list.

def select_tools_for_phase(
    all_tools: list[dict[str, Any]],
    phase: str,
    *,
    available_mcp_names: set[str] | None = None,
) -> list[dict[str, Any]]: ...

def validate_tool_call(
    name: str,
    args: dict[str, Any],
    all_tools: list[dict[str, Any]],
) -> str | None: ...  # None = valid; else short error fed back to the model

Reflection / peer-consult / reselect hooks

All three are advisory-only user-role injections — they never touch permission, scope, workspace, or audit.

Checkpoints (CheckpointContext/CheckpointOutcome/CheckpointHook, _impl.py:188-232): access fires right after the authoritative classifier confirms compromise/cred-dump (decision deferred until after audit + bookkeeping); no_path fires at the safe natural-termination boundary (minima met, no foothold). Both guarded against decision loops via _last_access_action / _last_no_path_action. finish/cancel outcomes set the break path; cancel sets final_result["cancelled_by_operator"].

Context assembly (context.py)

Config keys

Related documentation

• Exploit Agent overview
• Exploit Agent policy
• Exploit Agent context
• Exploit Agent loop
• Exploit Agent prompts
• Swarm overview
• Kernel overview
• Run service

Source map

• tools/exploit_agent/runner/_impl.py
• tools/exploit_agent/runner/loop.py
• tools/exploit_agent/runner/context.py
• tools/exploit_agent/runner/phase.py
• tools/exploit_agent/runner/__init__.py
• tools/exploit_agent/tool_catalog.py
• tools/exploit_agent/context.py
• tools/exploit_agent/outcome_truth.py
• tools/exploit_agent/outcome_adapter.py
• tools/exploit_agent/outcome_classify.py
• tools/exploit_agent/reflection.py
• tools/exploit_agent/skills.py
• tools/exploit_agent/phase_tracker.py
• tools/exploit_agent/tool_calls.py