"Write ONE specific GitHub issue title",

"Write a tight, factual 2-3 sentence explanation",

"strict PII redactor",

"duplicate-detection bot",

"false_positive=TRUE",

"cyber": "Cybersecurity safety-filter false positive",

"aup": "AUP / Usage-Policy block (false positive)",

"harness": "Claude Code harness / auto-mode classifier denial",

"cyber": ("Legitimate, in-scope security / defensive / administration work was flagged by the "

"cybersecurity-topic safety classifier — pattern-matched on terminology, not on any "

"harmful intent. Securing or administering one's own systems is the opposite of attacking them."),

"aup": ("A benign, in-scope request was flagged as a Usage-Policy violation — a false positive on "

"ordinary, authorized work."),

"harness": ("The Claude Code auto-mode classifier denied an action during legitimate, authorized, "

"in-scope work. See the block reason below."),

t = text.lower()

# cyber = the cybersecurity-topic classifier flagged it. Match the flagging signature, not one

# exact sentence — Anthropic reworded it from "safety measures" to "Opus 4.8's safeguards" and

# the old fixed strings stopped matching, silently dropping real cyber blocks to 'other'.

# error_text() only ever feeds real API-error messages here, so this can't catch model refusals.

if (("cybersecurity" in t and "flag" in t)

or "safety filter detected cybersecurity" in t):

return "cyber"

if ("violate our usage policy" in t or "unable to respond to this request" in t

or "against our usage policy" in t or "usage policy violation" in t

or "content policy violation" in t):

return "aup"

if "overloaded" in t or "temporarily limiting" in t or "529" in t:

return "overloaded"

if "hit your limit" in t or "rate limit" in t or "429" in t or "· resets" in t:

return "limit"

if entry.get("type") != "user":

return None

content = (entry.get("message") or {}).get("content")

if isinstance(content, str):

return content.strip() or None

if isinstance(content, list):

if any(isinstance(b, dict) and b.get("type") == "tool_result" for b in content):

return None

text = "\n".join(b.get("text", "") for b in content

if isinstance(b, dict) and b.get("type") == "text").strip()

return text or None

if not entry.get("isApiErrorMessage"):

return None

content = (entry.get("message") or {}).get("content")

if isinstance(content, str):

return content

if isinstance(content, list):

return "\n".join(b.get("text", "") for b in content

if isinstance(b, dict) and b.get("type") == "text")

seen, out = set(), []

for o in f["occ"]:

if o["req"] and o["req"] not in seen:

seen.add(o["req"])

out.append(o)

"""ClAudit only auto-publishes server-side API false positives (cyber/aup) that carry a Request

"""Text of a Claude Code auto-mode-classifier denial (a tool_result error), or None."""

if entry.get("type") != "user":

return None

content = (entry.get("message") or {}).get("content")

if not isinstance(content, list):

return None

for b in content:

if not (isinstance(b, dict) and b.get("type") == "tool_result"):

continue

tc = b.get("content")

if isinstance(tc, str):

text = tc

elif isinstance(tc, list):

text = "\n".join(x.get("text", "") for x in tc

if isinstance(x, dict) and x.get("type") == "text")

else:

text = ""

low = text.lower()

if ("auto mode classifier" in low or "permission for this action was denied" in low

or "denied by the auto mode" in low or "action denied by auto mode" in low):

return text.strip()

"""Text of a normal (non-error) assistant turn, for capturing conversation leadup."""

if entry.get("type") != "assistant" or entry.get("isApiErrorMessage"):

return None

content = (entry.get("message") or {}).get("content")

if isinstance(content, str):

return content.strip() or None

if isinstance(content, list):

t = "\n".join(b.get("text", "") for b in content

if isinstance(b, dict) and b.get("type") == "text").strip()

return t or None

"""Parse ONE session transcript -> (findings list, log lines, logged-only counts). No shared

findings, log_lines = {}, []

counts = {"overloaded": 0, "limit": 0, "other": 0, "harness": 0}

last_prompt, recent = None, collections.deque(maxlen=6)

try:

with open(path, encoding="utf-8") as fh:

for line in fh:

try:

entry = json.loads(line)

except ValueError:

continue

hp = human_text(entry)

if hp:

last_prompt = hp

recent.append(("user", hp[:300]))

continue

hd = harness_denial(entry)

if hd:

ts = entry.get("timestamp", "")

log_lines.append(json.dumps({"kind": "harness", "ts": ts, "session": name, "req": None}))

if not REPORT_HARNESS: # log-only: local classifier decision, not an API FP

counts["harness"] += 1

continue

s = sig("harness", hd[:300])

f = findings.setdefault(s, {"sig": s, "kind": "harness",

"prompt": last_prompt or "(no preceding prompt)",

"occ": [], "block_text": hd, "leadup": list(recent)})

f["occ"].append({"req": None, "ts": ts, "session": name, "proj": os.path.basename(root)})

continue

err = error_text(entry)

if err is None:

at = assistant_text(entry)

if at:

recent.append(("assistant", at[:300]))

continue

if last_prompt and any(mk in last_prompt for mk in INTERNAL_PROMPTS):

continue # ClAudit's own LLM prompt got blocked — never self-report

kind = classify(err)

ts = entry.get("timestamp", "")

m = REQ_ID.search(err)

req = m.group(0) if m else None

log_lines.append(json.dumps({"kind": kind, "ts": ts, "session": name, "req": req}))

if kind not in FILE_KINDS:

counts[kind] = counts.get(kind, 0) + 1

continue

prompt = last_prompt or "(triggering prompt not recoverable)"

s = sig(kind, prompt)

f = findings.setdefault(s, {"sig": s, "kind": kind, "prompt": prompt,

"occ": [], "block_text": err, "leadup": list(recent)})

f["occ"].append({"req": req, "ts": ts, "session": name, "proj": os.path.basename(root)})

except OSError:

pass

"""Walk all sessions -> (findings dict[sig]->finding, logged-only counts). Incremental: each

if ttl and _SCAN_CACHE["val"] is not None and time.monotonic() - _SCAN_CACHE["t"] < ttl:

return _SCAN_CACHE["val"]

findings, logged, log_lines = {}, {"overloaded": 0, "limit": 0, "other": 0, "harness": 0}, []

seen = set()

for root, _, files in os.walk(PROJECTS):

for name in files:

if not name.endswith(".jsonl"):

continue

path = os.path.join(root, name)

seen.add(path)

try:

mtime = os.path.getmtime(path)

except OSError:

continue

cached = _FILE_CACHE.get(path)

if cached and cached[0] == mtime:

ff, fl, fc = cached[1], cached[2], cached[3]

else:

ff, fl, fc = _parse_file(path, name, root)

_FILE_CACHE[path] = (mtime, ff, fl, fc)

log_lines.extend(fl)

for k, v in fc.items():

logged[k] = logged.get(k, 0) + v

for f in ff: # merge by signature across files

ex = findings.get(f["sig"])

if ex:

ex["occ"].extend(f["occ"])

else:

findings[f["sig"]] = {**f, "occ": list(f["occ"])} # copy so cache isn't mutated

for p in [p for p in _FILE_CACHE if p not in seen]: # evict deleted files

del _FILE_CACHE[p]

os.makedirs(STATE_DIR, exist_ok=True)

with open(ERROR_LOG, "w", encoding="utf-8") as fh:

fh.write("\n".join(log_lines) + ("\n" if log_lines else ""))

result = (findings, logged)

_SCAN_CACHE["t"], _SCAN_CACHE["val"] = time.monotonic(), result

"""Best-effort cross-platform liveness check."""

if pid <= 0:

return False

if platform.system() == "Windows":

out = subprocess.run(["tasklist", "/FI", f"PID eq {pid}"],

capture_output=True, text=True).stdout

return str(pid) in out

try:

os.kill(pid, 0)

except ProcessLookupError:

return False

except PermissionError:

return True

except OSError:

return False

try:

if os.path.exists(LOCK_FILE) and open(LOCK_FILE).read().strip() == str(os.getpid()):

os.remove(LOCK_FILE)

except OSError:

"""Self-awareness: ensure only ONE watcher process runs at a time. Returns True if we

os.makedirs(STATE_DIR, exist_ok=True)

if os.path.exists(LOCK_FILE):

try:

other = int((open(LOCK_FILE).read().strip() or "0"))

except (OSError, ValueError):

other = 0

if other and other != os.getpid() and _pid_alive(other):

return False

with open(LOCK_FILE, "w") as fh:

fh.write(str(os.getpid()))

atexit.register(_release_singleton)

try:

with open(CONFIG_FILE, encoding="utf-8") as fh:

return json.load(fh)

except (OSError, ValueError):

try:

with open(STATE_FILE, encoding="utf-8") as fh:

data = json.load(fh)

except (OSError, ValueError):

return {}

if isinstance(data, list): # migrate old set-of-sigs format

return {s: {"issue": None, "url": None, "reqs": []} for s in data}

("cloud-iam", re.compile(r"\b(entra|azure ad|tenant|conditional access|app role|app registration|service principal|okta|oauth|iam)\b", re.I)),

("defensive-hardening", re.compile(r"\b(harden|hardening|mfa|firewall|edr|blue team|patch|cis benchmark|least privilege|lockdown)\b", re.I)),

("reverse-engineering", re.compile(r"\b(disassemble|decompile|ghidra|ida pro|opcode|unpack|reverse engineer)\b", re.I)),

("infra-devops", re.compile(r"\b(kubernetes|docker|terraform|nginx|ansible|systemd|hypervisor|reverse proxy)\b", re.I)),

("web-security", re.compile(r"\b(xss|sql injection|sqli|csrf|ssrf|burp|owasp)\b", re.I)),

("offensive-pentest", re.compile(r"\b(exploit|payload|msfvenom|metasploit|shellcode|reverse shell|\bc2\b|privilege escalation|lateral movement)\b", re.I)),

("crypto-secrets", re.compile(r"\b(encrypt|decrypt|private key|certificate|tls|keystore)\b", re.I)),

("malware-forensics", re.compile(r"\b(malware|forensic|memory dump|yara|incident response|ioc)\b", re.I)),

"""Heuristic work-domain tag for the report (helps triage which classifier over-fires)."""

text = (f.get("prompt", "") + " " + f.get("block_text", "") + " "

+ " ".join(t for _, t in (f.get("leadup") or []))).lower()

for cat, pat in DOMAIN_PATTERNS:

if pat.search(text):

return cat

"i can't write", "i cannot write", "i won't write", "i will not write", "i refuse to",

"i won't do that", "i won't file", "i can't file", "i'm not able to write",

"not a false positive", "is a true positive", "was a true positive", "block is accurate",

"block is correct", "block was accurate", "block was correct", "policy block is accurate",

"""True if LLM output is a refusal / editorializes that the block was CORRECT — never post it;

low = (text or "").lower()

"""True if the model is asking for / complaining about missing context instead of writing the

low = (text or "").lower()

return any(m in low for m in (

"appears to be missing", "is blank", "field is blank", "could you paste", "please paste",

"i don't have the specific", "the context you provided", "missing from the context",

"without the bot", "no bot comment", "can't reference", "cannot reference",

reqs = reqs_of(f)

first_ts = min((o["ts"] for o in f["occ"] if o["ts"]), default="")

sessions = len({o["session"] for o in f["occ"]})

projects = sorted({project_label(o["proj"]) for o in f["occ"] if o.get("proj")})

proj = projects[0].rstrip("/").split("/")[-1] if projects else "unknown"

# [Bug] first (issue template), then kind; ClAudit-tagged + a distinct discriminator.

if f["kind"] == "harness":

m = re.search(r"Reason:\s*(.+?)(?:\.\s|\n|$)", f["block_text"])

reason = (m.group(1).strip()[:70] if m else f"#{f['sig']}")

title = f"[Bug][harness] ClAudit: auto-mode classifier denied — {reason}"

else:

lead = reqs[0]["req"] if reqs else f"#{f['sig']}"

title = f"[Bug][{f['kind']}] ClAudit false-positive in {proj} — {lead}"

req_lines = "\n".join(f"- `{o['req']}` ({o['ts']})" for o in reqs) or "- (no Request ID captured)"

note_clean, _ = scrub(note or DEFAULT_NOTE)

# Full PII scrub on the block message (was token-only — leaked IPs/hosts/paths).

block_clean = scrub(TOKEN.sub("token=[SCRUBBED]", f["block_text"]))[0].strip()[:500]

if f["kind"] == "harness":

# Show ONLY the classifier's stated reason — never the quoted command (infra/sketch risk).

m = re.search(r"Reason:\s*(.+?)(?:\.\s|\n|If you have|$)", f["block_text"])

block_clean = (scrub(m.group(1).strip())[0][:300] if m

else "(auto-mode classifier denial — see Request IDs)")

leadup = f.get("leadup") or []

leadup_md = "\n".join(f"**{role}:** {scrub(re.sub(chr(10), ' ', txt))[0]}"

for role, txt in leadup) or "_(not captured)_"

why_text = WHY.get(f["kind"], WHY["cyber"])

neutral = False # when the LLM refuses, file FACTS ONLY (type + Request IDs)

if claudit.BURN_TOKENS: # spend tokens to craft a bespoke, specific title + explanation

ctx = f"work domain: {categorize(f)}\nblock message: {block_clean}\nconversation leadup:\n{leadup_md}"

bt = claudit.llm_compose(

f"Write ONE specific GitHub issue title (max ~95 chars) that starts EXACTLY with "

f"'[Bug][{f['kind']}]' and describes the concrete legitimate work this Claude Code safety "

f"block wrongly stopped. Output ONLY that title line — no preamble, no 'here is', no quotes, "

f"no explanation.", ctx)

if bt:

# the model sometimes adds a chatty preamble line — take the line that's actually a title

lines = [ln.strip().strip('"').strip("`") for ln in bt.splitlines() if ln.strip()]

cand = next((ln for ln in lines if ln.lower().startswith("[bug]")), "")

if cand: # only trust a real title; else keep the deterministic one

cand = scrub(cand)[0][:110]

lead_req = reqs[0]["req"] if reqs else ""

title = cand if (not lead_req or lead_req in cand) else f"{cand} ({lead_req})"

bw = claudit.llm_compose(

"Write a tight, factual 2-3 sentence explanation of why this Claude Code safety block is a "

"false positive on legitimate, in-scope work — suitable for a bug report to Anthropic.", ctx)

if bw and not _is_refusal(bw): # never post the model's refusal/editorial

why_text = scrub(bw)[0]

elif bw and _is_refusal(bw): # model wouldn't vouch -> assert NOTHING, file facts only

neutral = True

is_harness = f["kind"] == "harness"

recur = (f"Recurred **{len(f['occ'])}×** across {sessions} session(s); first seen {first_ts}.")

# harness denials are LOCAL auto-mode-classifier blocks — no server-side Request ID exists.

if reqs:

reqs_block = f"### Request IDs (lookup-able server-side)\n{req_lines}"

verify = "the triggering request is verifiable server-side via the Request ID(s) below"

else:

reqs_block = (

"### No Request ID (auto-mode-classifier denial)\n"

"This is a Claude Code **auto-mode-classifier denial** — a local harness block, not a "

"server-side API error — so it carries **no Request ID** to look up. The classifier's "

"stated reason is in the Block message below.")

verify = "the classifier's stated reason is in the Block message below"

blocked_phrase = ("A Claude Code **auto-mode-classifier denial** stopped authorized, in-scope work"

if is_harness else

"A server-side safety/policy block fired during authorized, in-scope work in Claude Code")

if neutral:

kindphrase = ("A Claude Code auto-mode-classifier denial" if is_harness

else f"A server-side **{f['kind']}** block")

why_block = f"{kindphrase} fired in Claude Code. No rationale is asserted — {verify}. {recur}"

note_block = ""

else:

why_block = (f"### Why this is a false positive\n{why_text}\n\n"

f"{blocked_phrase}. Filing as a false positive. {recur}")

note_block = f"\n### In-scope justification\n{note_clean}\n"

crossref_block = (f"\n### Related reports (same work session, linked)\n{scrub(crossref)[0]}\n"

if crossref else "")

# Structured triage line — ClAudit can't add GitHub labels (no triage perm on the repo), so it

# puts the categorization in the body for maintainers to sort on.

repro = ("yes — server-side via the Request ID(s) below" if reqs

else "local auto-mode-classifier denial (no server-side Request ID)")

triage = (f"**Triage:** kind `{f['kind']}` · domain `{categorize(f)}` · "

f"severity **session-halted** (blocked authorized work) · reproducible: {repro}")

body = f"""{triage}

# Title is built from already-scrubbed parts + the Request ID — regex/denylist scrub only

# (never the LLM, so the Request ID survives). Body gets the opt-in LLM pass.

"""Append a filed issue to the local issues DB (with PII-scrubbed leadup)."""

rec = {

"sig": f["sig"], "kind": f["kind"], "repo": repo, "url": url,

"version": __version__,

"issue": url.rsplit("/", 1)[-1],

"reqs": [o["req"] for o in reqs_of(f)],

"first_ts": min((o["ts"] for o in f["occ"] if o["ts"]), default=""),

"projects": sorted({project_label(o["proj"]) for o in f["occ"] if o.get("proj")}),

"chain": _chain_key(f), # session-precise chain key (no PII: session is a local UUID)

"leadup": [[role, scrub(txt)[0]] for role, txt in (f.get("leadup") or [])],

}

os.makedirs(STATE_DIR, exist_ok=True)

with open(ISSUES_DB, "a", encoding="utf-8") as fh:

"""Run `gh <args>` and parse stdout as JSON; return None on any failure."""

try:

out = subprocess.run(["gh", *args], capture_output=True, text=True, timeout=30)

return json.loads(out.stdout) if out.returncode == 0 and out.stdout.strip() else None

except Exception:

"""Solidarity 👍: react thumbs-up on OTHER ClAudit users' open cyber/aup false-positive issues to

me = me or gh_login()

done = state.setdefault("__amplified__", [])

seen = set(done)

issues = _gh_json(["issue", "list", "-R", repo, "--state", "open", "--limit", str(limit),

"--search", '"Filed automatically by ClAudit"',

"--json", "number,author,title"]) or []

n = 0

for it in issues:

num, author = it["number"], (it.get("author") or {}).get("login", "")

title = (it.get("title", "") or "").lower()

if author == me or not author or str(num) in seen:

continue # skip your own + already-reacted

if "[cyber]" not in title and "[aup]" not in title:

continue # real API false positives only

r = subprocess.run(["gh", "api", f"repos/{repo}/issues/{num}/reactions",

"-f", "content=+1"], capture_output=True, text=True)

if r.returncode == 0:

done.append(str(num)); seen.add(str(num)); n += 1

time.sleep(0.5)

if n:

save_state(state)

"""{'plus','minus','eyes','total'} from the pinned poll issue's reaction summary."""

d = _gh_json(["api", f"/repos/{POLL_REPO}/issues/{POLL_ISSUE}",

"--jq", '{plus: .reactions."+1", minus: .reactions."-1", eyes: .reactions.eyes}']) or {}

c = {k: int(d.get(k) or 0) for k in ("plus", "minus", "eyes")}

c["total"] = c["plus"] + c["minus"] + c["eyes"]

"""Cast/switch the user's vote. `choice` in {'plus','minus','eyes'}. Enforces one vote per

content = dict((o[0], o[1]) for o in POLL_OPTS)[choice]

me = me or gh_login()

subprocess.run(["gh", "api", "-X", "POST",

f"/repos/{POLL_REPO}/issues/{POLL_ISSUE}/reactions", "-f", f"content={content}"],

capture_output=True, text=True)

valid = {o[1] for o in POLL_OPTS}

for r in (_gh_json(["api", "--paginate",

f"/repos/{POLL_REPO}/issues/{POLL_ISSUE}/reactions"]) or []):

if ((r.get("user") or {}).get("login") == me and r.get("content") in valid

and r.get("content") != content):

subprocess.run(["gh", "api", "-X", "DELETE",

f"/repos/{POLL_REPO}/issues/{POLL_ISSUE}/reactions/{r['id']}"],

capture_output=True, text=True)

"""Create an issue, or comment if it already exists with fresh Request IDs.

rec = state.get(f["sig"])

cur_reqs = [o["req"] for o in reqs_of(f)]

if rec is None:

# Reserve the signature and persist BEFORE the network call so a concurrent pass

# (or a crash-restart) can never double-file the same finding.

state[f["sig"]] = {"issue": None, "url": None, "kind": f["kind"], "reqs": cur_reqs}

save_state(state)

title, body = build_issue(f, note)

try:

url = gh_create(repo, title, body)

except Exception:

state.pop(f["sig"], None) # release on failure so it can retry later

save_state(state)

raise

state[f["sig"]].update(issue=url.rsplit("/", 1)[-1], url=url)

log_issue(f, repo, url)

return ("new", title, url)

fresh = [r for r in cur_reqs if r not in rec.get("reqs", [])]

if fresh and rec.get("issue"):

gh_comment(repo, rec["issue"], "Recurred again. Additional Request IDs:\n" +

"\n".join(f"- `{r}`" for r in fresh))

rec["reqs"] = rec.get("reqs", []) + fresh

return ("updated", f"#{rec['issue']}", rec.get("url"))

findings, _ = scan()

for s, f in findings.items():

state.setdefault(s, {"issue": None, "url": None, "kind": f["kind"],

"reqs": [o["req"] for o in reqs_of(f)]})

state["__baselined__"] = True

save_state(state)

"""Never file the backlog: if we've never baselined, mark all current findings

if state.get("__baselined__"):

return 0

n = baseline(state)

if announce:

announce("baselined", f"{n} existing blocks marked seen (not filed)", "")

"""Best-effort desktop notification across Linux / macOS / Windows."""

try:

if shutil.which("notify-send"):

subprocess.run(["notify-send", "-a", "claudit", "-i", ICON, title, body], check=False)

elif platform.system() == "Darwin":

subprocess.run(["osascript", "-e",

f'display notification {json.dumps(body)} with title {json.dumps(title)}'],

check=False)

elif platform.system() == "Windows":

ps = (f"$ws=New-Object -ComObject WScript.Shell;"

f"[void]$ws.Popup({json.dumps(body)},5,{json.dumps(title)},64)")

subprocess.run(["powershell", "-NoProfile", "-Command", ps], check=False)

except Exception:

"""Toast with a clickable 'Report it' button (Linux/notify-send only); runs on_report()

if not shutil.which("notify-send"):

_toast(title, body)

return

res = subprocess.run(

# normal urgency + an explicit 20s expiry so the toast fades on its own if ignored

# (critical urgency made it persist forever); the GUI still queues them for later filing.

["notify-send", "-a", "claudit", "-i", ICON, "-u", "normal", "-t", "20000",

"-A", "report=Report it", "-A", "dismiss=Dismiss", title, body],

capture_output=True, text=True)

if res.stdout.strip() == "report":

"""Pop an actionable toast for whatever is currently queued (in a background thread)."""

n = len(pending_sigs(state))

if not n:

return

threading.Thread(target=notify_action, daemon=True, args=(

f"ClAudit: {n} false-positive block(s) queued",

"Click ‘Report it’ to file these to GitHub now.",

lambda: notify("filed", f"{file_pending(state, repo, False, delay, notify)} reported", repo),

"""Notify-only pass: detect NEW findings (not seen, not already pending), queue them,

findings, _ = scan()

pend = state.setdefault("__pending__", [])

fresh = [f for s, f in findings.items()

if s not in state and s not in pend and not s.startswith("__") and should_file(f)]

if fresh:

for f in fresh:

pend.append(f["sig"])

save_state(state)

on_detect(fresh)

"""True if the finding should be filed. By default ClAudit files EVERY genuine block — the

if not GATE:

return True

ctx = " ".join(t for _, t in (f.get("leadup") or []))

ok, reason = claudit.llm_is_false_positive(f["kind"], f.get("block_text", ""), ctx)

if not ok:

state[f["sig"]] = {"issue": None, "skipped": (reason or "judged a correct block")[:140],

"kind": f["kind"], "reqs": [o["req"] for o in reqs_of(f)]}

save_state(state)

print(f" skip {f['sig']} — not a false positive: {reason[:80]}", file=sys.stderr)

"""Auto-post pass: file every NEW finding, comment recurrences. Files nothing for

findings, _ = scan(ttl=8)

acted = 0

for f in findings.values():

if f["sig"] not in state and (not should_file(f) or not passes_gate(f, state)):

continue # only file NEW cyber/aup blocks that carry a Request ID

try:

action, ref, url = file_one(f, "", repo, state)

except Exception as e:

print(f" ! {f['sig']}: {e}", file=sys.stderr)

continue

if action:

save_state(state)

on_event(action, ref, url)

acted += 1

time.sleep(delay)

"""Which chain a finding belongs to. Keyed on the actual Claude Code SESSION (one conversation /

sessions = [o.get("session") for o in f["occ"] if o.get("session")]

if sessions:

return "sess:" + collections.Counter(sessions).most_common(1)[0][0]

projs = [o.get("proj") for o in f["occ"] if o.get("proj")]

"""A view of finding f narrowed to one Request ID — so each API ID files as its own bespoke

g = dict(f)

g["occ"] = [o for o in f["occ"] if o.get("req") == req] or f["occ"][:1]

"""Forward cross-reference body line: the prior bespoke issues from the same work session."""

if not chain:

return ""

return ("Distinct false-positive blocks from the same work session, each its own report:\n"

"""Auto-file on a dwell. Each NEW cyber/aup Request ID is held for `dwell` seconds (repeats in

dwell = DWELL_SECONDS if dwell is None else dwell

findings, _ = scan(ttl=8)

now = time.time()

seen = state.setdefault("__dwell_seen__", [])

hold = state.setdefault("__dwell_hold__", {}) # req -> first_seen_epoch

filed = state.setdefault("__filed_reqs__", {}) # req -> issue number

skipped = state.setdefault("__skipped_reqs__", [])

chains = state.setdefault("__proj_chain__", {}) # proj -> [issue numbers], the linked string

# req -> finding for every fileable (cyber/aup + Request ID) occurrence currently in the corpus

current = {}

for f in findings.values():

if not should_file(f):

continue

for o in f["occ"]:

if o.get("req"):

current[o["req"]] = f

seen_set, filed_set, skip_set = set(seen), set(filed), set(skipped)

# First run on this state: baseline everything already present so we don't flood the backlog.

if not state.get("__dwell_baselined__"):

seen[:] = sorted(set(seen) | set(current) | filed_set)

state["__dwell_baselined__"] = True

save_state(state)

return 0

# Drain the deprecated manual queue (__pending__) into the dwell pipeline so it doesn't hang as

# un-processable QUEUED rows — each pending block's Request IDs get held + filed like any other.

pend = state.get("__pending__", [])

if pend:

for s in pend:

f = findings.get(s)

if not (f and should_file(f)):

continue

for o in f["occ"]:

req = o.get("req")

if req and req not in filed_set and req not in skip_set:

hold.setdefault(req, now)

seen_set.add(req)

if req not in seen:

seen.append(req)

state["__pending__"] = []

save_state(state)

# Register genuinely new Request IDs into the dwell hold.

for req in current:

if req not in seen_set and req not in filed_set and req not in skip_set:

hold.setdefault(req, now)

seen.append(req)

# A create that keeps failing must NOT re-judge+re-compose every 30s tick (that's the only path

# that could burn tokens "like crazy" at idle) — back a failed Request ID off for FAIL_COOLDOWN.

fails = state.setdefault("__dwell_fail__", {}) # req -> last failed-create epoch