feat(2w): W2 WC7 trigger surface — bridge parses !testme --quick

bridge/bridge.py: parse_trigger(body) → (is_trigger, quick); accepts exactly
'!testme' (cold, default) and '!testme --quick' (opt-in fast lane), rejects
'!testmexyz'/'!testme foo'/etc. Threaded through both poll + webhook paths and
process_testme → trigger_build adds the CCCI_QUICK=1 Drone param (auto-exposed
to run_recipe_ci). PR comment labels a quick run lower-confidence. .drone.yml
echoes quick=. +3 unit tests (incl. the !testmexyz negative). 64 unit pass.
WC7: default !testme stays full cold; --quick opt-in, never gates merge.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.drone.yml

@@ -64,6 +64,8 @@ steps:
       # means no concurrent build shares /root/.abra.
       HOME: /root
     commands:
-      # RECIPE/REF/PR/SRC are injected as env vars from the build's custom params.
-      - 'echo "recipe-ci: RECIPE=$RECIPE REF=$REF PR=$PR SRC=$SRC stages=$STAGES"'
+      # RECIPE/REF/PR/SRC (+ CCCI_QUICK for `!testme --quick`) are injected as env vars from the
+      # build's custom params. CCCI_QUICK=1 makes run_recipe_ci take the opt-in fast lane (WC7);
+      # absent => full cold (default). run_quick ignores STAGES (always upgrade+custom).
+      - 'echo "recipe-ci: RECIPE=$RECIPE REF=$REF PR=$PR SRC=$SRC stages=$STAGES quick=${CCCI_QUICK:-0}"'
       - cc-ci-run runner/run_recipe_ci.py

bridge/bridge.py

@@ -43,6 +43,19 @@ GITEA_API = os.environ.get("GITEA_API", "https://git.autonomic.zone/api/v1")
 DRONE_URL = os.environ.get("DRONE_URL", "https://drone.ci.commoninternet.net")
 CI_REPO = os.environ.get("CI_REPO", "recipe-maintainers/cc-ci")
 TRIGGER = "!testme"
+
+
+def parse_trigger(body):
+    """Parse a PR comment body into (is_trigger, quick). Exactly two accepted forms (trimmed):
+    `!testme` → (True, False) = full COLD run (default, authoritative);
+    `!testme --quick` → (True, True) = opt-in LOWER-CONFIDENCE fast lane (WC4/WC7).
+    Anything else (`!testmexyz`, `!testme foo`, prose) → (False, False) — must NOT trigger."""
+    s = (body or "").strip()
+    if s == TRIGGER:
+        return True, False
+    if s == f"{TRIGGER} --quick":
+        return True, True
+    return False, False
 ALLOWLIST = {u.strip() for u in os.environ.get("AUTH_ALLOWLIST", "").split(",") if u.strip()}
 
 
@@ -105,11 +118,13 @@ def pr_head(owner, repo, number):
     return {"sha": head.get("sha"), "repo": (head.get("repo") or {}).get("full_name")}
 
 
-def trigger_build(recipe, ref, pr, src):
-    # Drone "create build" with custom params -> exposed to the pipeline as env vars.
-    q = urllib.parse.urlencode(
-        {"branch": "main", "RECIPE": recipe, "REF": ref, "PR": str(pr), "SRC": src}
-    )
+def trigger_build(recipe, ref, pr, src, quick=False):
+    # Drone "create build" with custom params -> exposed to the pipeline as env vars. `--quick`
+    # (WC7) sets CCCI_QUICK=1 so run_recipe_ci takes the opt-in fast lane; absent => full cold.
+    params = {"branch": "main", "RECIPE": recipe, "REF": ref, "PR": str(pr), "SRC": src}
+    if quick:
+        params["CCCI_QUICK"] = "1"
+    q = urllib.parse.urlencode(params)
     url = f"{DRONE_URL}/api/repos/{CI_REPO}/builds?{q}"
     status, build = _api(url, DRONE_TOKEN, method="POST", scheme="Bearer")
     if status in (200, 201) and build:
@@ -190,7 +205,7 @@ def _claim(comment_id) -> bool:
         return True
 
 
-def process_testme(full_name, owner, name, number, user, comment_id, source):
+def process_testme(full_name, owner, name, number, user, comment_id, source, quick=False):
     """Shared by both paths. Dedupes by comment id, checks authorization, resolves the PR head,
     triggers the build, comments the run link. Returns (run_url|None, reason)."""
     if not _claim(comment_id):
@@ -201,13 +216,15 @@ def process_testme(full_name, owner, name, number, user, comment_id, source):
     head = pr_head(owner, name, number)
     if not head or not head["sha"]:
         return None, "cannot resolve PR head"
-    num = trigger_build(name, head["sha"], number, head["repo"] or full_name)
+    num = trigger_build(name, head["sha"], number, head["repo"] or full_name, quick=quick)
     if not num:
         post_comment(owner, name, number, "cc-ci: failed to start a CI run (see bridge logs).")
         return None, "trigger failed"
     run_url = f"{DRONE_URL}/{CI_REPO}/{num}"
+    mode = " **(--quick: lower-confidence fast lane; does not gate merge)**" if quick else ""
     cid = post_comment(
-        owner, name, number, f"cc-ci: started CI run for `{name}` @ `{head['sha'][:8]}` → {run_url}"
+        owner, name, number,
+        f"cc-ci: started CI run for `{name}` @ `{head['sha'][:8]}`{mode} → {run_url}",
     )
     log(
         f"[{source}] triggered build {num} for {name}@{head['sha'][:8]} "
@@ -255,7 +272,8 @@ class Handler(BaseHTTPRequestHandler):
         c = payload.get("comment") or {}
         issue = payload.get("issue") or {}
         repo = payload.get("repository") or {}
-        if action != "created" or (c.get("body") or "").strip() != TRIGGER:
+        is_trigger, quick = parse_trigger(c.get("body"))
+        if action != "created" or not is_trigger:
             return self._send(204, "ignored")
         if not issue.get("pull_request"):
             return self._send(204, "not a PR")
@@ -268,6 +286,7 @@ class Handler(BaseHTTPRequestHandler):
             c.get("user", {}).get("login", ""),
             c.get("id"),
             "webhook",
+            quick=quick,
         )
         if not run_url:
             if reason == "duplicate":
@@ -292,14 +311,15 @@ def poll_loop():
             for pr in list_open_prs(full_name):
                 number = pr.get("number")
                 for c in list_comments(full_name, number):
-                    if (c.get("body") or "").strip() != TRIGGER:
+                    is_trigger, quick = parse_trigger(c.get("body"))
+                    if not is_trigger:
                         continue
                     cid = c.get("id")
                     if first:
                         _claim(cid)  # mark pre-existing comments seen; don't fire on startup
                         continue
                     user = (c.get("user") or {}).get("login", "")
-                    process_testme(full_name, owner, name, number, user, cid, "poll")
+                    process_testme(full_name, owner, name, number, user, cid, "poll", quick=quick)
         first = False
         time.sleep(interval)
 

tests/unit/test_bridge_trigger.py

@@ -0,0 +1,41 @@
+"""Unit tests for the bridge's `!testme` / `!testme --quick` trigger parser (WC7 + D1).
+
+Pure: imports bridge/bridge.py (stdlib-only, no side effects at import — main() is __main__-guarded).
+Locks the two accepted forms and the must-NOT-trigger cases the Adversary probes (`!testmexyz`, etc.).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+
+# bridge.py reads HMAC/DRONE/GITEA secret FILES at import; point them at /dev/null (readable, empty)
+# so the import works in a unit context — parse_trigger doesn't use any of them.
+for _v in ("HMAC_FILE", "DRONE_TOKEN_FILE", "GITEA_TOKEN_FILE"):
+    os.environ.setdefault(_v, "/dev/null")
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "bridge"))
+import bridge  # noqa: E402
+
+
+def test_plain_testme_is_cold():
+    assert bridge.parse_trigger("!testme") == (True, False)
+    assert bridge.parse_trigger("  !testme  ") == (True, False)  # trimmed
+
+
+def test_quick_form():
+    assert bridge.parse_trigger("!testme --quick") == (True, True)
+    assert bridge.parse_trigger("  !testme --quick \n") == (True, True)
+
+
+def test_non_trigger_forms_rejected():
+    for body in (
+        "!testmexyz",        # the Adversary's classic negative
+        "!testme xyz",
+        "!testme--quick",    # no space → not the quick form
+        "!testme  --quick",  # double space → not an exact match (conservative)
+        "please !testme",
+        "testme",
+        "",
+        None,
+    ):
+        assert bridge.parse_trigger(body) == (False, False), body