feat(engine): autopilot ledger — capture findings + skip provably-wasteful agent calls (#57)

* docs(spec): engine-native cross-engagement memory (the ledger)

Design for auto-capturing cross-target hunt memory from the engine's
structured state.json at existing phase transitions, with recall that feeds
back into surface ranking (+ negative memory to prune dead classes).
Original design on our engine; replaces the dangling vendored memory commands.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

* docs(spec): scope ledger to Option A — autopilot token-saving only

Cut the interactive recall->ranking loop (marginal value, complexity in the
engine hot path). Scope memory to the autonomous hunt loop: capture confirmed
findings + negatives, then skip provably-wasteful agent calls (known-confirmed
carry-forward + dead-class) in token-saving modes only. Coverage-safe:
--paranoid ignores memory, skips are gated + logged, unseen classes never skipped.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

* docs(plan): autopilot ledger implementation plan (6 tasks, TDD)

Co-Authored-By: Claude Opus 4.8 <[email protected]>

* feat(engine): add autopilot ledger store + capture (record_finding/record_run)

* feat(engine): ledger skip_decision (known-confirmed/dead-class) + rollup + gc CLI

* feat(engine): capture confirmed findings + run rollup into the ledger

* feat(engine): gated ledger skip filter in hunt loop + --use-memory flag

* docs(commands): rewrite memory commands to the autopilot ledger

* docs(autopilot): drop second stale hunt-memory/audit.jsonl reference

* docs(surface): replace hunt-memory terminology with ledger

* docs(credits): ledger is original work, not vendored

* docs(credits): fix command counts — 15 shipped vs 12 vendored

* fix(engine): guard capture I/O + rotate under a stable lockfile (final review)

Finding 1: memory.record_finding/record_run calls in validate()/report()
now wrap in try/except so a ledger I/O error (unwritable dir, disk full)
logs and continues instead of aborting the run after state.json was saved.

Finding 2: _append() now acquires an exclusive flock on a stable
<path>.lock file BEFORE calling _rotate_if_needed, closing the
cross-process rename race where two engines could interleave rotation.

* docs(usage): document --use-memory autopilot flag (ledger shipped)

Mark the 'per-engagement memory' next-iteration as delivered; add the
--use-memory usage + the /autopilot mode mapping and coverage-safety note.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

---------

Co-authored-by: Sachin Sharma <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
This commit is contained in:
authored and GitHub committed 2026-08-09 01:09:16 +05:30
1 parent cad37df989
commit d5aa6d1a1f
11 files changed
+1359 -128

No files matched your search

+8 -1
View File
@@ -434,7 +434,14 @@ The validation engagement that produced this stack illustrated all three: the or
If you keep using this and want to extend it:
1. **Per-engagement memory** — extend `bb-methodology` to record patterns you've seen pay off across engagements. After 5+ engagements, your personal patterns will outperform the disclosed-report patterns.
1. **Per-engagement memory — ✅ now shipped (the autopilot ledger).** The engine auto-captures confirmed findings and dead-class negatives to `~/.claude/bughunter/memory/`, and the autonomous hunt loop can skip provably-wasteful agent calls — re-confirming a finding already confirmed on a prior run, or re-testing a vuln class that never pays off on a stack.
- **Off by default = full coverage.** Enable per run with `--use-memory`:
```bash
python3 engine/engine.py --scope <engagement>.json --hunt --use-memory
```
- `/autopilot` maps modes to it: `--paranoid` (default) / `--normal` → memory **off** (full coverage); `--quick` / `--yolo` → memory **on** (token-saving skips).
- A class never hunted on a stack is never skipped, and every skip is logged — coverage is never silently reduced. Manage the store with `/memory-gc`; review a target's history with `/pickup <host>`.
- Benefit scales with repeat targets / recurring stacks; expect little effect on a cold start (new or one-off stack).
2. **HackerOne MCP integration** — wire up the H1 MCP in shuvonsec's repo for live duplicate-search and program intel during reports.
3. **Specialized `hunt-*` skills** for high-payout niches you focus on (e.g., `hunt-fintech-graphql`, `hunt-healthcare-fhir`).
4. **A `program-rules-parser` skill** that takes program text and produces a structured `scope.md` automatically.
+13 -2
View File
@@ -63,7 +63,6 @@ Runs the full hunt cycle without stopping for approval at each step:
## Safety Guarantees
- **Every URL** is checked against the scope allowlist before any request
- **Every request** is logged to `hunt-memory/audit.jsonl`
- **Reports are NEVER auto-submitted** — always requires explicit approval
- **PUT/DELETE/PATCH** require human approval in --yolo mode (safe methods only)
- **Circuit breaker** stops hammering if 5 consecutive 403/429/timeout on same host
@@ -77,8 +76,20 @@ Runs the full hunt cycle without stopping for approval at each step:
| `--normal` | After validation batch | Systematic coverage |
| `--yolo` | After full surface exhausted | Familiar targets, experienced hunters |
## Ledger memory (token saving)
The autonomous loop can consult the **autopilot ledger** to skip provably-wasteful agent
calls — an item already **confirmed** on a prior run (carried forward, not re-tested) or a
vuln_class with strong **negative** history on this tech stack (`dead-class`).
- **`--paranoid` (default) / `--normal`:** memory is **off** — full coverage, no skips.
- **`--quick` / `--yolo`:** the command passes `--use-memory` to the engine — skips on.
- A vuln_class never hunted on this stack is **never** skipped, and every skip is logged
(`hunt: skipped via ledger (...)`). Coverage is never silently reduced.
- Capture is automatic; the ledger fills itself as you run. Manage size with `/memory-gc`.
## After Autopilot
- Run `/remember` to log successful patterns to hunt memory
- Run `/pickup target.com` next time to pick up where you left off
- Check `hunt-memory/audit.jsonl` for a full request log
- Check the ledger (`~/.claude/bughunter/memory/findings.jsonl`) or run `/memory-gc` for its size
+8 -32
View File
@@ -1,19 +1,11 @@
---
name: memory-gc
description: Inspect or rotate hunt-memory JSONL files (audit.jsonl, patterns.jsonl, journal.jsonl). Caps file size and keeps N rotated backups so memory does not grow unbounded.
description: Inspect or rotate the autopilot ledger JSONL files (findings.jsonl, negatives.jsonl). Caps file size and keeps N rotated backups so memory does not grow unbounded.
---
# /memory-gc
Garbage-collect the hunt-memory directory. Reports current sizes, rotates oversized files past a configurable cap, or purges old backups.
## Why This Exists
Append-only logs grow without bound. On active hunters:
- `audit.jsonl` can reach 100 MB+ in months (every outbound request)
- `patterns.jsonl` and `journal.jsonl` accumulate forever
This command surfaces that growth and gives you a one-shot fix.
Garbage-collect the autopilot ledger (`~/.claude/bughunter/memory/`). Reports current sizes, rotates oversized files past a configurable cap, or purges old backups.
## Usage
@@ -22,36 +14,20 @@ This command surfaces that growth and gives you a one-shot fix.
/memory-gc --rotate # rotate files above 10 MB (default cap)
/memory-gc --rotate --max-mb 5 # custom cap
/memory-gc --purge-backups # delete all .1/.2/.3 backups
/memory-gc --dir <path> # scan a non-default hunt-memory dir
/memory-gc --dir <path> # scan a non-default ledger dir
```
## What It Does
1. Walks the hunt-memory directory recursively.
2. Finds `audit.jsonl`, `patterns.jsonl`, and `journal.jsonl` files at any depth.
3. Prints a per-file table: live size, total (live + backups), backup count, status.
4. With `--rotate`: renames oversize files to `<file>.1`, shifting older backups up to `<file>.{keep}`. The oldest is dropped.
5. With `--purge-backups`: removes every `.1`/`.2`/`.3` backup, keeping only live files.
## Implementation
The agent shells out to:
The agent shells out to, from the repo root:
```bash
python -m tools.memory_gc [args]
python3 engine/memory.py --gc [--rotate] [--purge-backups] [--dir PATH] [--max-mb N]
```
from the repo root.
## Defaults
- **Rotation cap:** 10 MB per file
- **Backups kept:** 3 (so `<file>.1` newest → `<file>.3` oldest)
- **Scope:** `hunt-memory/` and any nested target dirs
- **Rotation cap:** 10 MB per file · **Backups kept:** 3 (`<file>.1` newest → `<file>.3` oldest)
- **Scope:** every `*.jsonl` under the ledger dir (`findings.jsonl`, `negatives.jsonl`)
Auto-rotation fires automatically in two places:
1. **On every write** — inside `AuditLog.log()` and `PatternDB.save()` when the next append would exceed the cap.
2. **On session end** — a `Stop` hook in `.claude/settings.json` runs `python3 -m tools.memory_gc --rotate` so long sessions that wrote a lot but never crossed the cap mid-session still get cleaned up.
So this slash command is mainly for ad-hoc reporting (`/memory-gc` with no args) and manual cleanup of accumulated backups (`/memory-gc --purge-backups`).
Rotation also fires automatically on append inside the ledger writer, so files stay bounded without any session-end hook.
+10 -33
View File
@@ -1,21 +1,19 @@
---
name: pickup
description: Pick up a previous hunt on a target — shows hunt history, untested endpoints, and memory-informed suggestions. Usage: /pickup target.com
description: Pick up a previous hunt on a target — shows hunt history and untested surface from the autopilot ledger. Usage: /pickup target.com
---
# /pickup
Pick up where you left off on a target.
> **Renamed from `/resume`** — `/resume` is a reserved Claude Code command. Use `/pickup` to continue a previous hunt.
> **Renamed from `/resume`** — `/resume` is a reserved Claude Code command.
## What This Does
1. Reads the target profile from `hunt-memory/targets/<target>.json`
2. Shows hunt history (sessions, findings, payouts)
3. Lists untested endpoints from last recon
4. Suggests techniques based on tech stack + pattern DB
5. Asks: continue hunting or re-run recon?
1. Reads the target rollup from `~/.claude/bughunter/memory/targets/<host>.json`.
2. Shows hunt history (sessions, last seen, tech stack).
3. Lists confirmed findings and the endpoints already tested.
## Usage
@@ -23,37 +21,16 @@ Pick up where you left off on a target.
/pickup target.com
```
## Output
## Implementation
```
PICKUP: target.com
═══════════════════════════════════════
The agent reads the rollup directly:
Hunt History:
Sessions: 3
Last hunt: 2026-03-24
Total time: 2h 00m
Findings: 1 confirmed (IDOR, $1500 paid)
Untested Surface:
3 endpoints from last recon:
1. /api/v2/users/{id}/export
2. /api/v2/users/{id}/share
3. /api/v2/users/{id}/history
Memory Suggestions:
Tech stack [Next.js, GraphQL, PostgreSQL] matches 2 targets
where you found auth bypass. Try introspection → mutation pattern.
Actions:
[r] Continue hunting untested endpoints
[n] Re-run recon first (surface may have changed)
[s] Show full hunt journal for this target
```bash
python3 -c "import sys; sys.path.insert(0,'engine'); import memory, json; print(json.dumps(memory.rollup('target.com'), indent=2))"
```
## If No Previous Hunt
```
No previous hunt data for target.com.
Run /recon target.com first, then /hunt target.com.
No ledger data for target.com. Run /recon then /autopilot target.com first.
```
+9 -50
View File
@@ -1,63 +1,22 @@
---
name: remember
description: Log current finding or successful pattern to hunt memory. Auto-fills from /validate output if available. Usage: /remember
description: Optional manual note on a target or the last confirmed finding. Capture is automatic during autopilot; this is for extra context. Usage: /remember
---
# /remember
Save a finding or successful pattern to persistent hunt memory.
## What This Does
1. Auto-populates fields from session context (target, endpoint, vuln_class, technique)
2. If `/validate` was run in this session, pre-fills from validation output
3. Prompts you to confirm or edit before saving
4. Writes to `journal.jsonl` (always) + `patterns.jsonl` (if confirmed + payout > 0)
5. Updates the target profile's `tested_endpoints` and `findings`
> **Capture is automatic.** During an engine/autopilot run, every confirmed finding is
> written to the ledger (`~/.claude/bughunter/memory/findings.jsonl`) with no action from you.
> `/remember` is only for adding an **optional manual note** — a technique detail or
> follow-up idea — that the automatic capture would not include.
## Usage
```
/remember # after finding something
/remember --from-validate # explicitly pull from last /validate
/remember # then describe the note; it is appended to the target's rollup
```
## Interactive Flow
## What It Is Not
```
REMEMBER — Log finding to hunt memory
Target: target.com (auto-detected)
Endpoint: /api/v2/users/{id}/orders (from session)
Vuln Class: idor (from session)
Technique: numeric_id_swap_with_put_method
Result: [confirmed / rejected / partial / informational]?
Severity: [critical / high / medium / low]?
Payout: $___?
Notes: ___?
Tags: [comma-separated]?
Save to hunt memory? [y/n]
```
## Minimum Required Fields
- target
- vuln_class
- endpoint
- result
## What Gets Written
| Field | journal.jsonl | patterns.jsonl | target profile |
|---|---|---|---|
| Finding details | Always | If confirmed + payout > 0 | findings[] updated |
| Tested endpoint | — | — | tested_endpoints[] updated |
| Tech stack | — | From target profile | — |
## Why This Matters
- Next time you hunt a target with similar tech stack, your successful patterns are suggested first
- `/pickup target.com` shows which endpoints you've tested and which remain
- Cross-target learning: patterns from target A inform hunting on target B
This is no longer the primary way findings enter memory — the engine captures those
deterministically. Use it sparingly for human context.
+2 -2
View File
@@ -1,6 +1,6 @@
---
name: surface
description: Show ranked attack surface for a target from its recon manifest + hunt memory. Deterministic backing is `cbh surface <target>` (reads recon/<target>/manifest.json); LLM layer adds hunt-memory signal. Usage: /surface target.com
description: Show ranked attack surface for a target from its recon manifest + hunt memory. Deterministic backing is `cbh surface <target>` (reads recon/<target>/manifest.json); LLM layer adds ledger signal. Usage: /surface target.com
---
# /surface
@@ -12,7 +12,7 @@ View the prioritized attack surface for a target.
1. Reads the recon manifest at `recon/<target>/manifest.json` (produced by `cbh recon`)
— the deterministic ranking comes from `cbh surface <target>`
2. Reads hunt memory for patterns and previously tested endpoints
3. Layers `engine/skill_map.py` (surface → bug-class → skill) + hunt-memory signal on
3. Layers `engine/skill_map.py` (surface → bug-class → skill) + ledger signal on
top of the manifest's `ranked_surface` (P1/P2/Kill)
4. Outputs P1 (start here), P2 (after P1), and Kill List (skip)
+5 -4
View File
@@ -13,7 +13,7 @@ This is a **bundle**: skills and commands are vendored directly into `skills/` a
| **Original / personally-curated skills** | 54 | This repo |
| Community-contributed skills (v3) | 20 | community PRs (e.g. #7) |
| Vendored foundation skills | 8 | shuvonsec/claude-bug-bounty (MIT) |
| Vendored slash commands | 15 | shuvonsec/claude-bug-bounty (MIT) |
| Vendored slash commands | 12 | shuvonsec/claude-bug-bounty (MIT) |
| **Total** | 82 skills + 15 commands | |
---
@@ -80,6 +80,7 @@ Built from authorized red-team engagements (enterprise targets including on-prem
- **`hunt <target>` command** — Engagement-folder scaffolding: creates `~/Targets/<name>/` with `CLAUDE.md`, `scope.md`, `findings/`, `evidence/`, `submissions.txt`, `notes.md`, and a sensible `.gitignore` for engagement artifacts. Ships as `scripts/hunt.sh` (bash) and `scripts/hunt.ps1` (PowerShell).
- **Bundle packaging** — Single-step installer that copies all 82 skills, 15 commands, and the hunt scaffold into `~/.claude/`: `scripts/install.sh` (macOS/Linux) and `scripts/install.ps1` (Windows/PowerShell).
- **Autopilot ledger (`engine/memory.py`) + the `/remember`, `/memory-gc`, `/pickup` commands** — Original design and implementation (cross-engagement capture + skip-decision for the engine hunt loop). Not derived from any external memory implementation.
- **`assets/banner-v2.svg`** — Hand-coded SVG banner.
- **Documentation** — `README.md`, `INSTALL.md`, `USAGE.md`, `CONTRIBUTING.md`, `docs/architecture.md`, this credits file.
@@ -87,7 +88,7 @@ Built from authorized red-team engagements (enterprise targets including on-prem
## Vendored foundation (from shuvonsec/claude-bug-bounty)
These 8 skills + 15 slash commands form the methodology backbone of the bundle. Vendored as-is (MIT-licensed) so the entire stack installs in one step.
These 8 skills + 12 slash commands form the methodology backbone of the bundle. Vendored as-is (MIT-licensed) so the entire stack installs in one step.
### Skills (8)
@@ -102,9 +103,9 @@ These 8 skills + 15 slash commands form the methodology backbone of the bundle.
| `web3-audit` | 10 DeFi bug classes, Foundry PoC template |
| `meme-coin-audit` | Token rug-pull detection |
### Slash commands (15)
### Slash commands (12)
`/hunt` `/recon` `/scope` `/triage` `/validate` `/report` `/autopilot` `/chain` `/intel` `/pickup` `/surface` `/remember` `/memory-gc` `/token-scan` `/web3-audit`
`/hunt` `/recon` `/scope` `/triage` `/validate` `/report` `/autopilot` `/chain` `/intel` `/surface` `/token-scan` `/web3-audit`
**Repo**: https://github.com/shuvonsec/claude-bug-bounty
**License**: MIT (verify in upstream repo)
@@ -0,0 +1,770 @@
# Autopilot Ledger Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Give the engine's autonomous hunt loop a cross-engagement memory that captures confirmed findings + dead classes, then skips provably-wasteful agent calls in token-saving modes.
**Architecture:** One new stdlib-only module `engine/memory.py` (capture + skip-decision + gc), import-isolated from siblings. `engine.py` calls capture at its existing `confirm()`/`report` transitions and consults `skip_decision` in the hunt loop only when `--use-memory` is set (default off = full coverage). `engine/state.py` unchanged.
**Tech Stack:** Python 3 stdlib only (`fcntl`, `json`, `os`, `urllib.parse`, `datetime`). No new deps. House test convention: in-file `_selftest()` under `__main__`, run via `python engine/<file>.py` (matches `state.py`/`recon.py`).
## Global Constraints
- **stdlib only** — no third-party imports in `engine/memory.py`.
- **Import isolation** — `engine/memory.py` imports nothing from sibling engine modules; `engine.py` passes state data in.
- **Storage root** — `~/.claude/bughunter/memory/`, overridable via `BUGHUNTER_MEMORY_DIR`.
- **Coverage safety** — skips are OFF by default (`--use-memory` opt-in); a class never hunted is never skipped; every skip is logged.
- **Filenames** — `findings.jsonl`, `negatives.jsonl`, `targets/<host>.json`. Dedup key for findings: `(host, url, param, vuln_class)`.
- **Constants (tuning knobs, top of module)** — `SCHEMA_VERSION=1`, `MAX_BYTES=10*1024*1024`, `KEEP=3`, `DEAD_CLASS_THRESHOLD=5`.
- **House test style** — no pytest fixtures; assert-based `_selftest()` run as a script.
- **Invocation** — script style `python3 engine/memory.py`, never `python -m` (repo has no `__init__.py` in `engine/`).
---
### Task 1: `engine/memory.py` — store, rotation, capture
**Files:**
- Create: `engine/memory.py`
**Interfaces:**
- Produces: `record_finding(engagement:str, host:str, tech_stack:list[str], finding:dict, verdict:dict) -> bool` (True if newly saved, False if dup/invalid); `record_run(engagement:str, tech_stack:list[str], tested:list[str], confirmed:list[dict]) -> None`; helpers `_host(url)->str`, `_stack_sig(list)->str`, `_root()->str`, `_append`, `_read`, `_rotate_if_needed`.
- [ ] **Step 1: Write the module with capture + a failing self-test**
Create `engine/memory.py`:
```python
#!/usr/bin/env python3
"""
memory.py — autopilot ledger: cross-engagement capture + skip decisions.
Lets the autonomous hunt loop skip provably-wasteful agent calls: re-confirming a
known finding, or re-testing a vuln_class that never pays off on this tech stack.
Capture is harvested from the engine's own state; skip is gated (off unless the
engine passes use_memory, i.e. token-saving modes only).
Storage (~/.claude/bughunter/memory/, override BUGHUNTER_MEMORY_DIR):
findings.jsonl one row per confirmed finding (cross-target)
negatives.jsonl (host, stack_sig, vuln_class) hunted -> not confirmed
targets/<host>.json per-target rollup for /pickup
stdlib-only; imports nothing from sibling engine modules.
"""
import fcntl
import json
import os
import sys
from datetime import datetime, timezone
from urllib.parse import urlparse
SCHEMA_VERSION = 1
MAX_BYTES = 10 * 1024 * 1024 # rotate a JSONL past 10 MB
KEEP = 3 # backups kept: .1 (newest) .. .3
DEAD_CLASS_THRESHOLD = 5 # negatives before a (stack,class) is "dead"
def _root():
return os.path.expanduser(os.environ.get("BUGHUNTER_MEMORY_DIR", "~/.claude/bughunter/memory"))
def _paths():
root = _root()
os.makedirs(os.path.join(root, "targets"), exist_ok=True)
return root
def _now():
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _host(url):
u = url or ""
if "://" not in u:
u = "//" + u
return (urlparse(u).hostname or "").lower().rstrip(".")
def _stack_sig(tech_stack):
return "|".join(sorted({t.lower() for t in (tech_stack or []) if t}))
def _rotate_if_needed(path):
try:
if os.path.getsize(path) < MAX_BYTES:
return
except FileNotFoundError:
return
oldest = f"{path}.{KEEP}"
if os.path.exists(oldest):
os.remove(oldest)
for i in range(KEEP - 1, 0, -1):
s, d = f"{path}.{i}", f"{path}.{i + 1}"
if os.path.exists(s):
os.replace(s, d)
os.replace(path, f"{path}.1")
def _append(path, obj):
line = json.dumps(obj, separators=(",", ":")) + "\n"
_rotate_if_needed(path)
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
try:
fcntl.flock(fd, fcntl.LOCK_EX)
os.write(fd, line.encode("utf-8"))
finally:
fcntl.flock(fd, fcntl.LOCK_UN)
os.close(fd)
def _read(path):
out = []
if not os.path.exists(path):
return out
with open(path, encoding="utf-8") as f:
for ln in f:
ln = ln.strip()
if not ln:
continue
try:
out.append(json.loads(ln))
except json.JSONDecodeError:
continue
return out
def _valid_finding(e):
if not (e.get("host") and e.get("url") and e.get("vuln_class")):
return False
try:
datetime.fromisoformat(e["ts"].replace("Z", "+00:00"))
except Exception:
return False
return isinstance(e.get("tech_stack", []), list)
def record_finding(engagement, host, tech_stack, finding, verdict):
"""Capture one confirmed finding. Dedup on (host,url,param,vuln_class)."""
root = _paths()
fp = os.path.join(root, "findings.jsonl")
entry = {
"schema_version": SCHEMA_VERSION, "ts": _now(), "engagement": engagement,
"host": host or _host(finding.get("url", "")), "url": finding.get("url", ""),
"param": finding.get("param", ""), "vuln_class": finding.get("vuln_class", ""),
"technique": finding.get("technique", ""),
"severity": (verdict or {}).get("severity") or finding.get("severity", ""),
"verdict_reason": (verdict or {}).get("reason", ""),
"tech_stack": list(tech_stack or []),
"evidence_ref": finding.get("evidence_ref", ""),
}
if not _valid_finding(entry):
print(f"memory: skipping invalid finding {entry.get('url')!r}", file=sys.stderr)
return False
key = (entry["host"], entry["url"], entry["param"], entry["vuln_class"])
for e in _read(fp):
if (e.get("host"), e.get("url"), e.get("param"), e.get("vuln_class")) == key:
return False
_append(fp, entry)
return True
def _parse_key(k):
"""state 'url|param|class' -> (host, vuln_class)."""
url, _, rest = k.partition("|")
_param, _, cls = rest.partition("|")
return _host(url), cls
def _write_rollup(host, tech_stack, tested, confirmed):
root = _paths()
p = os.path.join(root, "targets", f"{host}.json")
prev = {}
if os.path.exists(p):
try:
prev = json.load(open(p))
except Exception:
prev = {}
hc_conf = [f for f in confirmed if _host(f.get("url", "")) == host]
doc = {
"schema_version": SCHEMA_VERSION, "host": host,
"first_seen": prev.get("first_seen", _now()), "last_seen": _now(),
"sessions": prev.get("sessions", 0) + 1, "tech_stack": list(tech_stack or []),
"tested": [k for k in tested if _parse_key(k)[0] == host],
"confirmed": [{"url": f.get("url"), "param": f.get("param", ""),
"vuln_class": f.get("vuln_class"), "severity": f.get("severity", "")}
for f in hc_conf],
}
tmp = p + ".tmp"
json.dump(doc, open(tmp, "w"), indent=2)
os.replace(tmp, p)
def record_run(engagement, tech_stack, tested, confirmed):
"""End-of-run capture: negatives for tested-but-unconfirmed classes + per-host rollup."""
root = _paths()
sig = _stack_sig(tech_stack)
tested_hc = {_parse_key(k) for k in tested}
confirmed_hc = {(_host(f.get("url", "")), f.get("vuln_class", "")) for f in confirmed}
neg = os.path.join(root, "negatives.jsonl")
for host, cls in tested_hc:
if cls and (host, cls) not in confirmed_hc:
_append(neg, {"schema_version": SCHEMA_VERSION, "ts": _now(),
"host": host, "stack_sig": sig, "vuln_class": cls})
for host in {h for h, _ in tested_hc} | {h for h, _ in confirmed_hc}:
if host:
_write_rollup(host, tech_stack, tested, confirmed)
def _selftest():
import tempfile, shutil
d = tempfile.mkdtemp()
os.environ["BUGHUNTER_MEMORY_DIR"] = d
try:
f = {"url": "http://a.com/x", "param": "id", "vuln_class": "idor", "severity": "high"}
assert record_finding("e1", "a.com", ["nextjs"], f, {"reason": "pii", "severity": "high"}) is True
assert record_finding("e1", "a.com", ["nextjs"], f, {"reason": "pii"}) is False # dedup
record_run("e1", ["nextjs"], ["http://a.com/y|q|xss", "http://a.com/x|id|idor"],
[{"url": "http://a.com/x", "vuln_class": "idor"}])
negs = _read(os.path.join(d, "negatives.jsonl"))
assert any(n["vuln_class"] == "xss" for n in negs) # xss tested, not confirmed -> negative
assert not any(n["vuln_class"] == "idor" for n in negs) # idor confirmed -> no negative
assert json.load(open(os.path.join(d, "targets", "a.com.json")))["sessions"] == 1
print("memory.py self-test: PASS")
finally:
shutil.rmtree(d)
os.environ.pop("BUGHUNTER_MEMORY_DIR", None)
if __name__ == "__main__":
_selftest()
```
- [ ] **Step 2: Run the self-test, expect PASS**
Run: `python3 engine/memory.py`
Expected: `memory.py self-test: PASS`. (If it errors, the module is broken — fix before committing.)
- [ ] **Step 3: Commit**
```bash
git add engine/memory.py
git commit -m "feat(engine): add autopilot ledger store + capture (record_finding/record_run)"
```
---
### Task 2: `skip_decision`, `rollup`, and the `--gc` CLI
**Files:**
- Modify: `engine/memory.py`
**Interfaces:**
- Consumes: everything from Task 1.
- Produces: `skip_decision(host:str, tech_stack:list[str], item:dict) -> {"skip":bool,"reason":str,"carry":dict|None}`; `rollup(host:str) -> dict|None`; `gc(action:str, max_mb:int, keep:int, root:str|None) -> list[dict]`; a `__main__` that runs `--gc` when args are present, else the self-test.
- [ ] **Step 1: Add skip_decision + rollup + gc + extend the self-test**
Insert these functions **before** `_selftest()` in `engine/memory.py`:
```python
def skip_decision(host, tech_stack, item):
"""Should the hunt loop skip this worklist item? Off-path returns skip=False."""
root = _paths()
url, param, cls = item.get("url", ""), item.get("param", ""), item.get("vuln_class", "")
# 1) known-confirmed -> carry the finding forward, no agent
for e in _read(os.path.join(root, "findings.jsonl")):
if (e.get("host"), e.get("url"), e.get("param"), e.get("vuln_class")) == (host, url, param, cls):
carry = {"url": url, "param": param, "vuln_class": cls,
"severity": e.get("severity", ""), "evidence": e.get("verdict_reason", ""),
"verdict_reason": e.get("verdict_reason", ""),
"verdict": {"real": True, "severity": e.get("severity", ""),
"reason": e.get("verdict_reason", ""), "from_memory": True}}
return {"skip": True, "reason": "known-confirmed", "carry": carry}
# 2) dead-class: >= threshold negatives for (stack_sig, cls) AND zero confirmations for it
sig = _stack_sig(tech_stack)
negs = sum(1 for e in _read(os.path.join(root, "negatives.jsonl"))
if e.get("stack_sig") == sig and e.get("vuln_class") == cls)
if negs >= DEAD_CLASS_THRESHOLD:
confirmed_here = any(_stack_sig(e.get("tech_stack", [])) == sig and e.get("vuln_class") == cls
for e in _read(os.path.join(root, "findings.jsonl")))
if not confirmed_here:
return {"skip": True, "reason": f"dead-class ({negs} negatives, 0 confirmed)", "carry": None}
return {"skip": False, "reason": "", "carry": None}
def rollup(host):
p = os.path.join(_root(), "targets", f"{host}.json")
return json.load(open(p)) if os.path.exists(p) else None
def gc(action="report", max_mb=10, keep=KEEP, root=None):
"""report | rotate | purge-backups over every *.jsonl in the ledger dir."""
base = root or _root()
rows = []
for dirpath, _dirs, names in os.walk(base):
for n in sorted(names):
if not n.endswith(".jsonl"):
continue
live = os.path.join(dirpath, n)
backups = [f"{live}.{i}" for i in range(1, keep + 1) if os.path.exists(f"{live}.{i}")]
live_sz = os.path.getsize(live) if os.path.exists(live) else 0
total = live_sz + sum(os.path.getsize(b) for b in backups)
if action == "rotate" and live_sz >= max_mb * 1024 * 1024:
_rotate_if_needed(live)
if action == "purge-backups":
for b in backups:
os.remove(b)
rows.append({"file": live, "live_bytes": live_sz, "total_bytes": total,
"backups": len(backups)})
return rows
```
Then **replace** the `__main__` block at the bottom with:
```python
if __name__ == "__main__":
import argparse
ap = argparse.ArgumentParser(description="autopilot ledger")
ap.add_argument("--gc", action="store_true", help="run garbage-collection instead of the self-test")
ap.add_argument("--rotate", action="store_true", help="with --gc: rotate files over the cap")
ap.add_argument("--purge-backups", action="store_true", help="with --gc: delete all .1/.2/.3 backups")
ap.add_argument("--dir", default=None, help="ledger dir (default ~/.claude/bughunter/memory)")
ap.add_argument("--max-mb", type=int, default=10)
a = ap.parse_args()
if a.gc:
act = "purge-backups" if a.purge_backups else ("rotate" if a.rotate else "report")
for r in gc(act, a.max_mb, KEEP, a.dir):
print(f" {r['file']}: live={r['live_bytes']}B total={r['total_bytes']}B backups={r['backups']}")
else:
_selftest()
```
And **extend** `_selftest()` — add before its final `print(...)`:
```python
# skip_decision: known-confirmed carries the finding
d = skip_decision("a.com", ["nextjs"], {"url": "http://a.com/x", "param": "id", "vuln_class": "idor"})
assert d["skip"] and d["reason"] == "known-confirmed" and d["carry"]["verdict"]["real"] is True
# unseen class is never skipped
assert skip_decision("a.com", ["nextjs"], {"url": "http://a.com/x", "param": "id", "vuln_class": "rce"})["skip"] is False
# dead-class only after >= threshold negatives with zero confirmations
for _ in range(DEAD_CLASS_THRESHOLD):
record_run("e", ["django"], ["http://b.com/z|q|csrf"], [])
assert skip_decision("b.com", ["django"], {"url": "http://b.com/z", "param": "q", "vuln_class": "csrf"})["skip"] is True
# a confirmation on that (stack,class) revives it (not dead)
record_finding("e", "b.com", ["django"], {"url": "http://b.com/z", "param": "q", "vuln_class": "csrf", "severity": "low"}, {"reason": "x"})
assert skip_decision("b.com", ["django"], {"url": "http://b.com/z", "param": "q", "vuln_class": "csrf"})["reason"] == "known-confirmed"
```
- [ ] **Step 2: Run the self-test, expect PASS**
Run: `python3 engine/memory.py`
Expected: `memory.py self-test: PASS`
- [ ] **Step 3: Smoke the gc CLI**
Run: `python3 engine/memory.py --gc`
Expected: prints a per-file line for any `*.jsonl` under the ledger dir (or nothing if empty). Exit 0.
- [ ] **Step 4: Commit**
```bash
git add engine/memory.py
git commit -m "feat(engine): ledger skip_decision (known-confirmed/dead-class) + rollup + gc CLI"
```
---
### Task 3: Wire capture into `engine.py`
**Files:**
- Modify: `engine/engine.py` (import; `_tech()` helper; capture after `confirm()`; capture in `report()`)
**Interfaces:**
- Consumes: `memory.record_finding`, `memory.record_run`, `memory._host` from Tasks 1-2.
- Produces: `Engine._tech() -> list[str]` (sorted tech union from state targets).
- [ ] **Step 1: Import memory + add the tech-stack helper**
In `engine/engine.py`, after line 27 (`import agent as A # noqa: E402`), add:
```python
import memory # noqa: E402 (autopilot ledger; import-isolated)
```
Add this method to the `Engine` class (place it just above `def hunt(self):`):
```python
def _tech(self):
"""Sorted union of tech fingerprints across recon targets (the stack signature source)."""
return sorted({t for tg in self.eng.state.get("targets", []) for t in tg.get("tech", [])})
```
- [ ] **Step 2: Capture each confirmed finding (validate phase)**
In `validate()`, the confirm call is:
```python
self.eng.confirm(c, v)
```
Immediately **after** that line, add:
```python
memory.record_finding(self.scope.name, memory._host(c["url"]), self._tech(), c, v)
```
- [ ] **Step 3: Capture the run at report time**
In `report()`, after the line:
```python
self.eng.log(f"report: wrote {len(c)} confirmed finding(s) -> {path}")
```
and before `return path`, add:
```python
memory.record_run(self.scope.name, self._tech(), self.eng.state["tested"], self.eng.state["confirmed"])
```
- [ ] **Step 4: Regression + capture check via the mock engine**
Run:
```bash
python3 engine/state.py
BUGHUNTER_MEMORY_DIR=/tmp/led1 python3 engine/engine.py --scope engine/engagement.example.json --base /tmp/eng1 --mock --hunt
```
Expected: `state.py self-test: PASS`; the mock run finishes; `/tmp/led1/findings.jsonl` exists with ≥1 row and `/tmp/led1/targets/` has a `<host>.json`.
Verify: `cat /tmp/led1/findings.jsonl` shows a confirmed finding; `ls /tmp/led1/targets`.
- [ ] **Step 5: Commit**
```bash
git add engine/engine.py
git commit -m "feat(engine): capture confirmed findings + run rollup into the ledger"
```
---
### Task 4: Skip filter in the hunt loop + `--use-memory` flag
**Files:**
- Modify: `engine/engine.py` (`Engine.__init__` signature; hunt loop; argparse; `Engine(...)` construction)
**Interfaces:**
- Consumes: `memory.skip_decision`, `Engine._tech` from Task 3.
- Produces: `Engine(..., use_memory=False)` constructor param; `--use-memory` CLI flag.
- [ ] **Step 1: Add `use_memory` to the constructor**
Change the `Engine.__init__` signature from:
```python
def __init__(self, scope_path, base, model, max_hunts, max_turns, timeout, mock=False,
allow_intrusive=False, parallel=3, expand=False):
```
to (add `use_memory=False` at the end):
```python
def __init__(self, scope_path, base, model, max_hunts, max_turns, timeout, mock=False,
allow_intrusive=False, parallel=3, expand=False, use_memory=False):
```
And after the line `self.parallel = max(1, parallel)` add:
```python
self.use_memory = use_memory
```
- [ ] **Step 2: Insert the gated skip filter in the hunt worklist loop**
The current loop in `hunt()` is:
```python
for it in wl:
if not self.scope.in_scope_host(it["url"]): # belt-and-suspenders scope gate
self.eng.log(f"hunt: REFUSING out-of-scope {it['url']} ({self.scope.reject_reason(it['url'])})")
self.eng.mark_tested(it)
else:
todo.append(it)
```
Replace it with (adds one `elif` branch; scope gate unchanged, default path unchanged):
```python
tech = self._tech()
for it in wl:
if not self.scope.in_scope_host(it["url"]): # belt-and-suspenders scope gate
self.eng.log(f"hunt: REFUSING out-of-scope {it['url']} ({self.scope.reject_reason(it['url'])})")
self.eng.mark_tested(it)
continue
if self.use_memory:
d = memory.skip_decision(memory._host(it["url"]), tech, it)
if d["skip"]:
self.eng.log(f"hunt: skipped via ledger ({d['reason']}): "
f"{it.get('vuln_class')}@{it['url']}")
if d["carry"]:
self.eng.confirm(d["carry"], d["carry"]["verdict"])
self.eng.mark_tested(it)
continue
todo.append(it)
```
- [ ] **Step 3: Add the `--use-memory` CLI flag**
In `main()`, after the `--expand` argument block, add:
```python
ap.add_argument("--use-memory", action="store_true",
help="OPT-IN: skip provably-wasteful agent calls (known-confirmed / dead-class) "
"using the autopilot ledger. Default OFF = full coverage.")
```
- [ ] **Step 4: Pass it into the Engine constructor**
Change:
```python
eng = Engine(a.scope, a.base, a.model, a.max_hunts, a.max_turns, a.timeout, a.mock,
a.allow_intrusive, a.parallel, a.expand)
```
to:
```python
eng = Engine(a.scope, a.base, a.model, a.max_hunts, a.max_turns, a.timeout, a.mock,
a.allow_intrusive, a.parallel, a.expand, a.use_memory)
```
- [ ] **Step 5: Verify skip behavior on a second mock run**
Run:
```bash
# Run 1: populate the ledger (no memory used)
BUGHUNTER_MEMORY_DIR=/tmp/led2 python3 engine/engine.py --scope engine/engagement.example.json --base /tmp/eng2 --mock --hunt
# Run 2: same target WITH memory -> known-confirmed items are skipped + logged
BUGHUNTER_MEMORY_DIR=/tmp/led2 python3 engine/engine.py --scope engine/engagement.example.json --base /tmp/eng2b --mock --hunt --use-memory 2>&1 | grep "skipped via ledger"
```
Expected: run 2 prints at least one `hunt: skipped via ledger (known-confirmed): ...` line.
Run the coverage-safety check (default = no skips):
```bash
BUGHUNTER_MEMORY_DIR=/tmp/led2 python3 engine/engine.py --scope engine/engagement.example.json --base /tmp/eng2c --mock --hunt 2>&1 | grep -c "skipped via ledger" || true
```
Expected: `0` (no `--use-memory` = full coverage, no skips).
- [ ] **Step 6: Commit**
```bash
git add engine/engine.py
git commit -m "feat(engine): gated ledger skip filter in hunt loop + --use-memory flag"
```
---
### Task 5: Rewrite the memory command docs
**Files:**
- Modify: `commands/memory-gc.md`, `commands/pickup.md`, `commands/remember.md`, `commands/autopilot.md`
**Interfaces:** none (documentation). Goal: no dangling references to `tools.memory_gc`, `PatternDB`, `AuditLog`, `.claude/settings.json`, `patterns.jsonl`, `journal.jsonl` remain.
- [ ] **Step 1: Rewrite `commands/memory-gc.md`**
Replace the whole file with:
```markdown
---
name: memory-gc
description: Inspect or rotate the autopilot ledger JSONL files (findings.jsonl, negatives.jsonl). Caps file size and keeps N rotated backups so memory does not grow unbounded.
---
# /memory-gc
Garbage-collect the autopilot ledger (`~/.claude/bughunter/memory/`). Reports current sizes, rotates oversized files past a configurable cap, or purges old backups.
## Usage
```
/memory-gc # report only
/memory-gc --rotate # rotate files above 10 MB (default cap)
/memory-gc --rotate --max-mb 5 # custom cap
/memory-gc --purge-backups # delete all .1/.2/.3 backups
/memory-gc --dir <path> # scan a non-default ledger dir
```
## Implementation
The agent shells out to, from the repo root:
```bash
python3 engine/memory.py --gc [--rotate] [--purge-backups] [--dir PATH] [--max-mb N]
```
## Defaults
- **Rotation cap:** 10 MB per file · **Backups kept:** 3 (`<file>.1` newest → `<file>.3` oldest)
- **Scope:** every `*.jsonl` under the ledger dir (`findings.jsonl`, `negatives.jsonl`)
Rotation also fires automatically on append inside the ledger writer, so files stay bounded without any session-end hook.
```
- [ ] **Step 2: Rewrite `commands/pickup.md`**
Replace the whole file with:
```markdown
---
name: pickup
description: Pick up a previous hunt on a target — shows hunt history and untested surface from the autopilot ledger. Usage: /pickup target.com
---
# /pickup
Pick up where you left off on a target.
> **Renamed from `/resume`** — `/resume` is a reserved Claude Code command.
## What This Does
1. Reads the target rollup from `~/.claude/bughunter/memory/targets/<host>.json`.
2. Shows hunt history (sessions, last seen, tech stack).
3. Lists confirmed findings and the endpoints already tested.
## Usage
```
/pickup target.com
```
## Implementation
The agent reads the rollup directly:
```bash
python3 -c "import sys; sys.path.insert(0,'engine'); import memory, json; print(json.dumps(memory.rollup('target.com'), indent=2))"
```
## If No Previous Hunt
```
No ledger data for target.com. Run /recon then /autopilot target.com first.
```
```
- [ ] **Step 3: Rewrite `commands/remember.md`**
Replace the whole file with:
```markdown
---
name: remember
description: Optional manual note on a target or the last confirmed finding. Capture is automatic during autopilot; this is for extra context. Usage: /remember
---
# /remember
> **Capture is automatic.** During an engine/autopilot run, every confirmed finding is
> written to the ledger (`~/.claude/bughunter/memory/findings.jsonl`) with no action from you.
> `/remember` is only for adding an **optional manual note** — a technique detail or
> follow-up idea — that the automatic capture would not include.
## Usage
```
/remember # then describe the note; it is appended to the target's rollup
```
## What It Is Not
This is no longer the primary way findings enter memory — the engine captures those
deterministically. Use it sparingly for human context.
```
- [ ] **Step 4: Update `commands/autopilot.md` — memory section**
In `commands/autopilot.md`, replace the block under `## Safety Guarantees` line
`- **Every request** is logged to \`hunt-memory/audit.jsonl\`` — delete that single stale
line. Then add a new section immediately before `## After Autopilot`:
```markdown
## Ledger memory (token saving)
The autonomous loop can consult the **autopilot ledger** to skip provably-wasteful agent
calls — an item already **confirmed** on a prior run (carried forward, not re-tested) or a
vuln_class with strong **negative** history on this tech stack (`dead-class`).
- **`--paranoid` (default) / `--normal`:** memory is **off** — full coverage, no skips.
- **`--quick` / `--yolo`:** the command passes `--use-memory` to the engine — skips on.
- A vuln_class never hunted on this stack is **never** skipped, and every skip is logged
(`hunt: skipped via ledger (...)`). Coverage is never silently reduced.
- Capture is automatic; the ledger fills itself as you run. Manage size with `/memory-gc`.
```
- [ ] **Step 5: Verify no dangling references remain**
Run:
```bash
grep -rniE "tools\.memory_gc|PatternDB|AuditLog|patterns\.jsonl|journal\.jsonl|\.claude/settings\.json" commands/ || echo "CLEAN"
```
Expected: `CLEAN`.
- [ ] **Step 6: Commit**
```bash
git add commands/memory-gc.md commands/pickup.md commands/remember.md commands/autopilot.md
git commit -m "docs(commands): rewrite memory commands to the autopilot ledger"
```
---
### Task 6: Update credits + docs pointer, final verification
**Files:**
- Modify: `docs/credits.md`
**Interfaces:** none.
- [ ] **Step 1: Trim the vendored-commands claim in `docs/credits.md`**
`docs/credits.md` currently lists `/remember`, `/memory-gc`, `/pickup` among "Vendored slash
commands | shuvonsec/claude-bug-bounty (MIT)". Those three now describe an **original**
ledger backend, not the vendored interface. Edit the credits so they are no longer listed as
vendored: move `/remember`, `/memory-gc`, `/pickup` out of the vendored list, and add one line
under the author's original work, e.g.:
```markdown
- Autopilot ledger (`engine/memory.py`) + the `/remember`, `/memory-gc`, `/pickup` commands —
original design and implementation (cross-engagement capture + skip-decision for the engine
hunt loop). Not derived from any external memory implementation.
```
Leave the remaining genuinely-vendored entries unchanged.
- [ ] **Step 2: Full verification pass**
Run each and confirm:
```bash
python3 engine/memory.py # -> memory.py self-test: PASS
python3 engine/state.py # -> state.py self-test: PASS (regression)
python3 engine/memory.py --gc # -> exits 0
grep -rniE "tools\.memory_gc|PatternDB|AuditLog|patterns\.jsonl|journal\.jsonl" commands/ || echo CLEAN # -> CLEAN
```
Then the two-run skip proof from Task 4 Step 5 (known-confirmed skip appears with `--use-memory`, absent without).
- [ ] **Step 3: Commit**
```bash
git add docs/credits.md
git commit -m "docs(credits): ledger is original work, not vendored"
```
---
## Self-Review Notes
- **Spec coverage:** capture (Task 3) ✔; negatives + rollup (Tasks 1,3) ✔; skip filter known-confirmed + dead-class (Tasks 2,4) ✔; coverage gate `--use-memory` default off (Task 4) ✔; skips logged (Task 4) ✔; command rewrites (Task 5) ✔; gc CLI (Task 2) ✔; `state.py` untouched / no pyproject change ✔; credits (Task 6) ✔.
- **Type consistency:** `record_finding`/`record_run`/`skip_decision`/`rollup`/`gc` signatures identical across tasks; `skip_decision` return dict shape (`skip`/`reason`/`carry`) consistent; `carry["verdict"]` shape matches `Engagement.confirm(finding, verdict)`.
- **Coverage safety:** `--use-memory` default off (bare run = full coverage); unseen class never skipped (self-test asserts); every skip logged.
@@ -0,0 +1,222 @@
# Autopilot memory ("the ledger") — design
**Date:** 2026-08-09
**Status:** approved design (scoped to Option A), pre-implementation
**Repo:** elementalsouls/Claude-BugHunter
## Context
The repo ships three memory slash commands — `/remember`, `/memory-gc`, `/pickup` —
with **no backend**: they reference `python -m tools.memory_gc`, `PatternDB`, and
`patterns.jsonl`/`journal.jsonl` that do not exist. Dangling.
A full "cross-engagement memory that reshapes interactive ranking" was considered
and **cut on purpose** (see Scope decision). For a solo interactive hunter that
feature is marginal: an expert already carries stack→class intuition, tech-stack
overlap is a weak predictor, memory risks anchoring you on old patterns, and a
first-time user gets an empty store. The recall-into-ranking loop was speculative
complexity in the engine's hottest file.
**The one place memory earns its keep is autopilot.** The engine's autonomous hunt
loop (`engine.py --hunt --max-hunts N`) spends an agent (~real tokens) per
`(url, param, vuln_class)` worklist item, every run, on every target. An
autonomous agent has **no memory across runs**, so it will:
- re-confirm a finding it already confirmed on a prior run of the same target, and
- re-test a vuln_class that has *never* paid off on this tech stack across many targets.
Both are pure wasted spend. This design captures confirmed findings and dead
classes from the engine's own `state.json`, then lets the autonomous loop **skip
provably-wasteful agent calls** — a measurable token/time saving, not a UX feature.
## Scope decision (what this is and is NOT)
**IS:** capture (confirmed findings + negatives) + a conservative **skip filter in
the engine hunt loop only**, active only in token-saving autopilot modes.
**IS NOT:**
- No recall feeding interactive `rank`/`map` priority. The engine's normal ranking is untouched.
- No map annotations, no `/surface` changes, no interactive-hunter suggestions.
- No autopilot request-loop plumbing (rate limiter / circuit breaker / safe-method guard).
- No changes to Claude-OSINT.
- No Claude Code hooks; capture is engine code.
Original design on our engine — different module, schema, trigger, and purpose
from any external repo. No vendored code; no attribution owed.
## Why this is safe for coverage (the anti-tunnel-vision gate)
Memory that makes you *skip* work can make you miss a novel bug. So skipping is
gated hard:
- **Mode-gated.** `--paranoid` (default, new targets) **ignores memory entirely** —
full coverage, no skips. Skips apply only in `--quick` / `--yolo` (explicitly
token-optimizing, familiar-target modes) and behind an overridable `--no-memory` flag.
- **Never skip the unseen.** A vuln_class that has *never* been hunted on this stack
is never skipped — only classes with strong negative evidence.
- **Never silent.** Every skip is logged (`engine.log` + run summary: "skipped 6 agent
calls via ledger: 4 known-confirmed, 2 dead-class"). Coverage reduction is always visible.
- **Known-confirmed is carry-forward, not deletion.** A skipped known-confirmed item is
re-attached to the run's findings from the ledger, not dropped.
## Architecture
One new module, one purpose.
```
engine/memory.py autopilot ledger: capture + skip-decision (stdlib-only, import-isolated)
├── record_finding(engagement_name, host, tech_stack, finding, verdict) # capture (from confirm)
├── record_run(engagement_name, host, tech_stack, tested, confirmed) # capture (from report/done): negatives + rollup
├── skip_decision(host, tech_stack, item) -> {"skip":bool,"reason":str,"carry":finding|None}
├── rollup(host) -> dict # for /pickup
└── gc(action, max_mb, keep) # report/rotate/purge; also CLI __main__
```
`engine.py` imports it (`import memory` resolves via the existing
`sys.path.insert(0, dirname(__file__))`), calls capture at transitions it already
passes through, and calls `skip_decision` inside the hunt loop **only when memory
mode is on**. `engine/state.py` is unchanged (read-only consumer). Conventions
match the engine: stdlib-only, no `__init__.py`, in-file `_selftest()` under
`__main__` (like `state.py`/`recon.py`), deterministic. `memory.py` imports
nothing from siblings — `engine.py` passes state data in.
### Storage
Root: `~/.claude/bughunter/memory/` (override via `BUGHUNTER_MEMORY_DIR`).
| File | Purpose | Written by |
|---|---|---|
| `findings.jsonl` | one row per confirmed finding, cross-target | `record_finding` |
| `negatives.jsonl` | per-run `(host, stack_sig, vuln_class)` hunted → not confirmed | `record_run` |
| `targets/<host>.json` | per-target rollup for `/pickup` | `record_run` |
Append serialized under an `fcntl` advisory lock (correctness boundary — parallel
engine runs must not interleave a line). Files size-rotate at 10 MB, keep 3 backups,
rotation fires on append (bounded without a session-end hook) and via `/memory-gc`.
`tech_stack` is normalized to a sorted **stack signature** (lowercased token set) so
negatives aggregate across targets sharing a stack.
### Schemas
`findings.jsonl`:
```json
{"schema_version":1,"ts":"...Z","engagement":"acme-bb","host":"app.acme.com","url":"https://app.acme.com/api/users",
"param":"id","vuln_class":"idor","technique":"numeric_id_swap","severity":"high",
"verdict_reason":"confirmed PII exposure for arbitrary id","tech_stack":["graphql","nextjs","postgres"],
"evidence_ref":"evidence/idor-1.txt"}
```
Dedup key: `(host, url, param, vuln_class)`.
`negatives.jsonl` (one row per (run, host, class) that was tested and not confirmed):
```json
{"schema_version":1,"ts":"...","host":"app.acme.com","stack_sig":"graphql|nextjs|postgres","vuln_class":"xss"}
```
`targets/<host>.json`: `{schema_version, host, first_seen, last_seen, sessions, tech_stack,
tested[], confirmed[<finding subset>], untested[<worklist subset>]}`.
Light validation on write (required fields, ISO `ts`, `tech_stack` list-of-str); invalid →
skip with stderr warning, never crash a run.
## Capture (deterministic, engine-native)
Additive calls in `engine.py` at existing transitions:
1. **After `self.eng.confirm(c, v)`** (validate phase, engine.py:301): call
`memory.record_finding(name, host_of(c), tech_stack_of(host), c, v)`. `tech_stack`
from `state["targets"]` fingerprint. Every confirmed finding is captured automatically.
2. **In the `report` phase** (engine.py:328): `memory.record_run(name, host, tech_stack, state["tested"], state["confirmed"])`:
- Update `targets/<host>.json` (sessions++, tested, confirmed, untested worklist, tech_stack).
- For each `vuln_class` present in `tested` but absent from `confirmed` for the host,
append a `negatives.jsonl` row keyed by the host's stack signature.
Wiring capture into `confirm()`/`report` in code = fires on 100% of engine runs. No
manual `/remember`, no agent reliance.
## Skip filter (the autopilot saving — hunt loop only)
In the hunt phase, the loop is `wl = self.eng.worklist()[:self.max_hunts]` then per item
spawn an agent (engine.py:225–244). Add, **only when `self.use_memory` is true**, before
spawning:
```
d = memory.skip_decision(host, tech_stack, item)
if d["skip"]:
self.eng.log(f"hunt: skipped via ledger ({d['reason']}): {item_key}")
if d["carry"]: self.eng.confirm(d["carry"], d["carry"]["verdict"]) # carry-forward known finding
self.eng.mark_tested(item)
continue # no agent spent
```
`skip_decision` returns skip=True in exactly two cases:
- **known-confirmed:** `(host, url, param, vuln_class)` already in `findings.jsonl` →
`reason="known-confirmed"`, `carry=<that finding>`.
- **dead-class:** `(stack_sig, vuln_class)` has ≥ `DEAD_CLASS_THRESHOLD` negative rows
across targets **and zero** confirmations anywhere for that `(stack_sig, vuln_class)` →
`reason="dead-class"`, `carry=None`.
Otherwise skip=False. Thresholds/weights are named constants at the top of `memory.py`
(a calibration knob — which classes actually recur needs tuning a fixed model can't guess);
`DEAD_CLASS_THRESHOLD` defaults conservative (e.g. 5).
`engine.py` gets a **single `--use-memory` flag, default OFF** (so a bare engine run keeps
full coverage — safe by default). The mode→flag mapping lives where the modes live, in
`commands/autopilot.md`: `--paranoid`/default → do **not** pass `--use-memory` (full
coverage); `--quick`/`--yolo` → pass `--use-memory` (token-saving skips on). This keeps
`engine.py` unaware of the slash-command mode names — it only knows the one flag.
## Commands (rewritten to this system)
- **`/pickup <host>`** — read `targets/<host>.json`: history (sessions, findings), untested
surface, and the host's confirmed findings. No interactive recall/suggestions.
- **`/memory-gc`** — report / `--rotate` / `--purge-backups` / `--dir` / `--max-mb` via
`python3 engine/memory.py --gc ...` (script invocation, matching `python3 engine/engine.py`).
Rewrite doc to our filenames + entrypoint; drop `AuditLog`/`tools.memory_gc`/Stop-hook prose.
- **`/remember`** — optional manual annotation (note/tag on a host or the last confirmed
finding). No longer the primary path; capture is automatic.
- **`commands/autopilot.md`** — document the memory-skip behavior, the mode→`--use-memory`
mapping (`--quick`/`--yolo` on; `--paranoid`/default off), and that skips are logged, never silent.
## Testing
`engine/memory.py` `_selftest()` under `__main__`, asserting:
- `record_finding` dedup on `(host, url, param, vuln_class)`.
- `skip_decision` = **known-confirmed** (returns carry) for a captured finding's exact key.
- `skip_decision` = **dead-class** once negatives ≥ threshold with zero confirmations; and
**NOT dead-class** if a confirmation exists for that `(stack_sig, class)`, or if never hunted.
- rotation at the size cap; `gc --purge-backups` removes `.1`..`.3`.
Round-trip (integration): run the engine mock (`--mock --hunt`, mock verdicts at engine.py:381)
→ confirm a finding → assert a `findings.jsonl` row → run the same target again in `--yolo`
→ assert that item is **skipped as known-confirmed** (agent not spawned, finding carried) and
the skip is logged.
## Verification (from repo root)
1. `python engine/memory.py` → `memory.py self-test: PASS`.
2. `python engine/state.py` still passes (unchanged; regression guard).
3. `python3 engine/engine.py --scope engine/engagement.example.json --base /tmp/eng --mock --hunt`
twice on the same target: run 1 populates `findings.jsonl`; run 2 in `--yolo` logs
"skipped N via ledger" and spawns fewer agents.
4. `--paranoid` run performs **no** skips (full coverage regression).
5. `python3 engine/memory.py --gc` reports; `--gc --rotate` exits 0.
6. Grep clean: no dangling `tools.memory_gc`/`PatternDB` references remain in commands/.
## Files
**New:** `engine/memory.py`; this spec.
**Modified:** `engine/engine.py` (capture in confirm/report; gated skip filter + a single
`--use-memory` flag, default off); `commands/{pickup,memory-gc,remember,autopilot}.md`
(autopilot.md maps `--quick`/`--yolo` → `--use-memory`).
**Untouched:** `engine/state.py`, `pyproject.toml` (script-invoked, no packaging change), Claude-OSINT.
## Value framing (honest)
Value is **measurable agent-spend reduction on repeat/autonomous hunts** — not an
interactive-hunter feature. Expect ~zero benefit on a first run or a never-before-seen
stack (cold start), and real benefit only as the ledger accumulates repeat targets and
recurring stacks. Coverage is never silently reduced; `--paranoid` always runs full.
## Delivery
Branch `feat/ledger-memory`; one PR:
`feat(engine): autopilot ledger — capture findings + skip provably-wasteful agent calls`.
Squash-merge.
+31 -4
View File
@@ -25,6 +25,7 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from scope import Scope # noqa: E402
from state import Engagement # noqa: E402
import agent as A # noqa: E402
import memory # noqa: E402 (autopilot ledger; import-isolated)
# deterministic priority by class (impact-ish); used by rank
CLASS_WEIGHT = {"rce": 100, "sqli": 90, "ssrf": 85, "auth-bypass": 85, "idor": 80,
@@ -40,7 +41,7 @@ DEFAULT_PHASES = ["recon", "rank", "map"]
class Engine:
def __init__(self, scope_path, base, model, max_hunts, max_turns, timeout, mock=False,
allow_intrusive=False, parallel=3, expand=False):
allow_intrusive=False, parallel=3, expand=False, use_memory=False):
self.scope = Scope.load(scope_path)
self.eng = Engagement(base, self.scope.name)
self.model, self.max_hunts = model, max_hunts
@@ -48,6 +49,7 @@ class Engine:
self.allow_intrusive = allow_intrusive
self.expand = expand
self.parallel = max(1, parallel)
self.use_memory = use_memory
self.eng.log(f"engine start | scope={self.scope.name} in={self.scope.in_scope} "
f"out={self.scope.out_of_scope} seeds={self.scope.seeds} mock={mock} "
f"intrusive={'ALLOWED' if allow_intrusive else 'READ-ONLY (default)'}")
@@ -219,17 +221,31 @@ class Engine:
self.eng.log(f" worker error on {it.get('url')}: {e}")
yield it, {}
def _tech(self):
"""Sorted union of tech fingerprints across recon targets (the stack signature source)."""
return sorted({t for tg in self.eng.state.get("targets", []) for t in tg.get("tech", [])})
# ---------------- hunt ----------------
def hunt(self):
self.eng.set_phase("hunt")
wl = self.eng.worklist()[: self.max_hunts]
todo = []
tech = self._tech()
for it in wl:
if not self.scope.in_scope_host(it["url"]): # belt-and-suspenders scope gate
self.eng.log(f"hunt: REFUSING out-of-scope {it['url']} ({self.scope.reject_reason(it['url'])})")
self.eng.mark_tested(it)
else:
todo.append(it)
continue
if self.use_memory:
d = memory.skip_decision(memory._host(it["url"]), tech, it)
if d["skip"]:
self.eng.log(f"hunt: skipped via ledger ({d['reason']}): "
f"{it.get('vuln_class')}@{it['url']}")
if d["carry"]:
self.eng.confirm(d["carry"], d["carry"]["verdict"])
self.eng.mark_tested(it)
continue
todo.append(it)
self.eng.log(f"hunt: testing {len(todo)} item(s), {self.parallel} at a time")
for it, f in self._run_parallel(self._hunt_agent, todo):
if f and f.get("rate_limited"):
@@ -299,6 +315,10 @@ class Engine:
if v.get("severity"):
c["severity"] = v["severity"] # use the validator's CALIBRATED severity, not the hunt's
self.eng.confirm(c, v)
try:
memory.record_finding(self.scope.name, memory._host(c["url"]), self._tech(), c, v)
except Exception as e:
self.eng.log(f"memory: capture skipped ({e})")
self.eng.log(f"validate: CONFIRMED {c['vuln_class']} @ {c['url']} ({v.get('severity','')})"
+ (f" ⚠ scope-flagged: {', '.join(c['scope_warning'])}" if c.get('scope_warning') else ''))
else:
@@ -353,6 +373,10 @@ class Engine:
path = os.path.join(self.eng.dir, "report.md")
open(path, "w").write("\n".join(lines))
self.eng.log(f"report: wrote {len(c)} confirmed finding(s) -> {path}")
try:
memory.record_run(self.scope.name, self._tech(), self.eng.state["tested"], self.eng.state["confirmed"])
except Exception as e:
self.eng.log(f"memory: rollup skipped ({e})")
return path
def run(self, phases):
@@ -402,6 +426,9 @@ def main():
ap.add_argument("--expand", action="store_true",
help="OSINT bridge: enumerate subdomains, probe, and recon every live in-scope host "
"(the apex is usually just marketing; the real surface is on subdomains)")
ap.add_argument("--use-memory", action="store_true",
help="OPT-IN: skip provably-wasteful agent calls (known-confirmed / dead-class) "
"using the autopilot ledger. Default OFF = full coverage.")
a = ap.parse_args()
if a.phases:
phases = [p.strip() for p in a.phases.split(",") if p.strip()]
@@ -410,7 +437,7 @@ def main():
else:
phases = DEFAULT_PHASES # default: recon -> rank -> map, then STOP for the operator
eng = Engine(a.scope, a.base, a.model, a.max_hunts, a.max_turns, a.timeout, a.mock,
a.allow_intrusive, a.parallel, a.expand)
a.allow_intrusive, a.parallel, a.expand, a.use_memory)
eng.run(phases)
print("\n" + json.dumps(eng.eng.summary(), indent=2))
print("engagement dir:", eng.eng.dir)
+281
View File
@@ -0,0 +1,281 @@
#!/usr/bin/env python3
"""
memory.py — autopilot ledger: cross-engagement capture + skip decisions.
Lets the autonomous hunt loop skip provably-wasteful agent calls: re-confirming a
known finding, or re-testing a vuln_class that never pays off on this tech stack.
Capture is harvested from the engine's own state; skip is gated (off unless the
engine passes use_memory, i.e. token-saving modes only).
Storage (~/.claude/bughunter/memory/, override BUGHUNTER_MEMORY_DIR):
findings.jsonl one row per confirmed finding (cross-target)
negatives.jsonl (host, stack_sig, vuln_class) hunted -> not confirmed
targets/<host>.json per-target rollup for /pickup
stdlib-only; imports nothing from sibling engine modules.
"""
import fcntl
import json
import os
import sys
from datetime import datetime, timezone
from urllib.parse import urlparse
SCHEMA_VERSION = 1
MAX_BYTES = 10 * 1024 * 1024 # rotate a JSONL past 10 MB
KEEP = 3 # backups kept: .1 (newest) .. .3
DEAD_CLASS_THRESHOLD = 5 # negatives before a (stack,class) is "dead"
def _root():
return os.path.expanduser(os.environ.get("BUGHUNTER_MEMORY_DIR", "~/.claude/bughunter/memory"))
def _paths():
root = _root()
os.makedirs(os.path.join(root, "targets"), exist_ok=True)
return root
def _now():
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _host(url):
u = url or ""
if "://" not in u:
u = "//" + u
return (urlparse(u).hostname or "").lower().rstrip(".")
def _stack_sig(tech_stack):
return "|".join(sorted({t.lower() for t in (tech_stack or []) if t}))
def _rotate_if_needed(path):
try:
if os.path.getsize(path) < MAX_BYTES:
return
except FileNotFoundError:
return
oldest = f"{path}.{KEEP}"
if os.path.exists(oldest):
os.remove(oldest)
for i in range(KEEP - 1, 0, -1):
s, d = f"{path}.{i}", f"{path}.{i + 1}"
if os.path.exists(s):
os.replace(s, d)
os.replace(path, f"{path}.1")
def _append(path, obj):
line = json.dumps(obj, separators=(",", ":")) + "\n"
lock = str(path) + ".lock"
lfd = os.open(lock, os.O_WRONLY | os.O_CREAT, 0o644)
try:
fcntl.flock(lfd, fcntl.LOCK_EX)
_rotate_if_needed(path) # now inside the lock: no cross-process rename race
with open(path, "ab") as f:
f.write(line.encode("utf-8"))
finally:
fcntl.flock(lfd, fcntl.LOCK_UN)
os.close(lfd)
def _read(path):
out = []
if not os.path.exists(path):
return out
with open(path, encoding="utf-8") as f:
for ln in f:
ln = ln.strip()
if not ln:
continue
try:
out.append(json.loads(ln))
except json.JSONDecodeError:
continue
return out
def _valid_finding(e):
if not (e.get("host") and e.get("url") and e.get("vuln_class")):
return False
try:
datetime.fromisoformat(e["ts"].replace("Z", "+00:00"))
except Exception:
return False
return isinstance(e.get("tech_stack", []), list)
def record_finding(engagement, host, tech_stack, finding, verdict):
"""Capture one confirmed finding. Dedup on (host,url,param,vuln_class)."""
root = _paths()
fp = os.path.join(root, "findings.jsonl")
entry = {
"schema_version": SCHEMA_VERSION, "ts": _now(), "engagement": engagement,
"host": host or _host(finding.get("url", "")), "url": finding.get("url", ""),
"param": finding.get("param", ""), "vuln_class": finding.get("vuln_class", ""),
"technique": finding.get("technique", ""),
"severity": (verdict or {}).get("severity") or finding.get("severity", ""),
"verdict_reason": (verdict or {}).get("reason", ""),
"tech_stack": list(tech_stack or []),
"evidence_ref": finding.get("evidence_ref", ""),
}
if not _valid_finding(entry):
print(f"memory: skipping invalid finding {entry.get('url')!r}", file=sys.stderr)
return False
key = (entry["host"], entry["url"], entry["param"], entry["vuln_class"])
for e in _read(fp):
if (e.get("host"), e.get("url"), e.get("param"), e.get("vuln_class")) == key:
return False
_append(fp, entry)
return True
def _parse_key(k):
"""state 'url|param|class' -> (host, vuln_class)."""
url, _, rest = k.partition("|")
_param, _, cls = rest.partition("|")
return _host(url), cls
def _write_rollup(host, tech_stack, tested, confirmed):
root = _paths()
p = os.path.join(root, "targets", f"{host}.json")
prev = {}
if os.path.exists(p):
try:
prev = json.load(open(p))
except Exception:
prev = {}
hc_conf = [f for f in confirmed if _host(f.get("url", "")) == host]
doc = {
"schema_version": SCHEMA_VERSION, "host": host,
"first_seen": prev.get("first_seen", _now()), "last_seen": _now(),
"sessions": prev.get("sessions", 0) + 1, "tech_stack": list(tech_stack or []),
"tested": [k for k in tested if _parse_key(k)[0] == host],
"confirmed": [{"url": f.get("url"), "param": f.get("param", ""),
"vuln_class": f.get("vuln_class"), "severity": f.get("severity", "")}
for f in hc_conf],
}
tmp = p + ".tmp"
json.dump(doc, open(tmp, "w"), indent=2)
os.replace(tmp, p)
def record_run(engagement, tech_stack, tested, confirmed):
"""End-of-run capture: negatives for tested-but-unconfirmed classes + per-host rollup."""
root = _paths()
sig = _stack_sig(tech_stack)
tested_hc = {_parse_key(k) for k in tested}
confirmed_hc = {(_host(f.get("url", "")), f.get("vuln_class", "")) for f in confirmed}
neg = os.path.join(root, "negatives.jsonl")
for host, cls in tested_hc:
if cls and (host, cls) not in confirmed_hc:
_append(neg, {"schema_version": SCHEMA_VERSION, "ts": _now(),
"host": host, "stack_sig": sig, "vuln_class": cls})
for host in {h for h, _ in tested_hc} | {h for h, _ in confirmed_hc}:
if host:
_write_rollup(host, tech_stack, tested, confirmed)
def skip_decision(host, tech_stack, item):
"""Should the hunt loop skip this worklist item? Off-path returns skip=False."""
root = _paths()
url, param, cls = item.get("url", ""), item.get("param", ""), item.get("vuln_class", "")
# 1) known-confirmed -> carry the finding forward, no agent
for e in _read(os.path.join(root, "findings.jsonl")):
if (e.get("host"), e.get("url"), e.get("param"), e.get("vuln_class")) == (host, url, param, cls):
carry = {"url": url, "param": param, "vuln_class": cls,
"severity": e.get("severity", ""), "evidence": e.get("verdict_reason", ""),
"verdict_reason": e.get("verdict_reason", ""),
"verdict": {"real": True, "severity": e.get("severity", ""),
"reason": e.get("verdict_reason", ""), "from_memory": True}}
return {"skip": True, "reason": "known-confirmed", "carry": carry}
# 2) dead-class: >= threshold negatives for (stack_sig, cls) AND zero confirmations for it
sig = _stack_sig(tech_stack)
negs = sum(1 for e in _read(os.path.join(root, "negatives.jsonl"))
if e.get("stack_sig") == sig and e.get("vuln_class") == cls)
if negs >= DEAD_CLASS_THRESHOLD:
confirmed_here = any(_stack_sig(e.get("tech_stack", [])) == sig and e.get("vuln_class") == cls
for e in _read(os.path.join(root, "findings.jsonl")))
if not confirmed_here:
return {"skip": True, "reason": f"dead-class ({negs} negatives, 0 confirmed)", "carry": None}
return {"skip": False, "reason": "", "carry": None}
def rollup(host):
p = os.path.join(_root(), "targets", f"{host}.json")
return json.load(open(p)) if os.path.exists(p) else None
def gc(action="report", max_mb=10, keep=KEEP, root=None):
"""report | rotate | purge-backups over every *.jsonl in the ledger dir."""
base = root or _root()
rows = []
for dirpath, _dirs, names in os.walk(base):
for n in sorted(names):
if not n.endswith(".jsonl"):
continue
live = os.path.join(dirpath, n)
backups = [f"{live}.{i}" for i in range(1, keep + 1) if os.path.exists(f"{live}.{i}")]
live_sz = os.path.getsize(live) if os.path.exists(live) else 0
total = live_sz + sum(os.path.getsize(b) for b in backups)
if action == "rotate" and live_sz >= max_mb * 1024 * 1024:
_rotate_if_needed(live)
if action == "purge-backups":
for b in backups:
os.remove(b)
rows.append({"file": live, "live_bytes": live_sz, "total_bytes": total,
"backups": len(backups)})
return rows
def _selftest():
import tempfile, shutil
d = tempfile.mkdtemp()
os.environ["BUGHUNTER_MEMORY_DIR"] = d
try:
f = {"url": "http://a.com/x", "param": "id", "vuln_class": "idor", "severity": "high"}
assert record_finding("e1", "a.com", ["nextjs"], f, {"reason": "pii", "severity": "high"}) is True
assert record_finding("e1", "a.com", ["nextjs"], f, {"reason": "pii"}) is False # dedup
record_run("e1", ["nextjs"], ["http://a.com/y|q|xss", "http://a.com/x|id|idor"],
[{"url": "http://a.com/x", "vuln_class": "idor"}])
negs = _read(os.path.join(d, "negatives.jsonl"))
assert any(n["vuln_class"] == "xss" for n in negs) # xss tested, not confirmed -> negative
assert not any(n["vuln_class"] == "idor" for n in negs) # idor confirmed -> no negative
assert json.load(open(os.path.join(d, "targets", "a.com.json")))["sessions"] == 1
# skip_decision: known-confirmed carries the finding
skip_res = skip_decision("a.com", ["nextjs"], {"url": "http://a.com/x", "param": "id", "vuln_class": "idor"})
assert skip_res["skip"] and skip_res["reason"] == "known-confirmed" and skip_res["carry"]["verdict"]["real"] is True
# unseen class is never skipped
assert skip_decision("a.com", ["nextjs"], {"url": "http://a.com/x", "param": "id", "vuln_class": "rce"})["skip"] is False
# dead-class only after >= threshold negatives with zero confirmations
for _ in range(DEAD_CLASS_THRESHOLD):
record_run("e", ["django"], ["http://b.com/z|q|csrf"], [])
assert skip_decision("b.com", ["django"], {"url": "http://b.com/z", "param": "q", "vuln_class": "csrf"})["skip"] is True
# a confirmation on that (stack,class) revives it (not dead)
record_finding("e", "b.com", ["django"], {"url": "http://b.com/z", "param": "q", "vuln_class": "csrf", "severity": "low"}, {"reason": "x"})
assert skip_decision("b.com", ["django"], {"url": "http://b.com/z", "param": "q", "vuln_class": "csrf"})["reason"] == "known-confirmed"
print("memory.py self-test: PASS")
finally:
shutil.rmtree(d)
os.environ.pop("BUGHUNTER_MEMORY_DIR", None)
if __name__ == "__main__":
import argparse
ap = argparse.ArgumentParser(description="autopilot ledger")
ap.add_argument("--gc", action="store_true", help="run garbage-collection instead of the self-test")
ap.add_argument("--rotate", action="store_true", help="with --gc: rotate files over the cap")
ap.add_argument("--purge-backups", action="store_true", help="with --gc: delete all .1/.2/.3 backups")
ap.add_argument("--dir", default=None, help="ledger dir (default ~/.claude/bughunter/memory)")
ap.add_argument("--max-mb", type=int, default=10)
a = ap.parse_args()
if a.gc:
act = "purge-backups" if a.purge_backups else ("rotate" if a.rotate else "report")
for r in gc(act, a.max_mb, KEEP, a.dir):
print(f" {r['file']}: live={r['live_bytes']}B total={r['total_bytes']}B backups={r['backups']}")
else:
_selftest()