From 4ff20c12c23cc733a993c78fa457e9fda3ead939 Mon Sep 17 00:00:00 2001 From: Md Shariar Shanaz Shuvon <83355567+shuvonsec@users.noreply.github.com> Date: Wed, 16 Sep 2026 01:43:56 +0800 Subject: [PATCH 1/3] feat(contributors): add contextual contribution and learning system --- COMMANDS-QUICK-REF.md | 3 + cli/main.py | 46 +++- docs/contributors/README.md | 76 ++++++ docs/engagement.md | 4 + engines/contributors/__init__.py | 89 +++++++ engines/contributors/cli.py | 220 ++++++++++++++++ engines/contributors/extract.py | 114 +++++++++ engines/contributors/hooks.py | 123 +++++++++ engines/contributors/milestones.py | 139 +++++++++++ engines/contributors/prepare.py | 203 +++++++++++++++ engines/contributors/privacy.py | 262 ++++++++++++++++++++ engines/contributors/quality.py | 131 ++++++++++ engines/contributors/recognize.py | 190 ++++++++++++++ engines/contributors/sanitize.py | 129 ++++++++++ engines/contributors/schema.py | 221 +++++++++++++++++ engines/contributors/signals.py | 208 ++++++++++++++++ engines/contributors/suggest.py | 250 +++++++++++++++++++ engines/contributors/templates.py | 270 ++++++++++++++++++++ engines/engagement/__init__.py | 4 + engines/engagement/messages.py | 49 ++++ engines/engagement/schema.py | 4 + tests/test_contributors.py | 386 +++++++++++++++++++++++++++++ 22 files changed, 3119 insertions(+), 2 deletions(-) create mode 100644 docs/contributors/README.md create mode 100644 engines/contributors/__init__.py create mode 100644 engines/contributors/cli.py create mode 100644 engines/contributors/extract.py create mode 100644 engines/contributors/hooks.py create mode 100644 engines/contributors/milestones.py create mode 100644 engines/contributors/prepare.py create mode 100644 engines/contributors/privacy.py create mode 100644 engines/contributors/quality.py create mode 100644 engines/contributors/recognize.py create mode 100644 engines/contributors/sanitize.py create mode 100644 engines/contributors/schema.py create mode 100644 engines/contributors/signals.py create mode 100644 engines/contributors/suggest.py create mode 100644 engines/contributors/templates.py create mode 100644 tests/test_contributors.py diff --git a/COMMANDS-QUICK-REF.md b/COMMANDS-QUICK-REF.md index 86056ff..ca57c13 100644 --- a/COMMANDS-QUICK-REF.md +++ b/COMMANDS-QUICK-REF.md @@ -39,6 +39,7 @@ | Regenerate reports | `/axguard-report` | | Add CI gate | `/axguard-ci` | | About / engagement prefs | `axguard about` · `axguard engage disable` | +| Contribute / privacy (local) | `axguard contribute …` · `axguard privacy …` | ## Default pipeline @@ -91,6 +92,8 @@ axguard investigate . --finding FINDING_ID axguard investigate --explain FINDING_ID axguard about axguard engage disable +axguard privacy status +axguard contribute status axguard data discover axguard data report fixtures/data_pipeline axguard twin build . # Security Twin — see docs/twin/README.md diff --git a/cli/main.py b/cli/main.py index e49b247..9a696dd 100644 --- a/cli/main.py +++ b/cli/main.py @@ -699,6 +699,11 @@ def _mem_common(p: argparse.ArgumentParser) -> None: engage_sub.add_parser("enable", help="Re-enable promotional messaging") engage_sub.add_parser("dismiss", help="Dismiss the latest support ask (cooldown)") + from engines.contributors.cli import add_contribute_parser, add_privacy_parser + + add_privacy_parser(sub) + add_contribute_parser(sub) + github_cmd = sub.add_parser( "github", help="GitHub Security Bot — setup / validate / test / status", @@ -782,6 +787,7 @@ def _gh_common(p: argparse.ArgumentParser) -> None: GitHub Security Bot axguard github … | docs/github/README.md About AXGuard axguard about Engagement prefs axguard engage disable | enable | dismiss + Contribute / privacy (local) axguard contribute … · axguard privacy … First look at a new codebase /axguard-threat-model → /axguard-audit Secrets / auth / inject /axguard-secrets · /axguard-auth · /axguard-inject SQL / SSTI / path /axguard-sql · /axguard-ssti · /axguard-path @@ -797,7 +803,7 @@ def _gh_common(p: argparse.ArgumentParser) -> None: .findings/axguard/axguard-report.{html,md,json} Cheat sheet: COMMANDS-QUICK-REF.md -Docs: docs/engagement.md (local prefs, no telemetry) +Docs: docs/engagement.md · docs/contributors/README.md (local prefs, no telemetry) """.strip() @@ -885,6 +891,16 @@ def main(argv: list[str] | None = None) -> int: return 2 return 0 + if args.command == "privacy": + from engines.contributors.cli import run_privacy_command + + return run_privacy_command(args) + + if args.command == "contribute": + from engines.contributors.cli import run_contribute_command + + return run_contribute_command(args) + if args.command == "github": if not getattr(args, "no_banner", False): print_banner() @@ -1035,7 +1051,18 @@ def main(argv: list[str] | None = None) -> int: print() print(render_report(result, "md")) if not getattr(args, "no_engage", False): - _print_engagement(emit_after_audit(result)) + eng = emit_after_audit(result) + if eng: + _print_engagement(eng) + else: + from engines.contributors.hooks import emit_after_audit_soft + + _print_engagement( + emit_after_audit_soft( + result, + no_engage=False, + ) + ) return 1 if _should_fail(result["findings"], args.fail_on) else 0 if args.command == "surface": @@ -1134,6 +1161,21 @@ def main(argv: list[str] | None = None) -> int: print(f" REQUIRES_REVIEW {summary.get('REQUIRES_REVIEW', 0)}") print(f" json {paths['json']}") print(f" md {paths['markdown']}") + if not getattr(args, "no_engage", False): + from engines.contributors.hooks import emit_after_adversary_soft + + # Shape result like audit context helpers expect + soft_result = { + **adv_result, + "adversary_summary": summary, + "findings": adv_result.get("findings") or [], + } + _print_engagement( + emit_after_adversary_soft( + soft_result, + no_engage=False, + ) + ) return 0 if args.command == "evidence": diff --git a/docs/contributors/README.md b/docs/contributors/README.md new file mode 100644 index 0000000..9ac9dcf --- /dev/null +++ b/docs/contributors/README.md @@ -0,0 +1,76 @@ +# Contributor Engagement & Learning + +Transparent, respectful, **opt-in**, privacy-preserving contribution prompts and local packaging. + +AXGuard may recognize useful local work (verified findings, false-positive rejections, regressions, attack paths) and optionally invite a contribution. Scoring measures whether a **contribution opportunity** exists — never personal worth, streaks, or rankings. + +## Principles + +1. **Opt-in** — packaging and learning stay off until you say so. +2. **Local by default** — no network or telemetry for learning. +3. **Prepare only** — never silent push or PR. GitHub OAuth / remote submit is future work. +4. **No dark patterns** — no fake praise, guilt, fake urgency, or fake leaderboards. +5. **Reuse** — scrubbing uses `engines.data.scrub`; messaging reuses the engagement engine. + +## Journey + +```text +observable work → opportunity signal → quality gate → recognition / invite + ↓ (explicit) + axguard contribute prepare + ↓ + .findings/axguard/contribute// +``` + +Dismissal options: + +- `not_now` — cooldown; no more invites this session +- `never_prompts` — stop contribution prompts +- `type` — never suggest that contribution type again + +## Privacy + +Prefs: `~/.axguard/privacy.json` + +```bash +axguard privacy status +axguard privacy show # includes included/excluded field lists +axguard privacy opt-in +axguard privacy opt-in --learning # local learning copy only; still no upload +axguard privacy opt-out +axguard privacy export -o .findings/axguard/privacy-export.json +axguard privacy delete # remove local learning data +axguard privacy reset # prefs → defaults + clear learning data +``` + +Defaults: + +```yaml +contributions: + enabled: true + prompts: true + auto_prepare: false + auto_push: false + auto_pr: false + learning: false # opt-in +learning: + contribution_data: + enabled: false +``` + +## CLI + +```bash +axguard contribute status +axguard contribute suggest [--context-json FILE] [--force] +axguard contribute prepare [--type TYPE] [--context-json FILE] [--learning] +axguard contribute dismiss not_now|never_prompts|type [--type TYPE] +axguard contribute milestones +axguard contribute templates +``` + +## Future (not in this MVP) + +- GitHub OAuth App / authenticated PR creation +- Remote learning submit +- Hosted contribution dashboards diff --git a/docs/engagement.md b/docs/engagement.md index 2053476..3fd3eb2 100644 --- a/docs/engagement.md +++ b/docs/engagement.md @@ -35,3 +35,7 @@ axguard paths . --no-engage 2. **No dark patterns** — no urgency, fake social proof, or invented YC status. 3. **Local only** — state file is optional; delete it to reset. 4. **One message** — CLI prints at most one engagement block after the real report. + +## Contributors + +Optional contribution recognition and local packaging live in a separate engine. See [docs/contributors/README.md](contributors/README.md) for privacy defaults, `axguard privacy` / `axguard contribute` CLI, and the prepare-local-only rule (no auto push/PR). diff --git a/engines/contributors/__init__.py b/engines/contributors/__init__.py new file mode 100644 index 0000000..1f494a6 --- /dev/null +++ b/engines/contributors/__init__.py @@ -0,0 +1,89 @@ +"""AXGuard Contributor Engagement & Learning System. + +Transparent, respectful, opt-in, privacy-preserving. +Local-only by default. Never silently push or open PRs. +""" + +from __future__ import annotations + +from engines.contributors.cli import ( + add_contribute_parser, + add_privacy_parser, + run_contribute_command, + run_privacy_command, +) +from engines.contributors.extract import extract_pattern +from engines.contributors.hooks import ( + emit_after_adversary_soft, + emit_after_audit_soft, + emit_after_memory_soft, +) +from engines.contributors.milestones import list_milestones, record_prepare_milestone +from engines.contributors.prepare import prepare +from engines.contributors.privacy import ( + delete_learning_data, + export_privacy_bundle, + is_contribute_opted_in, + is_learning_opted_in, + opt_in, + opt_out, + reset, + status as privacy_status, +) +from engines.contributors.quality import assess_quality, filter_strong +from engines.contributors.recognize import ( + CONTRIBUTE_MESSAGE_CATALOG, + build_recognition_message, + recognize, +) +from engines.contributors.sanitize import sanitize_example, sanitize_text +from engines.contributors.schema import ( + CONTRIBUTION_TYPES, + DEFAULT_CONTRIBUTIONS_CONFIG, + DIFFICULTIES, + PROVENANCE_STATES, +) +from engines.contributors.signals import OpportunitySignal, best_opportunity, signals_from_context +from engines.contributors.suggest import ContributionInvite, dismiss, suggest +from engines.contributors.templates import get_template, list_templates + +__all__ = [ + "add_contribute_parser", + "add_privacy_parser", + "run_contribute_command", + "run_privacy_command", + "privacy_status", + "opt_in", + "opt_out", + "export_privacy_bundle", + "delete_learning_data", + "reset", + "is_contribute_opted_in", + "is_learning_opted_in", + "OpportunitySignal", + "signals_from_context", + "best_opportunity", + "assess_quality", + "filter_strong", + "build_recognition_message", + "recognize", + "CONTRIBUTE_MESSAGE_CATALOG", + "ContributionInvite", + "suggest", + "dismiss", + "sanitize_text", + "sanitize_example", + "extract_pattern", + "prepare", + "list_templates", + "get_template", + "list_milestones", + "record_prepare_milestone", + "emit_after_audit_soft", + "emit_after_adversary_soft", + "emit_after_memory_soft", + "CONTRIBUTION_TYPES", + "DIFFICULTIES", + "PROVENANCE_STATES", + "DEFAULT_CONTRIBUTIONS_CONFIG", +] diff --git a/engines/contributors/cli.py b/engines/contributors/cli.py new file mode 100644 index 0000000..7003357 --- /dev/null +++ b/engines/contributors/cli.py @@ -0,0 +1,220 @@ +"""CLI for ``axguard contribute …`` and ``axguard privacy …``.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + + +def add_privacy_parser(sub: argparse._SubParsersAction) -> None: + privacy = sub.add_parser( + "privacy", + help="Local contribution / learning privacy prefs (no network)", + ) + privacy_sub = privacy.add_subparsers(dest="privacy_command", required=True) + + privacy_sub.add_parser("status", help="Show privacy / opt-in status") + privacy_sub.add_parser("show", help="Show status + included/excluded fields") + + opt_in = privacy_sub.add_parser("opt-in", help="Opt in to local contribution packaging") + opt_in.add_argument( + "--learning", + action="store_true", + help="Also enable local learning contribution_data (still no network)", + ) + + privacy_sub.add_parser("opt-out", help="Opt out of contribution packaging / learning") + + export = privacy_sub.add_parser("export", help="Export local privacy prefs JSON") + export.add_argument( + "-o", + "--output", + default=".findings/axguard/privacy-export.json", + help="Output path (local only)", + ) + + privacy_sub.add_parser("delete", help="Delete local learning data") + privacy_sub.add_parser("reset", help="Reset privacy prefs and clear learning data") + + +def add_contribute_parser(sub: argparse._SubParsersAction) -> None: + contrib = sub.add_parser( + "contribute", + help="Local contribution suggestions and prepare (never auto-push)", + ) + contrib_sub = contrib.add_subparsers(dest="contribute_command", required=True) + + contrib_sub.add_parser("status", help="Show contribute prefs and last invite state") + suggest_p = contrib_sub.add_parser("suggest", help="Suggest one contribution from context") + suggest_p.add_argument( + "--context-json", + default=None, + help="Optional JSON file with local analysis context", + ) + suggest_p.add_argument( + "--force", + action="store_true", + help="Bypass cooldown (still respects never_prompts / type dismiss)", + ) + + prepare_p = contrib_sub.add_parser( + "prepare", + help="Prepare a local contribution package (no push / no PR)", + ) + prepare_p.add_argument( + "--type", + dest="contribution_type", + default=None, + help="Contribution type (e.g. REGRESSION_TEST, FALSE_POSITIVE_FIX)", + ) + prepare_p.add_argument( + "--context-json", + default=None, + help="Optional JSON file with local analysis context", + ) + prepare_p.add_argument( + "--learning", + action="store_true", + help="Also write a learning copy (requires learning opt-in)", + ) + prepare_p.add_argument( + "--out-root", + default=None, + help="Project root for .findings/axguard/contribute/ (default: cwd)", + ) + + dismiss_p = contrib_sub.add_parser( + "dismiss", + help="Dismiss invites: not_now | never_prompts | type", + ) + dismiss_p.add_argument( + "mode", + choices=("not_now", "never_prompts", "type"), + help="Dismissal mode", + ) + dismiss_p.add_argument( + "--type", + dest="contribution_type", + default=None, + help="Required when mode=type", + ) + + contrib_sub.add_parser("milestones", help="List local contribution milestones") + contrib_sub.add_parser("templates", help="List contribution templates") + + +def _load_context(path: str | None) -> dict[str, Any]: + if not path: + return {} + data = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError("context JSON must be an object") + return data + + +def _print_json(payload: Any) -> None: + print(json.dumps(payload, indent=2, sort_keys=True, default=str)) + + +def run_privacy_command(args: argparse.Namespace) -> int: + from engines.contributors import privacy as priv + + cmd = getattr(args, "privacy_command", None) + try: + if cmd in ("status", "show"): + _print_json(priv.status() if cmd == "status" else priv.show()) + return 0 + if cmd == "opt-in": + _print_json(priv.opt_in(contribute=True, learning=bool(getattr(args, "learning", False)))) + print("Opted in locally. Nothing was uploaded.", file=sys.stderr) + return 0 + if cmd == "opt-out": + _print_json(priv.opt_out(clear_learning=False)) + print("Opted out locally.", file=sys.stderr) + return 0 + if cmd == "export": + dest = Path(getattr(args, "output") or ".findings/axguard/privacy-export.json") + path = priv.export_privacy_bundle(dest) + print(str(path)) + return 0 + if cmd == "delete": + removed = priv.delete_learning_data() + _print_json({"deleted": removed, "learning_data_dir": str(priv.learning_data_dir())}) + return 0 + if cmd == "reset": + _print_json(priv.reset()) + print("Privacy prefs reset; learning data cleared.", file=sys.stderr) + return 0 + except Exception as exc: # noqa: BLE001 + print(f"error: privacy {cmd} failed: {exc}", file=sys.stderr) + return 2 + print(f"error: unknown privacy command: {cmd}", file=sys.stderr) + return 2 + + +def run_contribute_command(args: argparse.Namespace) -> int: + from engines.contributors import milestones as mile + from engines.contributors import privacy as priv + from engines.contributors.prepare import prepare + from engines.contributors.suggest import dismiss, suggest + from engines.contributors.templates import list_templates + + cmd = getattr(args, "contribute_command", None) + try: + if cmd == "status": + st = mile.load_state() + payload = { + "privacy": priv.status(), + "last_invite_at": st.get("last_invite_at"), + "last_invite_type": st.get("last_invite_type"), + "session_invite_count": st.get("session_invite_count"), + "milestones": mile.list_milestones(), + "note": "Prepare is local-only. auto_push and auto_pr stay false.", + } + _print_json(payload) + return 0 + + if cmd == "suggest": + ctx = _load_context(getattr(args, "context_json", None)) + invite = suggest(ctx, force=bool(getattr(args, "force", False))) + if invite is None: + print("No strong contribution opportunity right now.") + return 0 + print(invite.message) + return 0 + + if cmd == "prepare": + ctx = _load_context(getattr(args, "context_json", None)) + root = Path(args.out_root) if getattr(args, "out_root", None) else None + result = prepare( + ctx, + contribution_type=getattr(args, "contribution_type", None), + project_root=root, + write_learning_copy=bool(getattr(args, "learning", False)), + ) + _print_json(result) + return 0 if result.get("ok") else 2 + + if cmd == "dismiss": + result = dismiss( + getattr(args, "mode"), + contribution_type=getattr(args, "contribution_type", None), + ) + _print_json(result) + return 0 if result.get("ok") else 2 + + if cmd == "milestones": + _print_json(mile.list_milestones()) + return 0 + + if cmd == "templates": + _print_json(list_templates()) + return 0 + except Exception as exc: # noqa: BLE001 + print(f"error: contribute {cmd} failed: {exc}", file=sys.stderr) + return 2 + print(f"error: unknown contribute command: {cmd}", file=sys.stderr) + return 2 diff --git a/engines/contributors/extract.py b/engines/contributors/extract.py new file mode 100644 index 0000000..a851cd4 --- /dev/null +++ b/engines/contributors/extract.py @@ -0,0 +1,114 @@ +"""Extract technical patterns without identity.""" + +from __future__ import annotations + +import re +from typing import Any + +from engines.contributors.sanitize import sanitize_text + +_LANG_HINTS: list[tuple[re.Pattern[str], str]] = [ + (re.compile(r"\.(py)$", re.I), "python"), + (re.compile(r"\.(ts|tsx)$", re.I), "typescript"), + (re.compile(r"\.(js|jsx|mjs|cjs)$", re.I), "javascript"), + (re.compile(r"\.(go)$", re.I), "go"), + (re.compile(r"\.(rs)$", re.I), "rust"), + (re.compile(r"\.(java)$", re.I), "java"), + (re.compile(r"\.(rb)$", re.I), "ruby"), + (re.compile(r"\.(php)$", re.I), "php"), +] + +_FRAMEWORK_HINTS: list[tuple[re.Pattern[str], str]] = [ + (re.compile(r"\b(django|flask|fastapi|starlette)\b", re.I), "python-web"), + (re.compile(r"\b(express|next\.?js|nestjs|react)\b", re.I), "node-web"), + (re.compile(r"\b(rails|sinatra)\b", re.I), "ruby-web"), + (re.compile(r"\b(spring|ktor)\b", re.I), "jvm-web"), + (re.compile(r"\b(mcp|model.?context.?protocol)\b", re.I), "mcp"), + (re.compile(r"\b(langchain|llamaindex|openai|anthropic)\b", re.I), "ai-agent"), +] + +_PATTERN_HINTS: list[tuple[re.Pattern[str], str]] = [ + (re.compile(r"\bssrf\b", re.I), "ssrf"), + (re.compile(r"\bxss\b", re.I), "xss"), + (re.compile(r"\bsqli?\b|\bsql.?injection\b", re.I), "sqli"), + (re.compile(r"\bpath.?traversal\b|\blfi\b", re.I), "path-traversal"), + (re.compile(r"\bidor\b|\bbroken.?access\b", re.I), "authz-idor"), + (re.compile(r"\bfalse.?positive\b", re.I), "false-positive"), + (re.compile(r"\bprompt.?injection\b", re.I), "prompt-injection"), + (re.compile(r"\bcommand.?injection\b|\brce\b", re.I), "command-injection"), +] + +_CONTROL_HINTS: list[tuple[re.Pattern[str], str]] = [ + (re.compile(r"\bmissing\s+(authz|authorization|authn|authentication)\b", re.I), "auth"), + (re.compile(r"\bno\s+allowlist\b|\bmissing\s+allowlist\b", re.I), "allowlist"), + (re.compile(r"\bunescaped\b|\bmissing\s+encoding\b", re.I), "output-encoding"), + (re.compile(r"\bmissing\s+csrf\b", re.I), "csrf"), + (re.compile(r"\bmissing\s+rate.?limit\b", re.I), "rate-limit"), +] + + +def _detect_language(file_path: str | None, blob: str) -> str | None: + if file_path: + for pat, lang in _LANG_HINTS: + if pat.search(file_path): + return lang + lowered = blob.lower() + if "def " in lowered and "import " in lowered: + return "python" + if "function " in lowered or "const " in lowered: + return "javascript" + return None + + +def _first_match(patterns: list[tuple[re.Pattern[str], str]], text: str) -> str | None: + for pat, label in patterns: + if pat.search(text): + return label + return None + + +def extract_pattern(context: dict[str, Any] | None = None) -> dict[str, Any]: + """Extract language / framework / pattern / flow / missing control — no identity.""" + ctx = dict(context or {}) + file_path = str(ctx.get("file") or ctx.get("path") or "") + pieces = [ + str(ctx.get("title") or ""), + str(ctx.get("rule_id") or ""), + str(ctx.get("cwe") or ""), + str(ctx.get("finding_class") or ""), + str(ctx.get("message") or ""), + str(ctx.get("description") or ""), + str(ctx.get("snippet") or ctx.get("code") or ""), + str(ctx.get("flow_summary") or ""), + " ".join(str(x) for x in (ctx.get("tags") or [])), + ] + blob = "\n".join(p for p in pieces if p) + + language = ctx.get("language") or _detect_language(file_path, blob) + framework = ctx.get("framework") or _first_match(_FRAMEWORK_HINTS, blob) + pattern = ctx.get("pattern") or _first_match(_PATTERN_HINTS, blob) + missing_control = ctx.get("missing_control") or _first_match(_CONTROL_HINTS, blob) + + flow = ctx.get("flow_summary") + if not flow and ctx.get("source") and ctx.get("sink"): + flow = f"{ctx.get('source')} → {ctx.get('sink')}" + if isinstance(flow, str): + flow, _ = sanitize_text(flow) + + snippet = ctx.get("snippet") or ctx.get("code") + sanitized_snippet = None + if isinstance(snippet, str) and snippet.strip(): + sanitized_snippet, _ = sanitize_text(snippet) + + result = { + "language": language, + "framework": framework, + "pattern": pattern, + "flow_summary": flow, + "missing_control": missing_control, + "sanitized_snippet": sanitized_snippet, + "rule_id": ctx.get("rule_id"), + "cwe": ctx.get("cwe"), + } + # Drop empties for a clean technical fingerprint + return {k: v for k, v in result.items() if v not in (None, "", [])} diff --git a/engines/contributors/hooks.py b/engines/contributors/hooks.py new file mode 100644 index 0000000..deb6055 --- /dev/null +++ b/engines/contributors/hooks.py @@ -0,0 +1,123 @@ +"""Soft hooks after audit / adversary / memory signals. + +Respects contributions.prompts, invite cooldown, engage disable, and --no-engage. +Never weakens security-first promo blocking. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from engines.contributors.privacy import prompts_allowed +from engines.contributors.recognize import build_recognition_message +from engines.contributors.suggest import suggest +from engines.engagement.hooks import _context_from_result, _promo_blocked_for_cli +from engines.engagement.engine import render_cli +from engines.engagement.state import load_state + + +def _engage_disabled(state_path: Path | None = None) -> bool: + try: + state = load_state(state_path) + return bool(state.get("promo_disabled")) or state.get("state") == "DISABLED" + except Exception: # noqa: BLE001 + return False + + +def emit_after_audit_soft( + result: dict[str, Any], + *, + state_path: Path | None = None, + privacy_path: Path | None = None, + no_engage: bool = False, + session_id: str | None = None, +) -> str | None: + """Soft contribute recognition/invite after audit. Never raises.""" + try: + if no_engage: + return None + if _engage_disabled(state_path): + return None + if not prompts_allowed(privacy_path): + return None + + ctx = _context_from_result(result, command="audit") + # Prefer invite when quality is strong; else quiet recognition + invite = suggest( + ctx, + privacy_path=privacy_path, + session_id=session_id or "audit", + ) + if invite is not None: + return invite.message + + msg = build_recognition_message(ctx, with_cta=True) + if msg is None: + return None + if _promo_blocked_for_cli(msg, ctx): + # Still allow non-SUPPORT recognition; only block if policy says so + # Contribute CTA is not a star ask — keep product-info recognition. + if msg.priority in {"GITHUB", "YC", "AWAREXONE"}: + return None + text = render_cli(msg).strip() + return text or None + except Exception: # noqa: BLE001 + return None + + +def emit_after_adversary_soft( + result: dict[str, Any], + *, + state_path: Path | None = None, + privacy_path: Path | None = None, + no_engage: bool = False, + session_id: str | None = None, +) -> str | None: + """Soft hook when adversary rejects FPs.""" + try: + if no_engage or _engage_disabled(state_path) or not prompts_allowed(privacy_path): + return None + ctx = _context_from_result(result, command="adversary") + if int(ctx.get("fp_rejected") or 0) <= 0: + return None + invite = suggest(ctx, privacy_path=privacy_path, session_id=session_id or "adversary") + if invite is not None: + return invite.message + msg = build_recognition_message(ctx) + return render_cli(msg).strip() or None if msg else None + except Exception: # noqa: BLE001 + return None + + +def emit_after_memory_soft( + result: dict[str, Any], + *, + state_path: Path | None = None, + privacy_path: Path | None = None, + no_engage: bool = False, + session_id: str | None = None, +) -> str | None: + """Soft hook for memory regressions / longitudinal signals.""" + try: + if no_engage or _engage_disabled(state_path) or not prompts_allowed(privacy_path): + return None + ctx = dict(result or {}) + regs = result.get("regressions") or result.get("regression_summary") or {} + if isinstance(regs, dict): + count = int(regs.get("count") or regs.get("regression_count") or len(regs.get("items") or [])) + elif isinstance(regs, list): + count = len(regs) + else: + count = int(result.get("regression_count") or 0) + if count <= 0 and not result.get("regression"): + return None + ctx["regression"] = True + ctx["regression_count"] = count + invite = suggest(ctx, privacy_path=privacy_path, session_id=session_id or "memory") + if invite is not None: + return invite.message + msg = build_recognition_message(ctx) + return render_cli(msg).strip() or None if msg else None + except Exception: # noqa: BLE001 + return None diff --git a/engines/contributors/milestones.py b/engines/contributors/milestones.py new file mode 100644 index 0000000..0549bcb --- /dev/null +++ b/engines/contributors/milestones.py @@ -0,0 +1,139 @@ +"""Local contribution milestones — no streaks, pressure, or rankings.""" + +from __future__ import annotations + +import json +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from engines.contributors.schema import CONTRIBUTE_STATE_FILENAME + +DEFAULT_STATE_PATH = Path.home() / ".axguard" / CONTRIBUTE_STATE_FILENAME + +# Quiet, factual milestones only +MILESTONE_FIRST_REGRESSION = "first_regression_contribution" +MILESTONE_FIRST_RULE = "first_rule_contribution" +MILESTONE_FIRST_FP_FIX = "first_fp_fix_contribution" +MILESTONE_FIRST_DOCS = "first_docs_contribution" +MILESTONE_FIRST_ATTACK_PATH = "first_attack_path_contribution" +MILESTONE_FIRST_PREPARE = "first_local_prepare" + +MILESTONE_LABELS: dict[str, str] = { + MILESTONE_FIRST_REGRESSION: "First regression contribution prepared locally", + MILESTONE_FIRST_RULE: "First rule contribution prepared locally", + MILESTONE_FIRST_FP_FIX: "First false-positive fix prepared locally", + MILESTONE_FIRST_DOCS: "First documentation contribution prepared locally", + MILESTONE_FIRST_ATTACK_PATH: "First attack-path pattern prepared locally", + MILESTONE_FIRST_PREPARE: "First local contribution package prepared", +} + +_TYPE_TO_MILESTONE: dict[str, str] = { + "REGRESSION_TEST": MILESTONE_FIRST_REGRESSION, + "TEST": MILESTONE_FIRST_REGRESSION, + "SECURITY_RULE": MILESTONE_FIRST_RULE, + "RULE": MILESTONE_FIRST_RULE, + "FALSE_POSITIVE_FIX": MILESTONE_FIRST_FP_FIX, + "DOCUMENTATION": MILESTONE_FIRST_DOCS, + "DOCS": MILESTONE_FIRST_DOCS, + "ATTACK_PATH_PATTERN": MILESTONE_FIRST_ATTACK_PATH, +} + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def resolve_state_path(path: Path | None = None) -> Path: + if path is not None: + return Path(path) + return DEFAULT_STATE_PATH + + +def default_state() -> dict[str, Any]: + now = _utc_now_iso() + return { + "milestones": [], + "last_invite_at": None, + "last_invite_type": None, + "session_invite_count": 0, + "session_id": None, + "dismissed_invites": [], + "prepared_ids": [], + "created_at": now, + "updated_at": now, + } + + +def load_state(path: Path | None = None) -> dict[str, Any]: + dest = resolve_state_path(path) + if not dest.is_file(): + return default_state() + try: + raw = json.loads(dest.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + return default_state() + if not isinstance(raw, dict): + return default_state() + base = default_state() + merged = {**base, **raw} + for key in ("milestones", "dismissed_invites", "prepared_ids"): + if not isinstance(merged.get(key), list): + merged[key] = [] + return merged + + +def save_state(state: dict[str, Any], path: Path | None = None) -> Path: + dest = resolve_state_path(path) + dest.parent.mkdir(parents=True, exist_ok=True) + payload = deepcopy(state) + payload["updated_at"] = _utc_now_iso() + tmp = dest.with_suffix(dest.suffix + ".tmp") + tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + tmp.replace(dest) + return dest + + +def list_milestones(path: Path | None = None) -> list[dict[str, str]]: + state = load_state(path) + out: list[dict[str, str]] = [] + for mid in state.get("milestones") or []: + out.append( + { + "id": str(mid), + "label": MILESTONE_LABELS.get(str(mid), str(mid)), + } + ) + return out + + +def record_milestone(milestone_id: str, *, path: Path | None = None) -> bool: + """Record a local milestone once. Returns True if newly added.""" + if milestone_id not in MILESTONE_LABELS: + return False + state = load_state(path) + seen = list(state.get("milestones") or []) + if milestone_id in seen: + return False + seen.append(milestone_id) + state["milestones"] = seen + save_state(state, path) + return True + + +def record_prepare_milestone(contribution_type: str, *, path: Path | None = None) -> list[str]: + """Record prepare-related milestones. No streaks or rankings.""" + added: list[str] = [] + if record_milestone(MILESTONE_FIRST_PREPARE, path=path): + added.append(MILESTONE_FIRST_PREPARE) + typed = _TYPE_TO_MILESTONE.get(contribution_type) + if typed and record_milestone(typed, path=path): + added.append(typed) + return added + + +def clear_state(path: Path | None = None) -> None: + dest = resolve_state_path(path) + if dest.is_file(): + dest.unlink() diff --git a/engines/contributors/prepare.py b/engines/contributors/prepare.py new file mode 100644 index 0000000..999f499 --- /dev/null +++ b/engines/contributors/prepare.py @@ -0,0 +1,203 @@ +"""Prepare a local contribution package — never push or open a PR. + +Packages land under ``.findings/axguard/contribute//``. +Requires an explicit prepare call. Learning export requires privacy opt-in. +""" + +from __future__ import annotations + +import json +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") + +from engines.contributors.extract import extract_pattern +from engines.contributors.milestones import record_prepare_milestone +from engines.contributors.privacy import ( + is_contribute_opted_in, + is_learning_opted_in, + learning_data_dir, +) +from engines.contributors.quality import assess_quality +from engines.contributors.sanitize import sanitize_example, sanitize_text +from engines.contributors.schema import ( + CONTRIBUTE_PACKAGE_DIRNAME, + PROVENANCE_LOCAL_ONLY, + PROVENANCE_SANITIZED, + PROVENANCE_USER_APPROVED, + TYPE_LABELS, +) +from engines.contributors.signals import best_opportunity +from engines.contributors.templates import render_template_files + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def default_package_root(project_root: Path | None = None) -> Path: + root = Path(project_root) if project_root is not None else Path.cwd() + return root / ".findings" / "axguard" / CONTRIBUTE_PACKAGE_DIRNAME + + +def prepare( + context: dict[str, Any] | None = None, + *, + contribution_type: str | None = None, + project_root: Path | None = None, + package_id: str | None = None, + privacy_path: Path | None = None, + contribute_state_path: Path | None = None, + require_opt_in: bool = True, + write_learning_copy: bool = False, +) -> dict[str, Any]: + """Prepare a local contribution package. Never opens sockets or pushes. + + ``require_opt_in``: when True, contribute opt-in is required. + Learning export copy also requires learning opt-in. + """ + ctx = dict(context or {}) + if require_opt_in and not is_contribute_opted_in(privacy_path): + return { + "ok": False, + "error": "contribute_opt_in_required", + "hint": "Run: axguard privacy opt-in", + } + + signal = best_opportunity(ctx) + ctype = contribution_type or (signal.contribution_type if signal else None) + if not ctype: + ctype = "DOCUMENTATION" + difficulty = signal.difficulty if signal and signal.contribution_type == ctype else "MEDIUM" + reason = signal.reason if signal else "Manual local contribution package." + + qa = None + if signal is not None: + qa = assess_quality(signal, ctx) + + pattern = extract_pattern(ctx) + known_private = [] + for key in ("repo", "repo_name", "org", "project_name"): + val = ctx.get(key) + if isinstance(val, str) and val.strip(): + known_private.append(val.strip()) + + example = { + "contribution_type": ctype, + "difficulty": difficulty, + "reason": reason, + "pattern": pattern, + "code": ctx.get("snippet") or ctx.get("code") or "", + "title": ctx.get("title") or TYPE_LABELS.get(ctype, ctype), + "repo": ctx.get("repo") or ctx.get("repo_name") or "", + "file": ctx.get("file") or ctx.get("path") or "", + "provenance": PROVENANCE_LOCAL_ONLY, + "created_at": _utc_now_iso(), + } + sanitized, hits = sanitize_example(example, known_private_names=known_private) + sanitized["provenance"] = PROVENANCE_SANITIZED + if is_contribute_opted_in(privacy_path): + sanitized["provenance"] = PROVENANCE_USER_APPROVED + # Keep sanitized marker alongside approval for transparency + sanitized["sanitized"] = True + sanitized["scrub_hits"] = list(dict.fromkeys(hits))[:20] + + if qa is not None: + sanitized["quality_scores"] = qa.as_dict() + + pkg_id = package_id or f"c_{uuid4().hex[:12]}" + if not _SAFE_ID.match(pkg_id) or ".." in pkg_id or "/" in pkg_id or "\\" in pkg_id: + return { + "ok": False, + "error": "invalid_package_id", + "hint": "package_id must be a short alphanumeric id (no path separators).", + } + root = default_package_root(project_root).resolve() + out_dir = (root / pkg_id).resolve() + if not out_dir.is_relative_to(root): + return { + "ok": False, + "error": "invalid_package_id", + "hint": "package_id must stay under the contribute package directory.", + } + out_dir.mkdir(parents=True, exist_ok=True) + + # Template stubs + files = render_template_files( + ctype, + substitutions={ + "pattern": str(pattern.get("pattern") or "pattern"), + "case": str(pattern.get("pattern") or ctype.lower()), + "name": str(sanitized.get("title") or ctype), + "topic": str(sanitized.get("title") or ctype), + "framework": str(pattern.get("framework") or "framework"), + "behavior": reason[:80], + "path": "path", + "workflow": "workflow", + }, + ) + written: list[str] = [] + for name, body in files.items(): + # Flat filenames only — reject path traversal in template names + if not name or name != Path(name).name or ".." in name: + continue + clean_body, _ = sanitize_text(body, known_private_names=known_private) + target = (out_dir / name).resolve() + if not target.is_relative_to(out_dir): + continue + target.write_text(clean_body, encoding="utf-8") + written.append(str(target)) + + meta_path = out_dir / "contribution.json" + meta_path.write_text( + json.dumps(sanitized, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + written.append(str(meta_path)) + + note_path = out_dir / "LOCAL_ONLY.txt" + note_path.write_text( + "This package was prepared locally by AXGuard.\n" + "Nothing was pushed to GitHub.\n" + "Nothing was opened as a pull request.\n" + "Review, edit, and share only if you choose to.\n", + encoding="utf-8", + ) + written.append(str(note_path)) + + learning_copy = None + if write_learning_copy or ctx.get("export_learning"): + if not is_learning_opted_in(privacy_path): + return { + "ok": False, + "error": "learning_opt_in_required", + "hint": "Run: axguard privacy opt-in --learning", + "package_dir": str(out_dir), + "files": written, + } + learn_dir = learning_data_dir(privacy_path) + learn_dir.mkdir(parents=True, exist_ok=True) + learning_copy = learn_dir / f"{pkg_id}.json" + learning_copy.write_text( + json.dumps(sanitized, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + written.append(str(learning_copy)) + + record_prepare_milestone(ctype, path=contribute_state_path) + + return { + "ok": True, + "id": pkg_id, + "package_dir": str(out_dir), + "contribution_type": ctype, + "difficulty": difficulty, + "files": written, + "learning_copy": str(learning_copy) if learning_copy else None, + "network": False, + "pushed": False, + "pr_opened": False, + "provenance": sanitized.get("provenance"), + } diff --git a/engines/contributors/privacy.py b/engines/contributors/privacy.py new file mode 100644 index 0000000..2907e33 --- /dev/null +++ b/engines/contributors/privacy.py @@ -0,0 +1,262 @@ +"""Local privacy prefs for contribution learning — no network, opt-in only. + +State lives at ``~/.axguard/privacy.json``. Learning data defaults off. +""" + +from __future__ import annotations + +import json +import shutil +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from engines.contributors.schema import ( + DEFAULT_PRIVACY_CONFIG, + EXPORT_EXCLUDED_FIELDS, + EXPORT_INCLUDED_FIELDS, + PRIVACY_FILENAME, +) + +DEFAULT_PRIVACY_PATH = Path.home() / ".axguard" / PRIVACY_FILENAME +LEARNING_DATA_DIRNAME = "contribute-learning" + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def resolve_privacy_path(path: Path | None = None) -> Path: + if path is not None: + return Path(path) + return DEFAULT_PRIVACY_PATH + + +def default_privacy() -> dict[str, Any]: + base = deepcopy(DEFAULT_PRIVACY_CONFIG) + now = _utc_now_iso() + base["created_at"] = now + base["updated_at"] = now + return base + + +def load_privacy(path: Path | None = None) -> dict[str, Any]: + dest = resolve_privacy_path(path) + if not dest.is_file(): + return default_privacy() + try: + raw = json.loads(dest.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + return default_privacy() + if not isinstance(raw, dict): + return default_privacy() + base = default_privacy() + merged = {**base, **raw} + contributions = dict(base.get("contributions") or {}) + if isinstance(raw.get("contributions"), dict): + contributions.update(raw["contributions"]) + merged["contributions"] = contributions + learning = deepcopy(base.get("learning") or {}) + if isinstance(raw.get("learning"), dict): + learning.update(raw["learning"]) + cd = learning.get("contribution_data") + if isinstance(cd, dict): + base_cd = dict((base.get("learning") or {}).get("contribution_data") or {}) + base_cd.update(cd) + learning["contribution_data"] = base_cd + merged["learning"] = learning + if not isinstance(merged.get("dismissed_types"), list): + merged["dismissed_types"] = [] + return merged + + +def save_privacy(state: dict[str, Any], path: Path | None = None) -> Path: + dest = resolve_privacy_path(path) + dest.parent.mkdir(parents=True, exist_ok=True) + payload = deepcopy(state) + payload["updated_at"] = _utc_now_iso() + tmp = dest.with_suffix(dest.suffix + ".tmp") + tmp.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + tmp.replace(dest) + return dest + + +def learning_data_dir(privacy_path: Path | None = None) -> Path: + """Directory for local learning packs (never uploaded by this module).""" + root = resolve_privacy_path(privacy_path).parent + return root / LEARNING_DATA_DIRNAME + + +def field_explanations() -> dict[str, Any]: + """Clear explanation of what local export may include / never includes.""" + return { + "included": list(EXPORT_INCLUDED_FIELDS), + "excluded": list(EXPORT_EXCLUDED_FIELDS), + "notes": [ + "Learning and contribution packaging are opt-in.", + "Nothing is sent over the network by AXGuard learning defaults.", + "Secrets and private repo identity are scrubbed before any pack is written.", + "GitHub push/PR is never automatic — prepare locally only.", + ], + } + + +def status(path: Path | None = None) -> dict[str, Any]: + state = load_privacy(path) + contrib = state.get("contributions") or {} + learning = state.get("learning") or {} + cd = learning.get("contribution_data") or {} + return { + "privacy_path": str(resolve_privacy_path(path)), + "contribute_opt_in": bool(state.get("contribute_opt_in")), + "learning_opt_in": bool(state.get("learning_opt_in")), + "never_prompts": bool(state.get("never_prompts")), + "dismissed_types": list(state.get("dismissed_types") or []), + "contributions": { + "enabled": bool(contrib.get("enabled", True)), + "prompts": bool(contrib.get("prompts", True)), + "auto_prepare": bool(contrib.get("auto_prepare", False)), + "auto_push": bool(contrib.get("auto_push", False)), + "auto_pr": bool(contrib.get("auto_pr", False)), + "learning": bool(contrib.get("learning", False)), + }, + "learning": { + "contribution_data": { + "enabled": bool(cd.get("enabled", False)), + } + }, + "fields": field_explanations(), + "learning_data_dir": str(learning_data_dir(path)), + "learning_data_present": learning_data_dir(path).is_dir(), + } + + +def show(path: Path | None = None) -> dict[str, Any]: + """Alias for status with explicit field documentation.""" + return status(path) + + +def opt_in( + *, + contribute: bool = True, + learning: bool = False, + path: Path | None = None, +) -> dict[str, Any]: + """Explicit opt-in. Learning stays off unless ``learning=True``.""" + state = load_privacy(path) + state["contribute_opt_in"] = bool(contribute) + if learning: + state["learning_opt_in"] = True + contrib = dict(state.get("contributions") or {}) + contrib["learning"] = True + state["contributions"] = contrib + learning_cfg = deepcopy(state.get("learning") or {}) + cd = dict(learning_cfg.get("contribution_data") or {}) + cd["enabled"] = True + learning_cfg["contribution_data"] = cd + state["learning"] = learning_cfg + save_privacy(state, path) + return status(path) + + +def opt_out(*, path: Path | None = None, clear_learning: bool = False) -> dict[str, Any]: + state = load_privacy(path) + state["contribute_opt_in"] = False + state["learning_opt_in"] = False + contrib = dict(state.get("contributions") or {}) + contrib["learning"] = False + state["contributions"] = contrib + learning_cfg = deepcopy(state.get("learning") or {}) + cd = dict(learning_cfg.get("contribution_data") or {}) + cd["enabled"] = False + learning_cfg["contribution_data"] = cd + state["learning"] = learning_cfg + save_privacy(state, path) + if clear_learning: + delete_learning_data(path) + return status(path) + + +def export_privacy_bundle(dest: Path, *, path: Path | None = None) -> Path: + """Write a local JSON export of prefs + field explanations (no secrets).""" + dest = Path(dest) + dest.parent.mkdir(parents=True, exist_ok=True) + payload = { + "exported_at": _utc_now_iso(), + "status": status(path), + "note": "Local export only. AXGuard does not upload this file.", + } + dest.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return dest + + +def delete_learning_data(path: Path | None = None) -> bool: + """Delete local learning data directory. Returns True if something was removed.""" + data_dir = learning_data_dir(path) + if data_dir.is_dir(): + shutil.rmtree(data_dir) + return True + return False + + +def reset(*, path: Path | None = None) -> dict[str, Any]: + """Reset privacy prefs to defaults and clear local learning data.""" + delete_learning_data(path) + dest = resolve_privacy_path(path) + if dest.is_file(): + dest.unlink() + save_privacy(default_privacy(), path) + return status(path) + + +def is_contribute_opted_in(path: Path | None = None) -> bool: + return bool(load_privacy(path).get("contribute_opt_in")) + + +def is_learning_opted_in(path: Path | None = None) -> bool: + state = load_privacy(path) + if not state.get("learning_opt_in"): + return False + contrib = state.get("contributions") or {} + if not contrib.get("learning"): + return False + learning = state.get("learning") or {} + cd = learning.get("contribution_data") or {} + return bool(cd.get("enabled")) + + +def prompts_allowed(path: Path | None = None) -> bool: + state = load_privacy(path) + if state.get("never_prompts"): + return False + contrib = state.get("contributions") or {} + if not contrib.get("enabled", True): + return False + return bool(contrib.get("prompts", True)) + + +def set_never_prompts(value: bool = True, *, path: Path | None = None) -> dict[str, Any]: + state = load_privacy(path) + state["never_prompts"] = bool(value) + if value: + contrib = dict(state.get("contributions") or {}) + contrib["prompts"] = False + state["contributions"] = contrib + save_privacy(state, path) + return status(path) + + +def dismiss_type(contribution_type: str, *, path: Path | None = None) -> dict[str, Any]: + state = load_privacy(path) + dismissed = list(state.get("dismissed_types") or []) + if contribution_type and contribution_type not in dismissed: + dismissed.append(contribution_type) + state["dismissed_types"] = dismissed + save_privacy(state, path) + return status(path) + + +def is_type_dismissed(contribution_type: str, *, path: Path | None = None) -> bool: + dismissed = load_privacy(path).get("dismissed_types") or [] + return contribution_type in dismissed diff --git a/engines/contributors/quality.py b/engines/contributors/quality.py new file mode 100644 index 0000000..a2437a6 --- /dev/null +++ b/engines/contributors/quality.py @@ -0,0 +1,131 @@ +"""Quality gate for contribution opportunities. + +Only surface strong opportunities: novelty, reusability, relevance, testability. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from engines.contributors.schema import QUALITY_MIN_SCORE +from engines.contributors.signals import OpportunitySignal + + +@dataclass(frozen=True) +class QualityAssessment: + passed: bool + score: float + novelty: float + reusability: float + relevance: float + testability: float + reasons: tuple[str, ...] + + def as_dict(self) -> dict[str, Any]: + return { + "passed": self.passed, + "score": round(self.score, 3), + "novelty": round(self.novelty, 3), + "reusability": round(self.reusability, 3), + "relevance": round(self.relevance, 3), + "testability": round(self.testability, 3), + "reasons": list(self.reasons), + } + + +def _clamp(value: float) -> float: + return max(0.0, min(1.0, float(value))) + + +def assess_quality( + signal: OpportunitySignal, + context: dict[str, Any] | None = None, + *, + min_score: float = QUALITY_MIN_SCORE, +) -> QualityAssessment: + """Score a single opportunity. Does not rank people.""" + ctx = context or {} + evidence = signal.evidence or {} + + novelty = 0.35 + if evidence.get("novel") or ctx.get("novel_finding") or ctx.get("first_verified"): + novelty = 0.85 + elif evidence.get("regression") or ctx.get("regression"): + novelty = 0.7 + elif evidence.get("rule_gap") or evidence.get("mcp") or evidence.get("ai_agent"): + novelty = 0.75 + elif int(evidence.get("fp_rejected") or 0) > 0: + novelty = 0.55 + + reusability = 0.4 + ctype = signal.contribution_type + if ctype in { + "FALSE_POSITIVE_FIX", + "REGRESSION_TEST", + "SECURITY_RULE", + "ATTACK_PATH_PATTERN", + "TEST", + "RULE", + }: + reusability = 0.8 + elif ctype in {"DOCUMENTATION", "DOCS", "DX"}: + reusability = 0.65 + elif ctype in {"MCP_SECURITY_CASE", "SYNTHETIC_SECURITY_CASE", "AI_AGENT", "NOVEL_FINDING"}: + reusability = 0.7 + + relevance = _clamp(0.45 + signal.score * 0.45) + if ctx.get("security_primary") or ctx.get("verified_count") or ctx.get("fp_rejected"): + relevance = max(relevance, 0.7) + + testability = 0.4 + if ctype in {"REGRESSION_TEST", "TEST", "FALSE_POSITIVE_FIX", "BUG_FIX"}: + testability = 0.85 + elif ctype in {"SECURITY_RULE", "RULE", "ATTACK_PATH_PATTERN"}: + testability = 0.7 + elif ctype in {"DOCUMENTATION", "DOCS", "DX", "PERFORMANCE"}: + testability = 0.45 + if ctx.get("has_fixture") or ctx.get("testable"): + testability = max(testability, 0.8) + + score = _clamp(0.25 * novelty + 0.3 * reusability + 0.25 * relevance + 0.2 * testability) + # Blend with opportunity signal score lightly + score = _clamp(0.65 * score + 0.35 * signal.score) + + reasons: list[str] = [] + if novelty >= 0.7: + reasons.append("novel or uncommon pattern relative to typical noise") + if reusability >= 0.65: + reasons.append("likely reusable beyond this single repo session") + if relevance >= 0.65: + reasons.append("tied to observable security work in this session") + if testability >= 0.65: + reasons.append("can be expressed as a test, fixture, or rule case") + if score < min_score: + reasons.append("below quality threshold — not surfaced") + + return QualityAssessment( + passed=score >= min_score, + score=score, + novelty=_clamp(novelty), + reusability=_clamp(reusability), + relevance=_clamp(relevance), + testability=_clamp(testability), + reasons=tuple(reasons), + ) + + +def filter_strong( + signals: list[OpportunitySignal], + context: dict[str, Any] | None = None, + *, + min_score: float = QUALITY_MIN_SCORE, +) -> list[tuple[OpportunitySignal, QualityAssessment]]: + """Return only opportunities that clear the quality gate, best first.""" + out: list[tuple[OpportunitySignal, QualityAssessment]] = [] + for sig in signals: + qa = assess_quality(sig, context, min_score=min_score) + if qa.passed: + out.append((sig, qa)) + out.sort(key=lambda pair: pair[1].score, reverse=True) + return out diff --git a/engines/contributors/recognize.py b/engines/contributors/recognize.py new file mode 100644 index 0000000..7cd43c7 --- /dev/null +++ b/engines/contributors/recognize.py @@ -0,0 +1,190 @@ +"""Recognition messages tied to observable work — engineer-to-engineer tone. + +Calls engagement when appropriate. Never invents praise or rankings. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from engines.contributors.schema import ( + FORBIDDEN_CONTRIBUTE_PHRASES, + TYPE_LABELS, +) +from engines.contributors.signals import OpportunitySignal, best_opportunity +from engines.engagement.messages import EngagementCTA, EngagementMessage +from engines.engagement.policy import contains_banned_pattern, reject_if_manipulative +from engines.engagement.schema import ( + CTA_CONTRIBUTE, + EVENT_CONTRIBUTION_SUGGESTED, + PRIORITY_PRODUCT_INFO, + SIGNATURE, + STATE_COMMUNITY_INVITATION, + STATE_FIRST_MILESTONE, +) + + +def _assert_clean(text: str) -> None: + lowered = text.lower() + for phrase in FORBIDDEN_CONTRIBUTE_PHRASES: + if phrase in lowered: + raise ValueError(f"forbidden contribute phrase: {phrase}") + if contains_banned_pattern(text): + raise ValueError("banned engagement pattern in recognition copy") + + +def build_recognition_message( + context: dict[str, Any] | None = None, + *, + signal: OpportunitySignal | None = None, + with_cta: bool = True, +) -> EngagementMessage | None: + """Build a recognition message from observable work only.""" + ctx = context or {} + sig = signal or best_opportunity(ctx) + if sig is None: + # Still recognize FP / verified without a typed opportunity + if int(ctx.get("fp_rejected") or ctx.get("rejected_count") or 0) > 0: + body = ( + "AXGuard rejected a false positive in this session.", + "", + "That control-aware rejection is useful signal — the kind that improves rules and corpus cases.", + ) + msg = EngagementMessage( + id="recognize_fp_rejected", + stage=STATE_FIRST_MILESTONE, + event=EVENT_CONTRIBUTION_SUGGESTED, + body_lines=body, + signature=SIGNATURE, + cta=( + EngagementCTA( + kind=CTA_CONTRIBUTE, + label="Optional: axguard contribute suggest", + url=None, + ) + if with_cta + else None + ), + priority=PRIORITY_PRODUCT_INFO, + ) + elif int(ctx.get("verified_count") or 0) > 0 or ctx.get("novel_finding"): + body = ( + "A finding was verified with evidence in this session.", + "", + "If the pattern is reusable, a local contribution package can capture it — only if you want.", + ) + msg = EngagementMessage( + id="recognize_verified_finding", + stage=STATE_FIRST_MILESTONE, + event=EVENT_CONTRIBUTION_SUGGESTED, + body_lines=body, + signature=SIGNATURE, + cta=( + EngagementCTA( + kind=CTA_CONTRIBUTE, + label="Optional: axguard contribute suggest", + url=None, + ) + if with_cta + else None + ), + priority=PRIORITY_PRODUCT_INFO, + ) + else: + return None + else: + label = TYPE_LABELS.get(sig.contribution_type, sig.contribution_type) + body = ( + f"Observable work in this session points to a possible {label} contribution.", + "", + sig.reason, + "", + "This is optional. You stay in control of prepare, push, and PR.", + ) + msg = EngagementMessage( + id=f"recognize_{sig.contribution_type.lower()}", + stage=STATE_COMMUNITY_INVITATION, + event=EVENT_CONTRIBUTION_SUGGESTED, + body_lines=body, + signature=SIGNATURE, + cta=( + EngagementCTA( + kind=CTA_CONTRIBUTE, + label="Optional: axguard contribute prepare", + url=None, + ) + if with_cta + else None + ), + priority=PRIORITY_PRODUCT_INFO, + ) + + text = "\n".join(msg.body_lines) + _assert_clean(text) + if reject_if_manipulative(msg) is not None: + return None + return msg + + +def recognize( + context: dict[str, Any] | None = None, + *, + state_path: Path | None = None, + call_engagement: bool = True, +) -> EngagementMessage | None: + """Build recognition and optionally record an engagement event.""" + msg = build_recognition_message(context) + if msg is None: + return None + if call_engagement: + try: + from engines.engagement.engine import record_event + + record_event( + EVENT_CONTRIBUTION_SUGGESTED, + { + **(context or {}), + "count_meaningful": False, + "recognition_message_id": msg.id, + }, + state_path=state_path, + ) + except Exception: # noqa: BLE001 — recognition must not break callers + pass + return msg + + +# Catalog entries for anti-manipulation tests / introspection +RECOGNIZE_FP = EngagementMessage( + id="recognize_fp_rejected", + stage=STATE_FIRST_MILESTONE, + event=EVENT_CONTRIBUTION_SUGGESTED, + body_lines=( + "AXGuard rejected a false positive in this session.", + "", + "That control-aware rejection is useful signal — the kind that improves rules and corpus cases.", + ), + signature=SIGNATURE, + cta=EngagementCTA(kind=CTA_CONTRIBUTE, label="Optional: axguard contribute suggest", url=None), + priority=PRIORITY_PRODUCT_INFO, +) + +RECOGNIZE_VERIFIED = EngagementMessage( + id="recognize_verified_finding", + stage=STATE_FIRST_MILESTONE, + event=EVENT_CONTRIBUTION_SUGGESTED, + body_lines=( + "A finding was verified with evidence in this session.", + "", + "If the pattern is reusable, a local contribution package can capture it — only if you want.", + ), + signature=SIGNATURE, + cta=EngagementCTA(kind=CTA_CONTRIBUTE, label="Optional: axguard contribute suggest", url=None), + priority=PRIORITY_PRODUCT_INFO, +) + +CONTRIBUTE_MESSAGE_CATALOG: dict[str, EngagementMessage] = { + RECOGNIZE_FP.id: RECOGNIZE_FP, + RECOGNIZE_VERIFIED.id: RECOGNIZE_VERIFIED, +} diff --git a/engines/contributors/sanitize.py b/engines/contributors/sanitize.py new file mode 100644 index 0000000..2316b7e --- /dev/null +++ b/engines/contributors/sanitize.py @@ -0,0 +1,129 @@ +"""Sanitize contribution payloads — reuse data scrubbers, never fork them. + +Wraps ``engines.data.scrub`` + ``ensure_no_secret_values``. +Also minimizes PII and private repo identity. +""" + +from __future__ import annotations + +import re +from copy import deepcopy +from pathlib import Path +from typing import Any + +from engines.data.scrub import scrub_example, scrub_text +from engines.dataflow.schema import ensure_no_secret_values + +# Private absolute paths → generic placeholders +_ABS_PATH = re.compile( + r"(?P
^|[\s\"'=:])"
+    r"(?P(?:/Users/|/home/|/private/var/|/var/folders/)[^\s\"']+)"
+)
+_WIN_ABS = re.compile(
+    r"(?P
^|[\s\"'=:])"
+    r"(?P[A-Za-z]:\\(?:Users|home)\\[^\s\"']+)",
+    re.IGNORECASE,
+)
+_EMAIL = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
+_PRIVATE_REPO_HINTS = re.compile(
+    r"(?i)\b(github\.com|gitlab\.com|bitbucket\.org)/([A-Za-z0-9_.-]+)/([A-Za-z0-9_.-]+)"
+)
+
+
+def minimize_absolute_paths(text: str) -> str:
+    """Replace private absolute paths with a neutral placeholder."""
+
+    def _sub(m: re.Match[str]) -> str:
+        return f"{m.group('pre')}[REDACTED_PATH]"
+
+    out = _ABS_PATH.sub(_sub, text)
+    out = _WIN_ABS.sub(_sub, out)
+    return out
+
+
+def minimize_repo_identity(text: str, *, known_private_names: list[str] | None = None) -> str:
+    """Normalize private repo / org names when they appear in free text."""
+    out = text
+    for name in known_private_names or []:
+        if name and len(name) >= 2:
+            out = re.sub(re.escape(name), "[REDACTED_REPO]", out, flags=re.IGNORECASE)
+
+    def _repo_sub(m: re.Match[str]) -> str:
+        host, org, repo = m.group(1), m.group(2), m.group(3)
+        # Keep public Awarexone/AXguard identity; minimize others
+        if org.lower() == "awarexone" and repo.lower() == "axguard":
+            return m.group(0)
+        return f"{host}/[REDACTED_ORG]/[REDACTED_REPO]"
+
+    return _PRIVATE_REPO_HINTS.sub(_repo_sub, out)
+
+
+def minimize_pii(text: str) -> str:
+    out, _ = scrub_text(text)
+    out = _EMAIL.sub("[REDACTED_EMAIL]", out)
+    out = minimize_absolute_paths(out)
+    return out
+
+
+def sanitize_text(
+    text: str,
+    *,
+    known_private_names: list[str] | None = None,
+) -> tuple[str, list[str]]:
+    """Scrub secrets/PII and minimize private identity. Returns (text, hits)."""
+    cleaned, hits = scrub_text(text)
+    cleaned = minimize_absolute_paths(cleaned)
+    cleaned = minimize_repo_identity(cleaned, known_private_names=known_private_names)
+    cleaned = _EMAIL.sub("[REDACTED_EMAIL]", cleaned)
+    return cleaned, hits
+
+
+def sanitize_example(
+    example: dict[str, Any],
+    *,
+    known_private_names: list[str] | None = None,
+) -> tuple[dict[str, Any], list[str]]:
+    """Sanitize a contribution example dict via shared scrubbers + identity minimization."""
+    cleaned, hits = scrub_example(deepcopy(example))
+    ensure_no_secret_values(cleaned)
+    for key, value in list(cleaned.items()):
+        if isinstance(value, str):
+            cleaned[key], extra = sanitize_text(
+                value, known_private_names=known_private_names
+            )
+            hits.extend(extra)
+        elif isinstance(value, dict):
+            nested, extra = sanitize_example(value, known_private_names=known_private_names)
+            cleaned[key] = nested
+            hits.extend(extra)
+    # Never keep identity keys
+    for drop in ("author", "author_email", "github_login", "user", "email", "machine_id"):
+        cleaned.pop(drop, None)
+    if "repo" in cleaned and isinstance(cleaned["repo"], str):
+        cleaned["repo"], _ = sanitize_text(
+            cleaned["repo"], known_private_names=known_private_names
+        )
+    if "path" in cleaned and isinstance(cleaned["path"], str):
+        cleaned["path"] = _relativize_path(cleaned["path"])
+    if "file" in cleaned and isinstance(cleaned["file"], str):
+        cleaned["file"] = _relativize_path(cleaned["file"])
+    ensure_no_secret_values(cleaned)
+    return cleaned, hits
+
+
+def _relativize_path(path_str: str) -> str:
+    """Prefer basename / relative form over private absolute paths."""
+    text = minimize_absolute_paths(path_str)
+    if text == "[REDACTED_PATH]" or "[REDACTED_PATH]" in text:
+        # Keep filename if we can recover it from original
+        try:
+            name = Path(path_str).name
+            if name and name not in (".", ".."):
+                return f"[REDACTED_DIR]/{name}"
+        except (TypeError, ValueError):
+            pass
+        return "[REDACTED_PATH]"
+    p = Path(path_str)
+    if p.is_absolute():
+        return f"[REDACTED_DIR]/{p.name}" if p.name else "[REDACTED_PATH]"
+    return path_str.replace("\\", "/")
diff --git a/engines/contributors/schema.py b/engines/contributors/schema.py
new file mode 100644
index 0000000..5cf4764
--- /dev/null
+++ b/engines/contributors/schema.py
@@ -0,0 +1,221 @@
+"""Contributor engagement & learning — types, provenance, config defaults.
+
+Local-only by default. Learning and contribution packaging are opt-in.
+Scoring measures opportunity usefulness, never personal worth.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# Contribution opportunity types
+# ---------------------------------------------------------------------------
+TYPE_BUG_FIX = "BUG_FIX"
+TYPE_SECURITY_RULE = "SECURITY_RULE"
+TYPE_REGRESSION_TEST = "REGRESSION_TEST"
+TYPE_FALSE_POSITIVE_FIX = "FALSE_POSITIVE_FIX"
+TYPE_FRAMEWORK_SUPPORT = "FRAMEWORK_SUPPORT"
+TYPE_ATTACK_PATH_PATTERN = "ATTACK_PATH_PATTERN"
+TYPE_SYNTHETIC_SECURITY_CASE = "SYNTHETIC_SECURITY_CASE"
+TYPE_MCP_SECURITY_CASE = "MCP_SECURITY_CASE"
+TYPE_DOCUMENTATION = "DOCUMENTATION"
+TYPE_PERFORMANCE = "PERFORMANCE"
+TYPE_DX = "DX"
+TYPE_NOVEL_FINDING = "NOVEL_FINDING"
+TYPE_TEST = "TEST"
+TYPE_DOCS = "DOCS"
+TYPE_RULE = "RULE"
+TYPE_AI_AGENT = "AI_AGENT"
+
+CONTRIBUTION_TYPES = frozenset(
+    {
+        TYPE_BUG_FIX,
+        TYPE_SECURITY_RULE,
+        TYPE_REGRESSION_TEST,
+        TYPE_FALSE_POSITIVE_FIX,
+        TYPE_FRAMEWORK_SUPPORT,
+        TYPE_ATTACK_PATH_PATTERN,
+        TYPE_SYNTHETIC_SECURITY_CASE,
+        TYPE_MCP_SECURITY_CASE,
+        TYPE_DOCUMENTATION,
+        TYPE_PERFORMANCE,
+        TYPE_DX,
+        TYPE_NOVEL_FINDING,
+        TYPE_TEST,
+        TYPE_DOCS,
+        TYPE_RULE,
+        TYPE_AI_AGENT,
+    }
+)
+
+# Human labels (engineer-to-engineer, no hype)
+TYPE_LABELS: dict[str, str] = {
+    TYPE_BUG_FIX: "Bug Fix",
+    TYPE_SECURITY_RULE: "Security Rule",
+    TYPE_REGRESSION_TEST: "Regression Test",
+    TYPE_FALSE_POSITIVE_FIX: "False Positive Fix",
+    TYPE_FRAMEWORK_SUPPORT: "Framework Support",
+    TYPE_ATTACK_PATH_PATTERN: "Attack Path Pattern",
+    TYPE_SYNTHETIC_SECURITY_CASE: "Synthetic Security Case",
+    TYPE_MCP_SECURITY_CASE: "MCP Security Case",
+    TYPE_DOCUMENTATION: "Documentation",
+    TYPE_PERFORMANCE: "Performance",
+    TYPE_DX: "Developer Experience",
+    TYPE_NOVEL_FINDING: "Novel Finding Pattern",
+    TYPE_TEST: "Test",
+    TYPE_DOCS: "Docs",
+    TYPE_RULE: "Rule",
+    TYPE_AI_AGENT: "AI / Agent Security Case",
+}
+
+# ---------------------------------------------------------------------------
+# Difficulty
+# ---------------------------------------------------------------------------
+DIFFICULTY_EASY = "EASY"
+DIFFICULTY_MEDIUM = "MEDIUM"
+DIFFICULTY_ADVANCED = "ADVANCED"
+DIFFICULTY_RESEARCH = "RESEARCH"
+
+DIFFICULTIES = frozenset(
+    {
+        DIFFICULTY_EASY,
+        DIFFICULTY_MEDIUM,
+        DIFFICULTY_ADVANCED,
+        DIFFICULTY_RESEARCH,
+    }
+)
+
+TYPE_DEFAULT_DIFFICULTY: dict[str, str] = {
+    TYPE_DOCUMENTATION: DIFFICULTY_EASY,
+    TYPE_DOCS: DIFFICULTY_EASY,
+    TYPE_DX: DIFFICULTY_EASY,
+    TYPE_TEST: DIFFICULTY_EASY,
+    TYPE_FALSE_POSITIVE_FIX: DIFFICULTY_MEDIUM,
+    TYPE_REGRESSION_TEST: DIFFICULTY_MEDIUM,
+    TYPE_BUG_FIX: DIFFICULTY_MEDIUM,
+    TYPE_PERFORMANCE: DIFFICULTY_MEDIUM,
+    TYPE_SECURITY_RULE: DIFFICULTY_ADVANCED,
+    TYPE_RULE: DIFFICULTY_ADVANCED,
+    TYPE_FRAMEWORK_SUPPORT: DIFFICULTY_ADVANCED,
+    TYPE_ATTACK_PATH_PATTERN: DIFFICULTY_ADVANCED,
+    TYPE_NOVEL_FINDING: DIFFICULTY_ADVANCED,
+    TYPE_SYNTHETIC_SECURITY_CASE: DIFFICULTY_RESEARCH,
+    TYPE_MCP_SECURITY_CASE: DIFFICULTY_RESEARCH,
+    TYPE_AI_AGENT: DIFFICULTY_RESEARCH,
+}
+
+# ---------------------------------------------------------------------------
+# Provenance (contribution / learning pack lifecycle)
+# ---------------------------------------------------------------------------
+PROVENANCE_LOCAL_ONLY = "LOCAL_ONLY"
+PROVENANCE_USER_APPROVED = "USER_APPROVED"
+PROVENANCE_SANITIZED = "SANITIZED"
+PROVENANCE_VALIDATED = "VALIDATED"
+PROVENANCE_DATASET_APPROVED = "DATASET_APPROVED"
+PROVENANCE_EXCLUDED = "EXCLUDED"
+
+PROVENANCE_STATES = frozenset(
+    {
+        PROVENANCE_LOCAL_ONLY,
+        PROVENANCE_USER_APPROVED,
+        PROVENANCE_SANITIZED,
+        PROVENANCE_VALIDATED,
+        PROVENANCE_DATASET_APPROVED,
+        PROVENANCE_EXCLUDED,
+    }
+)
+
+# ---------------------------------------------------------------------------
+# Quality gate thresholds (opportunity existence — not personal worth)
+# ---------------------------------------------------------------------------
+QUALITY_MIN_SCORE = 0.65
+OPPORTUNITY_MIN_SCORE = 0.55
+INVITE_COOLDOWN_SEC = 4 * 3600  # one invite per meaningful session window
+MAX_INVITES_PER_SESSION = 1
+
+# ---------------------------------------------------------------------------
+# Config defaults (YAML-shaped; local prefs may override)
+# ---------------------------------------------------------------------------
+DEFAULT_CONTRIBUTIONS_CONFIG: dict[str, Any] = {
+    "enabled": True,
+    "prompts": True,
+    "auto_prepare": False,
+    "auto_push": False,
+    "auto_pr": False,
+    "learning": False,  # opt-in
+}
+
+DEFAULT_LEARNING_CONFIG: dict[str, Any] = {
+    "contribution_data": {
+        "enabled": False,
+    }
+}
+
+DEFAULT_PRIVACY_CONFIG: dict[str, Any] = {
+    "contributions": dict(DEFAULT_CONTRIBUTIONS_CONFIG),
+    "learning": dict(DEFAULT_LEARNING_CONFIG),
+    "contribute_opt_in": False,
+    "learning_opt_in": False,
+    "never_prompts": False,
+    "dismissed_types": [],
+    "version": 1,
+}
+
+# Fields included / excluded from local learning export (documented for transparency)
+EXPORT_INCLUDED_FIELDS: tuple[str, ...] = (
+    "contribution_type",
+    "difficulty",
+    "language",
+    "framework",
+    "pattern",
+    "flow_summary",
+    "missing_control",
+    "sanitized_snippet",
+    "provenance",
+    "quality_scores",
+    "created_at",
+)
+
+EXPORT_EXCLUDED_FIELDS: tuple[str, ...] = (
+    "author_name",
+    "author_email",
+    "github_login",
+    "absolute_paths",
+    "private_repo_name",
+    "private_org_name",
+    "raw_secrets",
+    "env_values",
+    "tokens",
+    "telemetry",
+    "ip_address",
+    "machine_id",
+)
+
+# Local paths
+PRIVACY_FILENAME = "privacy.json"
+CONTRIBUTE_STATE_FILENAME = "contribute-state.json"
+CONTRIBUTE_PACKAGE_DIRNAME = "contribute"
+
+# Phrases that must never appear in contributor recognition / invite copy
+FORBIDDEN_CONTRIBUTE_PHRASES: tuple[str, ...] = (
+    "you're a legend",
+    "you are a legend",
+    "rockstar",
+    "ninja",
+    "guilt",
+    "don't let us down",
+    "don't miss",
+    "limited time",
+    "act now",
+    "last chance",
+    "only you can",
+    "top contributor ranking",
+    "leaderboard",
+    "streak at risk",
+    "keep your streak",
+    "fake ranking",
+    "everyone else already",
+    "exclusive community",
+    "revolutionary",
+)
diff --git a/engines/contributors/signals.py b/engines/contributors/signals.py
new file mode 100644
index 0000000..3c162a5
--- /dev/null
+++ b/engines/contributors/signals.py
@@ -0,0 +1,208 @@
+"""Map observable work → contribution opportunity score + type.
+
+Scores opportunity existence only — never personal worth or ranking.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from engines.contributors.schema import (
+    DIFFICULTY_ADVANCED,
+    DIFFICULTY_EASY,
+    DIFFICULTY_MEDIUM,
+    DIFFICULTY_RESEARCH,
+    OPPORTUNITY_MIN_SCORE,
+    TYPE_AI_AGENT,
+    TYPE_ATTACK_PATH_PATTERN,
+    TYPE_DOCS,
+    TYPE_FALSE_POSITIVE_FIX,
+    TYPE_MCP_SECURITY_CASE,
+    TYPE_NOVEL_FINDING,
+    TYPE_REGRESSION_TEST,
+    TYPE_RULE,
+    TYPE_SECURITY_RULE,
+    TYPE_TEST,
+    TYPE_DEFAULT_DIFFICULTY,
+)
+
+
+@dataclass(frozen=True)
+class OpportunitySignal:
+    """A candidate contribution opportunity derived from local work."""
+
+    contribution_type: str
+    score: float
+    difficulty: str
+    reason: str
+    evidence: dict[str, Any] = field(default_factory=dict)
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "contribution_type": self.contribution_type,
+            "score": round(self.score, 3),
+            "difficulty": self.difficulty,
+            "reason": self.reason,
+            "evidence": dict(self.evidence),
+        }
+
+
+def _clamp(value: float) -> float:
+    return max(0.0, min(1.0, float(value)))
+
+
+def _difficulty_for(contribution_type: str) -> str:
+    return TYPE_DEFAULT_DIFFICULTY.get(contribution_type, DIFFICULTY_MEDIUM)
+
+
+def signals_from_context(context: dict[str, Any] | None) -> list[OpportunitySignal]:
+    """Derive opportunity signals from audit / adversary / memory context."""
+    ctx = context or {}
+    signals: list[OpportunitySignal] = []
+
+    fp = int(ctx.get("fp_rejected") or ctx.get("rejected_count") or 0)
+    if fp > 0:
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_FALSE_POSITIVE_FIX,
+                score=_clamp(0.55 + min(fp, 5) * 0.08),
+                difficulty=_difficulty_for(TYPE_FALSE_POSITIVE_FIX),
+                reason="A false positive was correctly rejected — a reusable FP fix or corpus case may help.",
+                evidence={"fp_rejected": fp},
+            )
+        )
+
+    verified = int(ctx.get("verified_count") or 0)
+    novel = bool(ctx.get("novel_finding") or ctx.get("first_verified"))
+    if verified > 0 and novel:
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_NOVEL_FINDING,
+                score=_clamp(0.6 + min(verified, 3) * 0.1),
+                difficulty=_difficulty_for(TYPE_NOVEL_FINDING),
+                reason="A verified finding looks novel enough to document as a pattern or rule case.",
+                evidence={"verified_count": verified, "novel": True},
+            )
+        )
+    elif verified > 0:
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_REGRESSION_TEST,
+                score=_clamp(0.5 + min(verified, 4) * 0.07),
+                difficulty=_difficulty_for(TYPE_REGRESSION_TEST),
+                reason="A verified finding can become a regression fixture so it stays covered.",
+                evidence={"verified_count": verified},
+            )
+        )
+
+    if ctx.get("regression") or int(ctx.get("regression_count") or 0) > 0:
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_REGRESSION_TEST,
+                score=_clamp(0.75),
+                difficulty=DIFFICULTY_MEDIUM,
+                reason="A regression was observed — a focused regression test would lock the fix in.",
+                evidence={"regression": True},
+            )
+        )
+
+    paths = int(ctx.get("attack_paths_confirmed") or 0)
+    if paths > 0:
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_ATTACK_PATH_PATTERN,
+                score=_clamp(0.62 + min(paths, 3) * 0.08),
+                difficulty=_difficulty_for(TYPE_ATTACK_PATH_PATTERN),
+                reason="A confirmed attack path can be captured as a reusable chain pattern.",
+                evidence={"attack_paths_confirmed": paths},
+            )
+        )
+
+    if ctx.get("rule_gap") or ctx.get("missing_rule"):
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_SECURITY_RULE,
+                score=_clamp(0.7),
+                difficulty=DIFFICULTY_ADVANCED,
+                reason="Observable work suggests a detection gap that a rule could cover.",
+                evidence={"rule_gap": True},
+            )
+        )
+    elif ctx.get("rule_candidate"):
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_RULE,
+                score=_clamp(0.58),
+                difficulty=DIFFICULTY_ADVANCED,
+                reason="A rule refinement candidate was observed in this session.",
+                evidence={"rule_candidate": True},
+            )
+        )
+
+    if ctx.get("test_gap") or ctx.get("missing_test"):
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_TEST,
+                score=_clamp(0.6),
+                difficulty=DIFFICULTY_EASY,
+                reason="A test gap was observed for a security behavior that already has evidence.",
+                evidence={"test_gap": True},
+            )
+        )
+
+    if ctx.get("docs_gap") or ctx.get("missing_docs"):
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_DOCS,
+                score=_clamp(0.55),
+                difficulty=DIFFICULTY_EASY,
+                reason="Documentation would clarify a behavior this session already exercised.",
+                evidence={"docs_gap": True},
+            )
+        )
+
+    if ctx.get("mcp") or ctx.get("mcp_finding") or str(ctx.get("domain") or "").upper() == "MCP":
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_MCP_SECURITY_CASE,
+                score=_clamp(0.68),
+                difficulty=DIFFICULTY_RESEARCH,
+                reason="MCP-related security work can become a shared MCP security case.",
+                evidence={"mcp": True},
+            )
+        )
+
+    if ctx.get("ai_agent") or ctx.get("agent_finding") or str(ctx.get("domain") or "").upper() in {
+        "AI_AGENT",
+        "AI_SECURITY",
+    }:
+        signals.append(
+            OpportunitySignal(
+                contribution_type=TYPE_AI_AGENT,
+                score=_clamp(0.66),
+                difficulty=DIFFICULTY_RESEARCH,
+                reason="AI-agent security behavior observed here could seed a reusable case.",
+                evidence={"ai_agent": True},
+            )
+        )
+
+    # Deduplicate by type — keep highest score
+    best: dict[str, OpportunitySignal] = {}
+    for sig in signals:
+        prev = best.get(sig.contribution_type)
+        if prev is None or sig.score > prev.score:
+            best[sig.contribution_type] = sig
+    return sorted(best.values(), key=lambda s: s.score, reverse=True)
+
+
+def best_opportunity(
+    context: dict[str, Any] | None,
+    *,
+    min_score: float = OPPORTUNITY_MIN_SCORE,
+) -> OpportunitySignal | None:
+    ranked = signals_from_context(context)
+    for sig in ranked:
+        if sig.score >= min_score:
+            return sig
+    return None
diff --git a/engines/contributors/suggest.py b/engines/contributors/suggest.py
new file mode 100644
index 0000000..b894386
--- /dev/null
+++ b/engines/contributors/suggest.py
@@ -0,0 +1,250 @@
+"""Suggest a specific contribution — cooldown, dismissal, no spam."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from engines.contributors import milestones as mile
+from engines.contributors.privacy import (
+    dismiss_type,
+    is_type_dismissed,
+    prompts_allowed,
+    set_never_prompts,
+)
+from engines.contributors.quality import filter_strong
+from engines.contributors.schema import (
+    FORBIDDEN_CONTRIBUTE_PHRASES,
+    INVITE_COOLDOWN_SEC,
+    MAX_INVITES_PER_SESSION,
+    TYPE_LABELS,
+)
+from engines.contributors.signals import OpportunitySignal, signals_from_context
+
+
+@dataclass(frozen=True)
+class ContributionInvite:
+    id: str
+    contribution_type: str
+    difficulty: str
+    reason: str
+    what: str
+    control: str
+    quality_score: float
+    message: str
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "contribution_type": self.contribution_type,
+            "difficulty": self.difficulty,
+            "reason": self.reason,
+            "what": self.what,
+            "control": self.control,
+            "quality_score": self.quality_score,
+            "message": self.message,
+        }
+
+
+def _parse_iso(value: str | None) -> datetime | None:
+    if not value or not isinstance(value, str):
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _clean(text: str) -> str:
+    lowered = text.lower()
+    for phrase in FORBIDDEN_CONTRIBUTE_PHRASES:
+        if phrase in lowered:
+            raise ValueError(f"forbidden contribute phrase: {phrase}")
+    return text
+
+
+def _in_cooldown(state: dict[str, Any]) -> bool:
+    last = _parse_iso(state.get("last_invite_at"))
+    if last is None:
+        return False
+    age = (_now() - last).total_seconds()
+    return age < INVITE_COOLDOWN_SEC
+
+
+def _session_cap_reached(state: dict[str, Any], session_id: str | None) -> bool:
+    if session_id and state.get("session_id") == session_id:
+        return int(state.get("session_invite_count") or 0) >= MAX_INVITES_PER_SESSION
+    return False
+
+
+def _build_invite(signal: OpportunitySignal, quality_score: float) -> ContributionInvite:
+    label = TYPE_LABELS.get(signal.contribution_type, signal.contribution_type)
+    reason = signal.reason
+    what = (
+        f"Prepare a local {label} package (fixture/test/docs stubs + commit/PR draft). "
+        "Nothing is pushed."
+    )
+    control = (
+        "You can dismiss with: not now · never prompts · don't suggest this type. "
+        "Push/PR only happen if you do them yourself."
+    )
+    message = _clean(
+        "\n".join(
+            [
+                f"Suggested contribution: {label} ({signal.difficulty})",
+                "",
+                f"Why: {reason}",
+                f"What: {what}",
+                f"Control: {control}",
+                "",
+                "Run: axguard contribute prepare",
+            ]
+        )
+    )
+    return ContributionInvite(
+        id=f"inv_{uuid4().hex[:10]}",
+        contribution_type=signal.contribution_type,
+        difficulty=signal.difficulty,
+        reason=reason,
+        what=what,
+        control=control,
+        quality_score=quality_score,
+        message=message,
+    )
+
+
+def suggest(
+    context: dict[str, Any] | None = None,
+    *,
+    privacy_path: Path | None = None,
+    state_path: Path | None = None,
+    session_id: str | None = None,
+    force: bool = False,
+) -> ContributionInvite | None:
+    """Return at most one strong invite per meaningful session (unless force)."""
+    if not force and not prompts_allowed(privacy_path):
+        return None
+
+    state = mile.load_state(state_path)
+    if not force:
+        if _in_cooldown(state):
+            return None
+        if _session_cap_reached(state, session_id):
+            return None
+
+    strong = filter_strong(signals_from_context(context), context)
+    chosen: OpportunitySignal | None = None
+    quality_score = 0.0
+    for sig, qa in strong:
+        if is_type_dismissed(sig.contribution_type, path=privacy_path):
+            continue
+        chosen = sig
+        quality_score = qa.score
+        break
+    if chosen is None:
+        return None
+
+    invite = _build_invite(chosen, quality_score)
+
+    # Persist cooldown / session accounting
+    sid = session_id or state.get("session_id") or uuid4().hex[:12]
+    count = int(state.get("session_invite_count") or 0)
+    if state.get("session_id") == sid:
+        count += 1
+    else:
+        count = 1
+    state["session_id"] = sid
+    state["session_invite_count"] = count
+    state["last_invite_at"] = _now().replace(microsecond=0).isoformat()
+    state["last_invite_type"] = chosen.contribution_type
+    mile.save_state(state, state_path)
+
+    # Soft engagement event (best-effort)
+    try:
+        from engines.engagement.engine import record_event
+        from engines.engagement.schema import EVENT_CONTRIBUTION_SUGGESTED
+
+        record_event(
+            EVENT_CONTRIBUTION_SUGGESTED,
+            {
+                **(context or {}),
+                "count_meaningful": False,
+                "contribution_type": chosen.contribution_type,
+                "invite_id": invite.id,
+            },
+        )
+    except Exception:  # noqa: BLE001
+        pass
+
+    return invite
+
+
+def dismiss(
+    mode: str,
+    *,
+    contribution_type: str | None = None,
+    privacy_path: Path | None = None,
+    state_path: Path | None = None,
+) -> dict[str, Any]:
+    """Dismiss invites: not_now | never_prompts | type.
+
+    Never spam after dismiss.
+    """
+    normalized = (mode or "").strip().lower().replace("-", "_").replace(" ", "_")
+    result: dict[str, Any] = {"mode": normalized, "ok": True}
+
+    state = mile.load_state(state_path)
+    dismissed = list(state.get("dismissed_invites") or [])
+    dismissed.append(
+        {
+            "at": _now().replace(microsecond=0).isoformat(),
+            "mode": normalized,
+            "contribution_type": contribution_type,
+        }
+    )
+    state["dismissed_invites"] = dismissed[-50:]
+
+    if normalized in {"not_now", "notnow", "later"}:
+        # Start cooldown immediately
+        state["last_invite_at"] = _now().replace(microsecond=0).isoformat()
+        state["session_invite_count"] = MAX_INVITES_PER_SESSION
+        result["effect"] = "cooldown"
+    elif normalized in {"never_prompts", "never", "no_prompts"}:
+        set_never_prompts(True, path=privacy_path)
+        result["effect"] = "never_prompts"
+    elif normalized in {"type", "dont_suggest_type", "don't_suggest_this_type", "dismiss_type"}:
+        if not contribution_type:
+            # Fall back to last invite type
+            contribution_type = state.get("last_invite_type")
+        if contribution_type:
+            dismiss_type(contribution_type, path=privacy_path)
+            result["effect"] = "type_dismissed"
+            result["contribution_type"] = contribution_type
+        else:
+            result["ok"] = False
+            result["error"] = "contribution_type required"
+    else:
+        result["ok"] = False
+        result["error"] = "unknown dismiss mode (use not_now | never_prompts | type)"
+
+    mile.save_state(state, state_path)
+
+    try:
+        from engines.engagement.engine import record_event
+        from engines.engagement.schema import EVENT_CONTRIBUTION_DISMISSED
+
+        record_event(
+            EVENT_CONTRIBUTION_DISMISSED,
+            {"count_meaningful": False, "dismiss_mode": normalized},
+        )
+    except Exception:  # noqa: BLE001
+        pass
+
+    return result
diff --git a/engines/contributors/templates.py b/engines/contributors/templates.py
new file mode 100644
index 0000000..0a403d6
--- /dev/null
+++ b/engines/contributors/templates.py
@@ -0,0 +1,270 @@
+"""Contribution templates — concrete stubs, no hype."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from engines.contributors.schema import (
+    TYPE_ATTACK_PATH_PATTERN,
+    TYPE_BUG_FIX,
+    TYPE_DOCUMENTATION,
+    TYPE_DX,
+    TYPE_FALSE_POSITIVE_FIX,
+    TYPE_FRAMEWORK_SUPPORT,
+    TYPE_LABELS,
+    TYPE_MCP_SECURITY_CASE,
+    TYPE_PERFORMANCE,
+    TYPE_REGRESSION_TEST,
+    TYPE_SECURITY_RULE,
+    TYPE_SYNTHETIC_SECURITY_CASE,
+)
+
+TEMPLATES: dict[str, dict[str, Any]] = {
+    TYPE_BUG_FIX: {
+        "title": "Bug Fix",
+        "files": {
+            "README.md": (
+                "# Bug Fix contribution\n\n"
+                "## What broke\n\nDescribe the incorrect behavior.\n\n"
+                "## Root cause\n\nShort technical note.\n\n"
+                "## Fix\n\nWhat changed and why it is safe.\n\n"
+                "## Test plan\n\n- [ ] Repro before fix\n- [ ] Pass after fix\n"
+            ),
+            "COMMIT_MSG.txt": "fix: describe the bug and the safe correction.\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Fixes \n\n"
+                "## Test plan\n- [ ] Unit / regression coverage\n"
+            ),
+        },
+    },
+    TYPE_SECURITY_RULE: {
+        "title": "Security Rule",
+        "files": {
+            "README.md": (
+                "# Security Rule contribution\n\n"
+                "## Pattern\n\nWhat the rule should detect.\n\n"
+                "## False positive notes\n\nWhen it must stay quiet.\n\n"
+                "## Fixtures\n\nAdd positive and negative examples under `fixtures/`.\n"
+            ),
+            "rule.yaml.stub": (
+                "id: AXG-TODO\n"
+                "severity: medium\n"
+                "title: TODO\n"
+                "description: TODO\n"
+                "patterns:\n  - TODO\n"
+            ),
+            "COMMIT_MSG.txt": "rules: add detection for .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Adds rule for \n\n"
+                "## Test plan\n- [ ] Positive fixture hits\n- [ ] Negative fixture stays clean\n"
+            ),
+        },
+    },
+    TYPE_REGRESSION_TEST: {
+        "title": "Regression Test",
+        "files": {
+            "README.md": (
+                "# Regression Test contribution\n\n"
+                "## Failure mode\n\nWhat regressed.\n\n"
+                "## Fixture\n\nMinimal case under `fixtures/`.\n"
+            ),
+            "test_regression.py.stub": (
+                "\"\"\"Regression coverage for .\"\"\"\n\n"
+                "def test_regression_case():\n"
+                "    assert True  # replace with real assertion\n"
+            ),
+            "COMMIT_MSG.txt": "test: add regression coverage for .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Adds regression test for \n\n"
+                "## Test plan\n- [ ] Fails without the fix\n- [ ] Passes with the fix\n"
+            ),
+        },
+    },
+    TYPE_FALSE_POSITIVE_FIX: {
+        "title": "False Positive Fix",
+        "files": {
+            "README.md": (
+                "# False Positive Fix contribution\n\n"
+                "## Why it looked dangerous\n\n"
+                "## Control that made it safe\n\n"
+                "## Proposed change\n\nRule / adversary / corpus update.\n"
+            ),
+            "fp_case.json.stub": (
+                "{\n"
+                '  "status": "FALSE_POSITIVE",\n'
+                '  "reason": "TODO",\n'
+                '  "control": "TODO"\n'
+                "}\n"
+            ),
+            "COMMIT_MSG.txt": "fix(fp): stop reporting safe .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Documents / fixes a false positive\n\n"
+                "## Test plan\n- [ ] Case stays rejected\n"
+            ),
+        },
+    },
+    TYPE_FRAMEWORK_SUPPORT: {
+        "title": "Framework Support",
+        "files": {
+            "README.md": (
+                "# Framework Support contribution\n\n"
+                "## Framework\n\n## Entry points / sinks observed\n\n"
+                "## Adapter notes\n"
+            ),
+            "COMMIT_MSG.txt": "feat: add support for .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Framework support for \n\n"
+                "## Test plan\n- [ ] Surface / flow fixtures\n"
+            ),
+        },
+    },
+    TYPE_ATTACK_PATH_PATTERN: {
+        "title": "Attack Path Pattern",
+        "files": {
+            "README.md": (
+                "# Attack Path Pattern contribution\n\n"
+                "## Chain\n\nSource → steps → sink\n\n"
+                "## Evidence that confirmed it\n"
+            ),
+            "path_pattern.json.stub": (
+                "{\n"
+                '  "source": "TODO",\n'
+                '  "steps": [],\n'
+                '  "sink": "TODO"\n'
+                "}\n"
+            ),
+            "COMMIT_MSG.txt": "docs/patterns: capture attack path .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Captures a confirmed attack-path pattern\n\n"
+                "## Test plan\n- [ ] Pattern reproduces on fixture\n"
+            ),
+        },
+    },
+    TYPE_SYNTHETIC_SECURITY_CASE: {
+        "title": "Synthetic Security Case",
+        "files": {
+            "README.md": (
+                "# Synthetic Security Case\n\n"
+                "Opt-in learning pack only. No private customer code.\n\n"
+                "## Intended lesson\n"
+            ),
+            "case.json.stub": "{\n  \"category\": \"TODO\",\n  \"expected\": \"TODO\"\n}\n",
+            "COMMIT_MSG.txt": "data: add synthetic security case .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Synthetic case for evaluation / research\n\n"
+                "## Test plan\n- [ ] Scrubbed\n- [ ] License-safe\n"
+            ),
+        },
+    },
+    TYPE_MCP_SECURITY_CASE: {
+        "title": "MCP Security Case",
+        "files": {
+            "README.md": (
+                "# MCP Security Case\n\n"
+                "## Tool / resource surface\n\n## Risk\n\n## Expected control\n"
+            ),
+            "mcp_case.json.stub": (
+                "{\n"
+                '  "surface": "TODO",\n'
+                '  "risk": "TODO",\n'
+                '  "control": "TODO"\n'
+                "}\n"
+            ),
+            "COMMIT_MSG.txt": "data: add MCP security case .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- MCP security case\n\n"
+                "## Test plan\n- [ ] No secrets\n- [ ] Reproducible locally\n"
+            ),
+        },
+    },
+    TYPE_DOCUMENTATION: {
+        "title": "Documentation",
+        "files": {
+            "README.md": (
+                "# Documentation contribution\n\n"
+                "## Audience\n\n## Gap observed\n\n## Proposed page / section\n"
+            ),
+            "COMMIT_MSG.txt": "docs: clarify .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Docs update for \n\n"
+                "## Test plan\n- [ ] Links resolve\n"
+            ),
+        },
+    },
+    TYPE_PERFORMANCE: {
+        "title": "Performance",
+        "files": {
+            "README.md": (
+                "# Performance contribution\n\n"
+                "## Slow path observed\n\n## Measurement\n\n## Change\n"
+            ),
+            "COMMIT_MSG.txt": "perf: improve .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- Performance improvement\n\n"
+                "## Test plan\n- [ ] Before/after notes\n"
+            ),
+        },
+    },
+    TYPE_DX: {
+        "title": "Developer Experience",
+        "files": {
+            "README.md": (
+                "# DX contribution\n\n"
+                "## Friction observed\n\n## Proposed improvement\n"
+            ),
+            "COMMIT_MSG.txt": "dx: improve .\n",
+            "PR_DRAFT.md": (
+                "## Summary\n- DX improvement\n\n"
+                "## Test plan\n- [ ] Command / docs still work\n"
+            ),
+        },
+    },
+}
+
+
+def list_templates() -> list[dict[str, str]]:
+    out = []
+    for key, tmpl in TEMPLATES.items():
+        out.append(
+            {
+                "type": key,
+                "title": str(tmpl.get("title") or TYPE_LABELS.get(key, key)),
+            }
+        )
+    return out
+
+
+def get_template(contribution_type: str) -> dict[str, Any] | None:
+    if contribution_type in TEMPLATES:
+        return dict(TEMPLATES[contribution_type])
+    # Aliases
+    aliases = {
+        "DOCS": TYPE_DOCUMENTATION,
+        "TEST": TYPE_REGRESSION_TEST,
+        "RULE": TYPE_SECURITY_RULE,
+        "NOVEL_FINDING": TYPE_SECURITY_RULE,
+        "AI_AGENT": TYPE_MCP_SECURITY_CASE,
+    }
+    mapped = aliases.get(contribution_type)
+    if mapped and mapped in TEMPLATES:
+        return dict(TEMPLATES[mapped])
+    return None
+
+
+def render_template_files(
+    contribution_type: str,
+    *,
+    substitutions: dict[str, str] | None = None,
+) -> dict[str, str]:
+    tmpl = get_template(contribution_type)
+    if tmpl is None:
+        return {}
+    subs = substitutions or {}
+    files = dict(tmpl.get("files") or {})
+    rendered: dict[str, str] = {}
+    for name, body in files.items():
+        text = str(body)
+        for key, value in subs.items():
+            text = text.replace(f"<{key}>", value)
+        rendered[name] = text
+    return rendered
diff --git a/engines/engagement/__init__.py b/engines/engagement/__init__.py
index 357e719..b80c3a2 100644
--- a/engines/engagement/__init__.py
+++ b/engines/engagement/__init__.py
@@ -33,6 +33,8 @@
     CTA_LEARN,
     CTA_SHARE,
     CTA_SUPPORT,
+    EVENT_CONTRIBUTION_DISMISSED,
+    EVENT_CONTRIBUTION_SUGGESTED,
     EVENTS,
     GITHUB_URL,
     SIGNATURE,
@@ -80,4 +82,6 @@
     "SUPPORT_SCORE_THRESHOLD",
     "GITHUB_URL",
     "SIGNATURE",
+    "EVENT_CONTRIBUTION_SUGGESTED",
+    "EVENT_CONTRIBUTION_DISMISSED",
 ]
diff --git a/engines/engagement/messages.py b/engines/engagement/messages.py
index 58b17d2..a089ff2 100644
--- a/engines/engagement/messages.py
+++ b/engines/engagement/messages.py
@@ -17,6 +17,7 @@
     EVENT_ABOUT_VIEWED,
     EVENT_ATTACK_PATH_CONFIRMED,
     EVENT_AUDIT_COMPLETE,
+    EVENT_CONTRIBUTION_SUGGESTED,
     EVENT_FIRST_RUN,
     EVENT_FIX_VERIFIED,
     EVENT_FP_REJECTED,
@@ -431,6 +432,36 @@ def _cta(kind: str, label: str, url: str | None = None) -> EngagementCTA:
     priority=PRIORITY_PRODUCT_INFO,
 )
 
+CONTRIBUTE_FP_RECOGNITION = EngagementMessage(
+    id="contribute_fp_recognition",
+    stage=STATE_FIRST_MILESTONE,
+    event=EVENT_CONTRIBUTION_SUGGESTED,
+    body_lines=(
+        "AXGuard rejected a false positive in this session.",
+        "",
+        "That control-aware rejection is useful signal — the kind that improves rules and corpus cases.",
+        "",
+        "If you want, you can prepare a local false-positive fix package. Nothing is pushed unless you do it.",
+    ),
+    signature=SIGNATURE,
+    cta=_cta(CTA_CONTRIBUTE, "Optional: axguard contribute prepare"),
+    priority=PRIORITY_PRODUCT_INFO,
+)
+
+CONTRIBUTE_VERIFIED_RECOGNITION = EngagementMessage(
+    id="contribute_verified_recognition",
+    stage=STATE_FIRST_MILESTONE,
+    event=EVENT_CONTRIBUTION_SUGGESTED,
+    body_lines=(
+        "A finding was verified with evidence in this session.",
+        "",
+        "If the pattern is reusable, a local contribution package can capture it — only if you want.",
+    ),
+    signature=SIGNATURE,
+    cta=_cta(CTA_CONTRIBUTE, "Optional: axguard contribute suggest"),
+    priority=PRIORITY_PRODUCT_INFO,
+)
+
 
 def _with_cta(msg: EngagementMessage, cta: EngagementCTA | None) -> EngagementMessage:
     return EngagementMessage(
@@ -552,6 +583,22 @@ def select_message(
             )
         return ABOUT_PAGE
 
+    # Contribution recognition (optional CTA_CONTRIBUTE — not a star ask)
+    if event == EVENT_CONTRIBUTION_SUGGESTED:
+        if int(ctx.get("fp_rejected") or ctx.get("rejected_count") or 0) > 0:
+            if CONTRIBUTE_FP_RECOGNITION.id not in seen:
+                return CONTRIBUTE_FP_RECOGNITION
+        if (
+            int(ctx.get("verified_count") or 0) > 0
+            or ctx.get("novel_finding")
+            or ctx.get("first_verified")
+        ):
+            if CONTRIBUTE_VERIFIED_RECOGNITION.id not in seen:
+                return CONTRIBUTE_VERIFIED_RECOGNITION
+        if COMMUNITY_BELONGING.id not in seen:
+            return COMMUNITY_BELONGING
+        return None
+
     # First run — once
     if event == EVENT_FIRST_RUN or (stage == STATE_FIRST_RUN and not state.get("first_run_shown")):
         return FIRST_RUN_INTRO
@@ -676,5 +723,7 @@ def select_message(
         VARIANT_SSRF_FLOW,
         VARIANT_SSRF_CHAIN,
         CONTROL_MATTERS,
+        CONTRIBUTE_FP_RECOGNITION,
+        CONTRIBUTE_VERIFIED_RECOGNITION,
     )
 }
diff --git a/engines/engagement/schema.py b/engines/engagement/schema.py
index a4f7645..c6e3de0 100644
--- a/engines/engagement/schema.py
+++ b/engines/engagement/schema.py
@@ -52,6 +52,8 @@
 EVENT_DISABLE_PROMO = "DISABLE_PROMO"
 EVENT_ENABLE_PROMO = "ENABLE_PROMO"
 EVENT_SESSION_RETURN = "SESSION_RETURN"
+EVENT_CONTRIBUTION_SUGGESTED = "CONTRIBUTION_SUGGESTED"
+EVENT_CONTRIBUTION_DISMISSED = "CONTRIBUTION_DISMISSED"
 
 EVENTS = frozenset(
     {
@@ -67,6 +69,8 @@
         EVENT_DISABLE_PROMO,
         EVENT_ENABLE_PROMO,
         EVENT_SESSION_RETURN,
+        EVENT_CONTRIBUTION_SUGGESTED,
+        EVENT_CONTRIBUTION_DISMISSED,
     }
 )
 
diff --git a/tests/test_contributors.py b/tests/test_contributors.py
new file mode 100644
index 0000000..1751031
--- /dev/null
+++ b/tests/test_contributors.py
@@ -0,0 +1,386 @@
+"""Tests for the Contributor Engagement & Learning System."""
+
+from __future__ import annotations
+
+import ast
+import json
+import socket
+from pathlib import Path
+
+import pytest
+
+from engines.contributors.milestones import clear_state as clear_contribute_state
+from engines.contributors.prepare import prepare
+from engines.contributors.privacy import (
+    delete_learning_data,
+    is_contribute_opted_in,
+    is_learning_opted_in,
+    learning_data_dir,
+    opt_in,
+    reset as privacy_reset,
+    status as privacy_status,
+)
+from engines.contributors.quality import filter_strong
+from engines.contributors.recognize import (
+    CONTRIBUTE_MESSAGE_CATALOG,
+    build_recognition_message,
+)
+from engines.contributors.sanitize import sanitize_example, sanitize_text
+from engines.contributors.schema import FORBIDDEN_CONTRIBUTE_PHRASES
+from engines.contributors.signals import signals_from_context
+from engines.contributors.suggest import dismiss, suggest
+from engines.engagement.messages import MESSAGE_CATALOG
+from engines.engagement.policy import contains_banned_pattern, reject_if_manipulative
+
+ROOT = Path(__file__).resolve().parents[1]
+CONTRIB_PKG = ROOT / "engines" / "contributors"
+
+
+@pytest.fixture
+def privacy_path(tmp_path: Path) -> Path:
+    return tmp_path / "privacy.json"
+
+
+@pytest.fixture
+def contribute_state_path(tmp_path: Path) -> Path:
+    return tmp_path / "contribute-state.json"
+
+
+# ---------------------------------------------------------------------------
+# Recognition
+# ---------------------------------------------------------------------------
+
+
+def test_recognition_on_meaningful_fp_discovery():
+    msg = build_recognition_message({"fp_rejected": 2, "rejected_count": 2})
+    assert msg is not None
+    text = "\n".join(msg.body_lines).lower()
+    assert "false positive" in text
+    assert msg.cta is not None
+    assert msg.cta.kind == "CONTRIBUTE"
+
+
+def test_recognition_on_verified_finding():
+    msg = build_recognition_message({"verified_count": 1, "novel_finding": True})
+    assert msg is not None
+    text = "\n".join(msg.body_lines).lower()
+    assert "verified" in text
+
+
+# ---------------------------------------------------------------------------
+# Suggest / cooldown / dismissal
+# ---------------------------------------------------------------------------
+
+
+def test_suggest_type_cooldown_and_dismissal(
+    privacy_path: Path,
+    contribute_state_path: Path,
+):
+    ctx = {"fp_rejected": 3, "security_primary": True}
+    invite = suggest(
+        ctx,
+        privacy_path=privacy_path,
+        state_path=contribute_state_path,
+        session_id="s1",
+    )
+    assert invite is not None
+    assert invite.contribution_type == "FALSE_POSITIVE_FIX"
+    assert "Suggested contribution" in invite.message
+
+    # Cooldown / max 1 per session — no spam
+    second = suggest(
+        ctx,
+        privacy_path=privacy_path,
+        state_path=contribute_state_path,
+        session_id="s1",
+    )
+    assert second is None
+
+    dismiss("not_now", privacy_path=privacy_path, state_path=contribute_state_path)
+    third = suggest(
+        ctx,
+        privacy_path=privacy_path,
+        state_path=contribute_state_path,
+        session_id="s2",
+        force=False,
+    )
+    assert third is None  # still in cooldown from not_now
+
+    # Type dismiss — even with force, type stays suppressed
+    # Reset cooldown by clearing last_invite (simulate) via dismiss type on fresh state
+    clear_contribute_state(contribute_state_path)
+    invite2 = suggest(
+        ctx,
+        privacy_path=privacy_path,
+        state_path=contribute_state_path,
+        session_id="s3",
+        force=True,
+    )
+    assert invite2 is not None
+    dismiss(
+        "type",
+        contribution_type="FALSE_POSITIVE_FIX",
+        privacy_path=privacy_path,
+        state_path=contribute_state_path,
+    )
+    clear_contribute_state(contribute_state_path)
+    suppressed = suggest(
+        ctx,
+        privacy_path=privacy_path,
+        state_path=contribute_state_path,
+        session_id="s4",
+        force=True,
+    )
+    assert suppressed is None
+
+    dismiss("never_prompts", privacy_path=privacy_path, state_path=contribute_state_path)
+    never = suggest(
+        {"verified_count": 2, "novel_finding": True, "test_gap": True},
+        privacy_path=privacy_path,
+        state_path=contribute_state_path,
+        force=False,
+    )
+    assert never is None
+
+
+# ---------------------------------------------------------------------------
+# Privacy / learning opt-in
+# ---------------------------------------------------------------------------
+
+
+def test_privacy_opt_in_required_for_pack_and_learning(privacy_path: Path, tmp_path: Path):
+    assert is_contribute_opted_in(privacy_path) is False
+    blocked = prepare(
+        {"fp_rejected": 2, "snippet": "x = 1"},
+        project_root=tmp_path,
+        privacy_path=privacy_path,
+        require_opt_in=True,
+    )
+    assert blocked["ok"] is False
+    assert blocked["error"] == "contribute_opt_in_required"
+
+    opt_in(contribute=True, learning=False, path=privacy_path)
+    assert is_contribute_opted_in(privacy_path) is True
+    assert is_learning_opted_in(privacy_path) is False
+
+    learning_blocked = prepare(
+        {"fp_rejected": 2, "snippet": "ok"},
+        project_root=tmp_path,
+        privacy_path=privacy_path,
+        write_learning_copy=True,
+    )
+    assert learning_blocked["ok"] is False
+    assert learning_blocked["error"] == "learning_opt_in_required"
+
+    opt_in(contribute=True, learning=True, path=privacy_path)
+    assert is_learning_opted_in(privacy_path) is True
+    ok = prepare(
+        {"fp_rejected": 2, "snippet": "ok"},
+        contribution_type="FALSE_POSITIVE_FIX",
+        project_root=tmp_path,
+        privacy_path=privacy_path,
+        write_learning_copy=True,
+    )
+    assert ok["ok"] is True
+    assert ok["learning_copy"]
+    assert Path(ok["learning_copy"]).is_file()
+
+
+def test_delete_reset_clears_learning_data(privacy_path: Path, tmp_path: Path):
+    # Point learning dir under tmp by placing privacy.json under tmp_path
+    opt_in(contribute=True, learning=True, path=privacy_path)
+    learn = learning_data_dir(privacy_path)
+    learn.mkdir(parents=True, exist_ok=True)
+    sample = learn / "sample.json"
+    sample.write_text("{}\n", encoding="utf-8")
+    assert sample.is_file()
+
+    assert delete_learning_data(privacy_path) is True
+    assert not learn.is_dir()
+
+    # recreate and reset
+    opt_in(contribute=True, learning=True, path=privacy_path)
+    learn.mkdir(parents=True, exist_ok=True)
+    (learn / "x.json").write_text("{}\n", encoding="utf-8")
+    st = privacy_reset(path=privacy_path)
+    assert st["contribute_opt_in"] is False
+    assert st["learning"]["contribution_data"]["enabled"] is False
+    assert not learning_data_dir(privacy_path).is_dir()
+
+
+# ---------------------------------------------------------------------------
+# Sanitize
+# ---------------------------------------------------------------------------
+
+
+def test_secrets_scrubbed_and_private_repo_minimized():
+    text, hits = sanitize_text(
+        "token=sk-abcdefghijklmnopqrstuvwxyz012345 "
+        "path=/Users/alice/secret-project/app.py "
+        "see github.com/acme-corp/private-app",
+        known_private_names=["private-app", "acme-corp"],
+    )
+    assert "sk-" not in text or "REDACTED" in text
+    assert "/Users/alice" not in text
+    assert "REDACTED_PATH" in text or "REDACTED_DIR" in text
+    assert "private-app" not in text.lower() or "REDACTED" in text
+    assert hits or "REDACTED" in text
+
+    cleaned, _ = sanitize_example(
+        {
+            "code": "password = 'hunter2'\napi_key=sk-abcdefghijklmnopqrstuvwxyz012345",
+            "repo": "acme-corp/private-app",
+            "file": "/Users/alice/work/private-app/main.py",
+            "author_email": "alice@example.com",
+        },
+        known_private_names=["acme-corp", "private-app"],
+    )
+    assert "author_email" not in cleaned
+    blob = json.dumps(cleaned)
+    assert "hunter2" not in blob
+    assert "sk-abcdef" not in blob
+    assert "/Users/alice" not in blob
+
+
+# ---------------------------------------------------------------------------
+# Prepare local only
+# ---------------------------------------------------------------------------
+
+
+def test_prepare_creates_files_locally_only(privacy_path: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    opt_in(contribute=True, path=privacy_path)
+
+    opened: list[tuple] = []
+
+    def _boom(*a, **k):
+        opened.append((a, k))
+        raise AssertionError("network forbidden")
+
+    monkeypatch.setattr(socket, "socket", _boom)
+    monkeypatch.setattr(socket, "create_connection", _boom)
+
+    result = prepare(
+        {
+            "fp_rejected": 2,
+            "snippet": "if user.is_admin: return",
+            "file": "app/views.py",
+            "repo": "acme/secret",
+        },
+        contribution_type="FALSE_POSITIVE_FIX",
+        project_root=tmp_path,
+        privacy_path=privacy_path,
+    )
+    assert result["ok"] is True
+    assert result["network"] is False
+    assert result["pushed"] is False
+    assert result["pr_opened"] is False
+    assert opened == []
+
+    pkg = Path(result["package_dir"])
+    assert pkg.is_dir()
+    assert (pkg / "contribution.json").is_file()
+    assert (pkg / "LOCAL_ONLY.txt").is_file()
+    assert (pkg / "COMMIT_MSG.txt").is_file()
+    assert (pkg / "PR_DRAFT.md").is_file()
+    assert str(pkg).startswith(str(tmp_path / ".findings" / "axguard" / "contribute"))
+    assert result["id"]
+
+
+def test_prepare_rejects_path_traversal_package_id(privacy_path: Path, tmp_path: Path):
+    opt_in(contribute=True, path=privacy_path)
+    bad = prepare(
+        {"fp_rejected": 1},
+        contribution_type="DOCUMENTATION",
+        project_root=tmp_path,
+        privacy_path=privacy_path,
+        package_id="../escape",
+    )
+    assert bad["ok"] is False
+    assert bad["error"] == "invalid_package_id"
+    assert not (tmp_path / "escape").exists()
+    assert not list((tmp_path / ".findings").rglob("*")) if not (tmp_path / ".findings").exists() else True
+
+
+def test_suggest_prepare_do_not_open_sockets(privacy_path: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    opt_in(contribute=True, path=privacy_path)
+    state_path = tmp_path / "cstate.json"
+    calls: list[str] = []
+
+    def _boom(*_a, **_k):
+        calls.append("socket")
+        raise AssertionError("socket forbidden")
+
+    monkeypatch.setattr(socket, "socket", _boom)
+    monkeypatch.setattr(socket, "create_connection", _boom)
+
+    invite = suggest(
+        {"attack_paths_confirmed": 2},
+        privacy_path=privacy_path,
+        state_path=state_path,
+        session_id="sock",
+        force=True,
+    )
+    assert invite is not None
+    prepare(
+        {"attack_paths_confirmed": 2},
+        contribution_type="ATTACK_PATH_PATTERN",
+        project_root=tmp_path,
+        privacy_path=privacy_path,
+    )
+    assert calls == []
+
+
+# ---------------------------------------------------------------------------
+# No manipulative copy
+# ---------------------------------------------------------------------------
+
+
+def test_no_manipulative_copy_in_catalogs():
+    for mid, msg in {**MESSAGE_CATALOG, **CONTRIBUTE_MESSAGE_CATALOG}.items():
+        reason = reject_if_manipulative(msg)
+        assert reason is None, f"{mid}: {reason}"
+        text = "\n".join(msg.body_lines).lower()
+        for phrase in FORBIDDEN_CONTRIBUTE_PHRASES:
+            assert phrase not in text, f"{mid} contains forbidden phrase {phrase!r}"
+        assert not contains_banned_pattern(text)
+
+
+def test_quality_gate_filters_weak_noise():
+    weak = signals_from_context({})
+    assert filter_strong(weak, {}) == []
+    strong = filter_strong(signals_from_context({"fp_rejected": 4, "security_primary": True}), {"fp_rejected": 4})
+    assert strong
+    assert strong[0][0].contribution_type == "FALSE_POSITIVE_FIX"
+
+
+# ---------------------------------------------------------------------------
+# AST: no network imports in contributors package
+# ---------------------------------------------------------------------------
+
+
+def test_contributors_modules_have_no_network_imports():
+    forbidden = {"urllib", "urllib.request", "requests", "httpx", "aiohttp"}
+    offenders: list[str] = []
+    for py in sorted(CONTRIB_PKG.glob("*.py")):
+        tree = ast.parse(py.read_text(encoding="utf-8"), filename=str(py))
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    root = alias.name.split(".")[0]
+                    if alias.name in forbidden or root in ("urllib", "requests", "httpx", "aiohttp"):
+                        offenders.append(f"{py.name}: import {alias.name}")
+            elif isinstance(node, ast.ImportFrom) and node.module:
+                root = node.module.split(".")[0]
+                if node.module in forbidden or root in ("urllib", "requests", "httpx", "aiohttp"):
+                    offenders.append(f"{py.name}: from {node.module}")
+    assert not offenders, offenders
+
+
+def test_privacy_status_explains_fields(privacy_path: Path):
+    st = privacy_status(privacy_path)
+    assert "included" in st["fields"]
+    assert "excluded" in st["fields"]
+    assert "author_email" in st["fields"]["excluded"]
+    assert st["contributions"]["auto_push"] is False
+    assert st["contributions"]["auto_pr"] is False
+    assert st["learning"]["contribution_data"]["enabled"] is False

From 4507329a46cc0f8cd76825e7520ac4a18c64df34 Mon Sep 17 00:00:00 2001
From: Md Shariar Shanaz Shuvon <83355567+shuvonsec@users.noreply.github.com>
Date: Wed, 16 Sep 2026 02:50:52 +0800
Subject: [PATCH 2/3] docs(commands): add /axguard-privacy and
 /axguard-contribute

---
 COMMANDS-QUICK-REF.md          |  3 ++-
 commands/axguard-contribute.md | 43 ++++++++++++++++++++++++++++++++++
 commands/axguard-privacy.md    | 42 +++++++++++++++++++++++++++++++++
 3 files changed, 87 insertions(+), 1 deletion(-)
 create mode 100644 commands/axguard-contribute.md
 create mode 100644 commands/axguard-privacy.md

diff --git a/COMMANDS-QUICK-REF.md b/COMMANDS-QUICK-REF.md
index ca57c13..66fc627 100644
--- a/COMMANDS-QUICK-REF.md
+++ b/COMMANDS-QUICK-REF.md
@@ -39,7 +39,7 @@
 | Regenerate reports | `/axguard-report` |
 | Add CI gate | `/axguard-ci` |
 | About / engagement prefs | `axguard about` · `axguard engage disable` |
-| Contribute / privacy (local) | `axguard contribute …` · `axguard privacy …` |
+| Contribute / privacy (local) | `/axguard-contribute` · `/axguard-privacy` |
 
 ## Default pipeline
 
@@ -94,6 +94,7 @@ axguard about
 axguard engage disable
 axguard privacy status
 axguard contribute status
+# slash: /axguard-privacy · /axguard-contribute
 axguard data discover
 axguard data report fixtures/data_pipeline
 axguard twin build .   # Security Twin — see docs/twin/README.md
diff --git a/commands/axguard-contribute.md b/commands/axguard-contribute.md
new file mode 100644
index 0000000..83870e6
--- /dev/null
+++ b/commands/axguard-contribute.md
@@ -0,0 +1,43 @@
+---
+description: Contextual contribution invites and local prepare (no auto-push). Usage: /axguard-contribute …
+---
+
+# /axguard-contribute
+
+**Specialist:** Contributor Engagement & Learning
+
+Recognizes meaningful local work, suggests a contribution type when quality is
+strong, and prepares a **local** package (templates, commit/PR draft text).
+Never pushes or opens a PR. No dark patterns — cooldown and dismiss controls.
+
+## Usage
+
+```
+/axguard-contribute status
+/axguard-contribute suggest [--context-json FILE] [--force]
+/axguard-contribute prepare [--type TYPE] [--context-json FILE] [--learning]
+/axguard-contribute dismiss not_now|never_prompts|type [--type TYPE]
+/axguard-contribute milestones
+/axguard-contribute templates
+```
+
+CLI equivalent: `axguard contribute `.
+
+## Focus
+
+- Opportunity score from observable work (not personal worth)
+- Quality gate: novelty, reusability, relevance, testability
+- Cooldown: at most one invite per meaningful session
+- Dismiss: `not_now` / `never_prompts` / type
+- Prepare writes under `.findings/axguard/contribute//`
+- Scrubbing via `engines.data.scrub`; learning copy needs `--learning` + opt-in
+
+## Steps
+
+1. `axguard privacy opt-in` (required before prepare)
+2. `axguard contribute suggest` after a useful audit/adversary session
+3. Review the invite; dismiss if not wanted
+4. `axguard contribute prepare` — inspect the package locally
+5. Share / push / PR only under your own control (GitHub OAuth is future work)
+
+See [docs/contributors/README.md](../docs/contributors/README.md).
diff --git a/commands/axguard-privacy.md b/commands/axguard-privacy.md
new file mode 100644
index 0000000..b50f0d0
--- /dev/null
+++ b/commands/axguard-privacy.md
@@ -0,0 +1,42 @@
+---
+description: Local privacy prefs for contribution learning (opt-in). Usage: /axguard-privacy …
+---
+
+# /axguard-privacy
+
+**Specialist:** Contributor privacy center
+
+Shows what AXGuard stores for contribution learning, what stays local, and how
+to opt in/out, export, or delete. Defaults are off — no network, no telemetry.
+
+## Usage
+
+```
+/axguard-privacy status
+/axguard-privacy show
+/axguard-privacy opt-in
+/axguard-privacy opt-in --learning
+/axguard-privacy opt-out
+/axguard-privacy export -o .findings/axguard/privacy-export.json
+/axguard-privacy delete
+/axguard-privacy reset
+```
+
+CLI equivalent: `axguard privacy `.
+
+## Focus
+
+- Prefs live at `~/.axguard/privacy.json`
+- Packaging and learning require explicit opt-in
+- Included vs excluded field lists (no credentials, tokens, private repo IDs)
+- Export / delete / reset for local learning data only
+- GitHub push/PR is never automatic
+
+## Steps
+
+1. `axguard privacy status` — current opt-in and learning flags
+2. `axguard privacy show` — included/excluded fields and notes
+3. Opt in only when you intend to prepare a local contribution package
+4. `axguard privacy delete` or `reset` to clear local learning copies
+
+See [docs/contributors/README.md](../docs/contributors/README.md).

From 6a0c8d0d26674885dd01286d1d3b280f58394880 Mon Sep 17 00:00:00 2001
From: Md Shariar Shanaz Shuvon <83355567+shuvonsec@users.noreply.github.com>
Date: Wed, 16 Sep 2026 03:17:11 +0800
Subject: [PATCH 3/3] fix(ci): make axguard review dogfood pass on shipped code

---
 .github/workflows/README.md   |  2 +-
 .github/workflows/axguard.yml |  6 +++++-
 engines/api/storage.py        | 14 ++++++--------
 rules/path.json               |  2 +-
 4 files changed, 13 insertions(+), 11 deletions(-)

diff --git a/.github/workflows/README.md b/.github/workflows/README.md
index 4f353b2..eff4444 100644
--- a/.github/workflows/README.md
+++ b/.github/workflows/README.md
@@ -5,7 +5,7 @@ CI and repository security workflows for this project.
 | Workflow | Purpose |
 |---|---|
 | `ci.yml` | Unit tests, skill validation, fixture self-scan |
-| `axguard.yml` | AXGuard Security Review on PRs / main (`audit --fail-on high`) |
+| `axguard.yml` | AXGuard Security Review on PRs / main (`audit engines` + `cli` `--fail-on high`; fixtures excluded) |
 | `codeql.yml` | CodeQL static analysis (Python) |
 | `gitleaks.yml` | Secret detection (Gitleaks CLI) |
 | `osv-scanner.yml` | Dependency vulns via OSV |
diff --git a/.github/workflows/axguard.yml b/.github/workflows/axguard.yml
index 690ad41..85590f4 100644
--- a/.github/workflows/axguard.yml
+++ b/.github/workflows/axguard.yml
@@ -31,5 +31,9 @@ jobs:
       - name: Install AXGuard
         run: python -m pip install -e .
 
+      # Dogfood shipped runtime code. fixtures/, skills/, commands/, and rules/
+      # intentionally contain vulnerable examples for regression tests.
       - name: Security review (fail on high+)
-        run: axguard audit . --fail-on high --no-banner
+        run: |
+          axguard audit engines --fail-on high --no-banner
+          axguard audit cli --fail-on high --no-banner
diff --git a/engines/api/storage.py b/engines/api/storage.py
index 298ca38..c9e8166 100644
--- a/engines/api/storage.py
+++ b/engines/api/storage.py
@@ -291,11 +291,10 @@ def update_project(self, project_id: str, **fields: Any) -> dict[str, Any] | Non
         if not updates:
             return self.get_project(project_id)
         updates["updated_at"] = time.time()
+        # Column names are allowlisted above; values stay bound parameters.
         sets = ", ".join(f"{k} = ?" for k in updates)
-        self._execute(
-            f"UPDATE projects SET {sets} WHERE id = ?",
-            tuple(updates.values()) + (project_id,),
-        )
+        sql = "UPDATE projects SET " + sets + " WHERE id = ?"
+        self._execute(sql, tuple(updates.values()) + (project_id,))
         return self.get_project(project_id)
 
     def delete_project(self, project_id: str) -> bool:
@@ -418,11 +417,10 @@ def update_scan(self, scan_id: str, **fields: Any) -> dict[str, Any] | None:
                 updates[k] = v
         if not updates:
             return self.get_scan(scan_id)
+        # Column names are allowlisted above; values stay bound parameters.
         sets = ", ".join(f"{k} = ?" for k in updates)
-        self._execute(
-            f"UPDATE scans SET {sets} WHERE id = ?",
-            tuple(updates.values()) + (scan_id,),
-        )
+        sql = "UPDATE scans SET " + sets + " WHERE id = ?"
+        self._execute(sql, tuple(updates.values()) + (scan_id,))
         return self.get_scan(scan_id)
 
     def cancel_scan(self, scan_id: str) -> dict[str, Any] | None:
diff --git a/rules/path.json b/rules/path.json
index ccef483..ec792e4 100644
--- a/rules/path.json
+++ b/rules/path.json
@@ -6,7 +6,7 @@
       "severity": "high",
       "cwe": "CWE-22",
       "languages": ["python"],
-      "pattern": "open\\s*\\(\\s*(os\\.path\\.join|pathlib\\.|Path\\()[^\\n]*\\+|open\\s*\\(\\s*[a-zA-Z_][\\w\\.]*\\s*[,)]",
+      "pattern": "\\bopen\\s*\\(\\s*(os\\.path\\.join|pathlib\\.|Path\\()[^\\n]*\\+|\\bopen\\s*\\(\\s*[a-zA-Z_][\\w\\.]*\\s*[,)]",
       "message": "Opening paths built from variables can enable path traversal / LFI.",
       "fix": "Resolve under a fixed root and reject .. segments; use allowlists."
     },