Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/AdithyaaSivamal/Agentic-AFL/llms.txt

Use this file to discover all available pages before exploring further.

CampaignRunner is the top-level orchestration class that manages a complete fuzzing campaign. It replaces manual AFL++ process management and AgentLoop setup with a single async interface. Internally it handles AFL++ process lifecycle, AgentLoop orchestration, custom mutator deployment, edge-coverage monitoring, and JSON result serialization — all behind one run() call.

​
Import

from pathlib import Path
from agentic_afl.campaign import CampaignRunner

​
Constructor

harness, seed_dir, and duration are positional parameters. Everything after duration — stall_minutes, accept_marker, custom_mutator, log_dir, target_name, and on_update — is keyword-only (enforced by the * separator in __init__). Always pass them by name.
​
harness
Path
required
Path to the AFL++-instrumented harness binary. The binary must already be compiled with AFL++ instrumentation (e.g., afl-clang-fast). The path is resolved to an absolute path at construction time.
​
seed_dir
Path
required
Directory containing initial seed corpus files. All files in this directory are copied into the AFL++ input directory before the campaign starts.
​
duration
int
default:"3600"
Campaign duration in seconds. The monitoring loop runs until this many seconds have elapsed or request_shutdown() is called. Defaults to one hour.
​
stall_minutes
int
default:"5"
Minutes of edge plateau before the AgentLoop’s stall detector triggers. Maps directly to StallDetector._min_stall_time_seconds. Lower values make the agent more aggressive; higher values reduce false-positive stall detection on slow targets.
​
accept_marker
str
default:"\"ACCEPT\""
Stdout/stderr substring that indicates a full bypass has been achieved. Every ~5 seconds the runner replays recent AFL++ queue entries against the harness binary. If this marker appears in the output, bypass_detected is set and (if configured) the custom mutator is deployed.
​
custom_mutator
Path | None
default:"None"
Path to a Python AFL++ custom mutator module (.py file). When a bypass is detected, the current AFL++ process is terminated and restarted with AFL_PYTHON_MODULE pointed at this file. This implements the Level 3 post-solve mutation strategy.
​
log_dir
Path | None
default:"None"
Directory for JSON result files. When set, a timestamped <target_name>_<YYYYMMDD_HHMMSS>.json file is written at campaign end containing the full CampaignResult dict.
​
target_name
str | None
default:"None"
Human-readable campaign name used in log output and the result filename. Defaults to harness.stem (the harness binary filename without extension).
​
on_update
Callable[[CampaignSnapshot], None] | None
default:"None"
Callback fired approximately every 30 seconds with a CampaignSnapshot containing the current campaign state. Use this for dashboards, progress bars, or custom alerting. The callback is called synchronously inside the monitoring loop — keep it fast.

​
Methods

​
async run() -> CampaignResult

Execute the full campaign lifecycle. Starts AFL++, initializes and runs AgentLoop, polls edge coverage, detects bypass via the accept marker, deploys the custom mutator if configured, and collects results. Blocks until duration seconds have elapsed or request_shutdown() is called. The working directory .agentic_afl_workdir/ is created as a sibling of the harness binary and cleaned up before each run. Any stale afl-fuzz processes are killed with SIGKILL before the new campaign starts.

​
request_shutdown() -> None

Request graceful campaign shutdown. Sets an internal _shutdown flag that the monitoring loop checks on its next iteration (~5 second latency). Safe to call from a signal handler. After shutdown, run() completes its current monitoring iteration, collects final metrics, terminates AFL++ and the AgentLoop, then returns the CampaignResult.

​
CampaignSnapshot

The CampaignSnapshot dataclass is passed to the on_update callback every ~30 seconds. All fields are read-only snapshots; mutating them has no effect on the running campaign.
​
elapsed
float
Seconds elapsed since run() was called.
​
edges
int
Current AFL++ edges_found value read from fuzzer_stats.
​
baseline_edges
int
Edge count recorded within the first 10 seconds of the campaign, used as the baseline for edge_gain calculations.
​
execs
int
Total executions performed by AFL++ (execs_done from fuzzer_stats).
​
execs_per_sec
float
Execution throughput measured over the last 30-second window.
​
corpus_count
int
Number of entries in AFL++‘s queue (corpus_count from fuzzer_stats).
​
cycles_done
int
Number of complete AFL++ fuzzing cycles (cycles_done from fuzzer_stats).
​
pending_favs
int
Number of pending favored inputs (pending_favs from fuzzer_stats).
​
stalls_detected
int
Total stall events detected and queued by AgentLoop.
​
stalls_solved
int
Stall events that resulted in a successful SAT solve and payload injection.
​
payloads_injected
int
Total payloads written to the AFL++ sync directory.
​
llm_calls
int
Total LLM API calls made across all stall processing (each ReAct turn is one call).
​
react_turns
int
Total ReAct loop iterations consumed across all processed stalls.
​
bypass_detected
bool
Whether the accept marker has been observed in harness output.
​
bypass_time
float
Seconds elapsed when bypass was first detected. Zero if no bypass yet.
​
bypass_evidence
str
Filename of the AFL++ queue entry that triggered the bypass, or empty string.
​
mutator_deployed
bool
Whether the custom mutator has been deployed and AFL++ restarted with it.
​
mutator_name
str
Stem of the custom mutator module filename, or empty string if none configured.
​
edge_history
list[int]
List of edges_found samples recorded every 30 seconds. Useful for plotting coverage growth curves.

​
CampaignResult

CampaignResult is the return value of run(). It is a frozen dataclass that can be serialized to JSON via to_dict().
​
target_name
str
Campaign name (harness stem or the value passed to target_name).
​
baseline_edges
int
Edge count at campaign start (measured within the first 10 seconds).
​
final_edges
int
Edge count at campaign end.
​
edge_gain
int
Absolute edge increase: final_edges - baseline_edges.
​
edge_gain_pct
float
Percentage edge increase: (edge_gain / baseline_edges) * 100. Zero if baseline_edges is zero.
​
bypass_detected
bool
Whether a full bypass (accept marker in harness output) was achieved.
​
bypass_time_seconds
int | None
Integer seconds when bypass was first detected, or None if no bypass.
​
bypass_evidence
str | None
Filename of the queue entry that triggered bypass, or None.
​
payloads_injected
int
Total payloads injected into AFL++‘s sync directory by AgentLoop.
​
stalls_detected
int
Total stall events detected during the campaign.
​
llm_calls
int
Total LLM API calls made during the campaign.
​
react_turns
int
Total ReAct loop iterations consumed during the campaign.
​
elapsed_seconds
int
Actual wall-clock seconds the campaign ran (may be less than duration_seconds if shutdown was requested early).
​
duration_seconds
int
Configured campaign duration in seconds.
​
timeline
list[dict]
List of timeline snapshots recorded approximately every 60 seconds. Each dict contains time_seconds, edges, stalls_detected, and payloads_injected.
​
mutator_deployed
bool
Whether the custom mutator was deployed (AFL++ restarted with AFL_PYTHON_MODULE).

​
to_dict() -> dict

Returns the full CampaignResult as a JSON-serializable dictionary via dataclasses.asdict(). Suitable for writing directly to a file with json.dumps().

​
Example

import asyncio
import signal
from pathlib import Path
from agentic_afl.campaign import CampaignRunner

def on_update(snap):
    print(f"{snap.elapsed:.0f}s  edges={snap.edges}  injected={snap.payloads_injected}")

runner = CampaignRunner(
    harness=Path("./harness"),
    seed_dir=Path("./seeds"),
    duration=3600,
    stall_minutes=5,
    log_dir=Path("./results"),
    on_update=on_update,
)

# Graceful shutdown on Ctrl-C
signal.signal(signal.SIGINT, lambda s, f: runner.request_shutdown())

result = asyncio.run(runner.run())
print(f"Edge gain: +{result.edge_gain} ({result.edge_gain_pct:+.1f}%)")
print(f"Bypass: {result.bypass_detected}")

Build docs developers (and LLMs) love

Get started for freeTalk to us