feat: enable guarded webhook canary ingestion

dashboard/src/main.tsx

@@ -1770,7 +1770,7 @@ function WebhookPage({
   ];
   return (
     <section className="grid gap-4">
-      <PageTitle icon={<Activity className="h-5 w-5 text-muted" aria-hidden />} title="GitHub webhooks" subtitle="Shadow ingestion health. Deliveries are observed but do not create jobs." action={<RefreshButton onClick={onRefresh} />} />
+      <PageTitle icon={<Activity className="h-5 w-5 text-muted" aria-hidden />} title="GitHub webhooks" subtitle={summary?.mode === "canary" ? "Canary ingestion health. Only enabled repositories may create jobs." : "Shadow ingestion health. Deliveries are observed but do not create jobs."} action={<RefreshButton onClick={onRefresh} />} />
       {error ? <Banner tone="error" text={error.message} /> : null}
       <div className="flex max-w-full flex-wrap rounded-md border border-border bg-white p-1" role="tablist" aria-label="Webhook dashboard section">
         {sections.map((item) => <button key={item.id} type="button" role="tab" aria-label={item.count === undefined ? item.label : `${item.label} (${item.count} total)`} aria-selected={section === item.id} className={cn("inline-flex h-8 items-center gap-2 rounded px-3 text-sm font-semibold", section === item.id ? "bg-primary text-white" : "text-muted hover:bg-slate-50 hover:text-foreground")} onClick={() => onSectionChange(item.id)}>{item.label}{item.count !== undefined ? <span className="rounded border border-current/30 px-1 font-mono text-[10px]">{item.count}</span> : null}</button>)}

docs/ingestion.md

@@ -128,5 +128,24 @@ Comment and review `edited` deliveries are observed under a distinct key and
 do not retrigger work. Phase 2 must make an explicit policy decision before
 any non-`created` action can enqueue a job.
 
+## Canary dual ingestion
+
+Set `GITHUB_AGENT_BRIDGE_WEBHOOK_MODE=canary` and point
+`GITHUB_AGENT_BRIDGE_WEBHOOK_POLICY` at the reader/executor policy file. Canary
+mode converts only supported actionable deliveries into the common queue and
+requires their repository to be explicitly listed in `enabledRepos`; an empty
+allowlist enqueues nothing. IMAP continues unchanged. Both transports use the
+same canonical event key, so the first committed receipt wins and the second is
+recorded as a duplicate of the same job.
+
+For enqueueing, the common queue transaction commits before the monitoring
+receipt. A crash in that narrow gap cannot lose work: GitHub retries the
+delivery, the durable `ingest_receipts(source='webhook', source_key=<delivery>)`
+row makes the queue operation idempotent, and the retry repairs the monitoring
+receipt. `edited` comments/reviews, unsupported families, and repositories
+outside `enabledRepos` remain observational only. For `workflow_run.completed`,
+only runs with `conclusion: failure` enqueue work; successful and other
+conclusions remain observational.
+
 Webhook enqueueing must not be enabled until recovery of persisted-but-
 unprocessed receipts and divergence metrics have been validated in production.

src/github_agent_bridge/backend.py

@@ -55,9 +55,10 @@
 from .mcp import MCPServer, authenticate_token, create_token, list_tokens, revoke_token, update_token_owner
 from .observability import configure_sentry, list_alerts, recent_process_samples
 from .queue import JobQueue
+from .policy import Policy
 from .systemd_status import allowed_unit_names, stream_journal_lines, systemd_status
 from .web_push import delete_subscription, save_subscription, subscription_status
-from .webhook import persist_shadow_delivery, verify_signature
+from .webhook import persist_shadow_delivery, verify_signature, webhook_notification
 
 
 DEFAULT_HOST = os.getenv("GITHUB_AGENT_BRIDGE_DASHBOARD_HOST", "127.0.0.1")
@@ -236,6 +237,8 @@ def __init__(
         webhook_secrets_by_owner: dict[str, tuple[str, ...]] | None = None,
         webhook_max_bytes: int | None = None,
         webhook_retention_days: int | None = None,
+        webhook_mode: str | None = None,
+        webhook_policy: str | Path | None = None,
     ) -> None:
         self.db = Path(db).expanduser()
         self.secret_key = secret_key or os.getenv("GITHUB_AGENT_BRIDGE_DASHBOARD_SECRET_KEY", "")
@@ -259,6 +262,13 @@ def __init__(
         self.webhook_secrets_by_owner = webhook_secrets_by_owner if webhook_secrets_by_owner is not None else _webhook_secrets_by_owner_env()
         self.webhook_max_bytes = webhook_max_bytes or int(os.getenv("GITHUB_AGENT_BRIDGE_WEBHOOK_MAX_BYTES", "1048576"))
         self.webhook_retention_days = webhook_retention_days or int(os.getenv("GITHUB_AGENT_BRIDGE_WEBHOOK_RETENTION_DAYS", "30"))
+        self.webhook_mode = (webhook_mode or os.getenv("GITHUB_AGENT_BRIDGE_WEBHOOK_MODE", "shadow")).lower()
+        if self.webhook_mode not in {"shadow", "canary"}:
+            raise ValueError("GITHUB_AGENT_BRIDGE_WEBHOOK_MODE must be shadow or canary")
+        policy_value = webhook_policy or os.getenv("GITHUB_AGENT_BRIDGE_WEBHOOK_POLICY", "")
+        self.webhook_policy = Path(policy_value).expanduser() if policy_value else None
+        if self.webhook_mode == "canary" and self.webhook_policy is None:
+            raise ValueError("GITHUB_AGENT_BRIDGE_WEBHOOK_POLICY is required in canary mode")
 
     @property
     def oauth_ready(self) -> bool:
@@ -766,6 +776,21 @@ async def github_webhook_shadow(request: Request) -> dict[str, Any]:
         if not verify_signature(raw_payload, signature, webhook_secrets):
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_signature")
         ensure_webhook_schema()
+        enqueue_status = None
+        job_id = None
+        if config.webhook_mode == "canary":
+            policy = Policy.from_file(config.webhook_policy)
+            notification = webhook_notification(event_name, delivery_id, payload)
+            repo = str(full_name or "").lower()
+            if notification is None:
+                enqueue_status = "ignored"
+            elif repo not in policy.enabled_repos:
+                enqueue_status = "outside_canary"
+            else:
+                job, enqueue_status = JobQueue(config.db).ingest(
+                    notification, policy, source="webhook", source_key=delivery_id,
+                )
+                job_id = job.id if job else None
         receipt = persist_shadow_delivery(
             config.db,
             delivery_id=delivery_id,
@@ -774,7 +799,14 @@ async def github_webhook_shadow(request: Request) -> dict[str, Any]:
             hook_id=hook_id,
             retention_days=config.webhook_retention_days,
         )
-        return {"mode": "shadow", "status": receipt.status, "event_key": receipt.event_key}
+        response = {
+            "mode": config.webhook_mode,
+            "status": receipt.status,
+            "event_key": receipt.event_key,
+        }
+        if config.webhook_mode != "shadow":
+            response.update({"enqueue_status": enqueue_status, "job_id": job_id})
+        return response
 
     @app.get("/api/webhooks/github/status")
     @app.get("/api/webhooks/github/summary")
@@ -813,7 +845,7 @@ def github_webhook_shadow_summary(_: dict[str, Any] = Depends(current_admin_prof
             if count
         }
         return {
-            "mode": "shadow",
+            "mode": config.webhook_mode,
             "configured": bool(config.webhook_secrets or config.webhook_secrets_by_owner),
             "receipts": counts,
             "duplicate_deliveries": row[2],

src/github_agent_bridge/dashboard_static/index.html

@@ -4,7 +4,7 @@
     <meta charset="UTF-8" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>GitHub Agent Bridge Dashboard</title>
-    <script type="module" crossorigin src="/assets/index-QvZxB06B.js"></script>
+    <script type="module" crossorigin src="/assets/index-DzCW97RR.js"></script>
     <link rel="modulepreload" crossorigin href="/assets/charts-SqBiqy9C.js">
     <link rel="stylesheet" crossorigin href="/assets/index-BDe9F9HC.css">
   </head>

src/github_agent_bridge/webhook.py

@@ -8,7 +8,7 @@
 from pathlib import Path
 from typing import Any
 
-from .models import utc_now
+from .models import Notification, utc_now
 
 
 @dataclass(frozen=True)
@@ -60,6 +60,58 @@ def canonical_webhook_event_key(event_name: str, payload: dict[str, Any]) -> str
     return None
 
 
+def webhook_notification(
+    event_name: str,
+    delivery_id: str,
+    payload: dict[str, Any],
+) -> Notification | None:
+    """Translate actionable webhook payloads into the transport-neutral queue input."""
+    action = str(payload.get("action") or "")
+    if (event_name, action) not in {
+        ("issue_comment", "created"),
+        ("pull_request_review_comment", "created"),
+        ("pull_request_review", "submitted"),
+        ("commit_comment", "created"),
+        ("workflow_run", "completed"),
+    }:
+        return None
+    repository = payload.get("repository") if isinstance(payload.get("repository"), dict) else {}
+    repo = str(repository.get("full_name") or "")
+    if not repo:
+        return None
+    subject = payload.get("pull_request") or payload.get("issue") or payload.get("workflow_run") or {}
+    number = subject.get("number") if isinstance(subject, dict) else None
+    source = (
+        payload.get("comment") or payload.get("review") or payload.get("workflow_run") or {}
+    )
+    if not isinstance(source, dict):
+        return None
+    if event_name == "workflow_run":
+        conclusion = str(source.get("conclusion") or "").lower()
+        if conclusion != "failure":
+            return None
+    url = str(source.get("html_url") or subject.get("html_url") or repository.get("html_url") or "")
+    if not url.startswith("https://github.com/"):
+        return None
+    body = (
+        "Workflow run failed (conclusion: failure)."
+        if event_name == "workflow_run"
+        else str(source.get("body") or "")
+    )
+    sender = payload.get("sender") if isinstance(payload.get("sender"), dict) else {}
+    login = str(sender.get("login") or "GitHub")
+    title = str(subject.get("title") or subject.get("name") or event_name)
+    suffix = f" (#{number})" if number else ""
+    return Notification(
+        uid=None,
+        message_id=f"<{delivery_id}@github.com>",
+        subject=f"[{repo}] {title}{suffix}",
+        from_addr=f"{login} <notifications@github.com>",
+        body=f"{body}\n\n{url}",
+        auth={"spf": True, "dkim": True, "dmarc": True},
+    )
+
+
 def persist_shadow_delivery(
     db: str | Path,
     *,

systemd/env.example

@@ -86,3 +86,7 @@ GITHUB_AGENT_BRIDGE_GITHUB_APP_ID=
 GITHUB_AGENT_BRIDGE_GITHUB_APP_SLUG=
 # Optional notification icon override.
 GITHUB_AGENT_BRIDGE_WEB_PUSH_ICON_URL=
+# Webhook ingestion is shadow-only by default. Canary mode additionally needs
+# the same policy used by the reader and only enqueues policy.enabledRepos.
+GITHUB_AGENT_BRIDGE_WEBHOOK_MODE=shadow
+GITHUB_AGENT_BRIDGE_WEBHOOK_POLICY=%h/.config/github-agent-bridge/policy.json

tests/test_webhook.py

@@ -37,6 +37,49 @@ def issue_comment_payload(*, action: str = "created") -> bytes:
     }).encode()
 
 
+def actionable_issue_comment_payload(*, comment_id: int = 5948901951) -> bytes:
+    return json.dumps({
+        "action": "created",
+        "repository": {"full_name": "gisce/github-agent-bridge"},
+        "issue": {
+            "number": 191,
+            "title": "Evaluate GitHub App webhooks",
+            "html_url": "https://github.com/gisce/github-agent-bridge/issues/191",
+        },
+        "comment": {
+            "id": comment_id,
+            "body": "@giscebot implement this",
+            "html_url": f"https://github.com/gisce/github-agent-bridge/issues/191#issuecomment-{comment_id}",
+        },
+        "sender": {"login": "ecarreras"},
+    }).encode()
+
+
+def workflow_run_payload(*, conclusion: str) -> bytes:
+    return json.dumps({
+        "action": "completed",
+        "repository": {"full_name": "gisce/github-agent-bridge"},
+        "workflow_run": {
+            "id": 33123456789,
+            "name": "pytest",
+            "conclusion": conclusion,
+            "html_url": "https://github.com/gisce/github-agent-bridge/actions/runs/33123456789",
+        },
+        "sender": {"login": "github-actions"},
+    }).encode()
+
+
+def canary_policy(tmp_path):
+    path = tmp_path / "policy.json"
+    path.write_text(json.dumps({
+        "trustedOrgs": ["gisce"],
+        "enabledRepos": ["gisce/github-agent-bridge"],
+        "botLogins": ["giscebot"],
+        "actions": {"trustedAuto": ["reply_comment", "workflow_run_failed"]},
+    }))
+    return path
+
+
 def test_webhook_shadow_verifies_and_persists_without_creating_job(tmp_path):
     db = tmp_path / "bridge.sqlite3"
     payload = issue_comment_payload()
@@ -96,6 +139,94 @@ def test_webhook_shadow_accepts_previous_rotation_secret(tmp_path):
     assert client.post("/api/webhooks/github", content=payload, headers=signed_headers(payload)).status_code == 200
 
 
+def test_webhook_canary_enqueues_enabled_actionable_repository_once(tmp_path):
+    payload = actionable_issue_comment_payload()
+    config = DashboardConfig(
+        db=tmp_path / "bridge.sqlite3",
+        require_auth=False,
+        webhook_secrets=(SECRET,),
+        webhook_mode="canary",
+        webhook_policy=canary_policy(tmp_path),
+    )
+    client = TestClient(create_app(config))
+
+    first = client.post("/api/webhooks/github", content=payload, headers=signed_headers(payload))
+    retry = client.post("/api/webhooks/github", content=payload, headers=signed_headers(payload))
+
+    assert first.json()["mode"] == "canary"
+    assert first.json()["enqueue_status"] == "enqueued"
+    assert first.json()["job_id"]
+    assert retry.json()["enqueue_status"] == "duplicate"
+    with sqlite3.connect(config.db) as con:
+        assert con.execute("SELECT COUNT(*) FROM jobs").fetchone()[0] == 1
+        assert con.execute(
+            "SELECT source,event_key,status FROM ingest_receipts"
+        ).fetchone() == (
+            "webhook",
+            "issue_comment:created:gisce/github-agent-bridge:5948901951",
+            "accepted",
+        )
+
+
+def test_webhook_canary_ignores_repo_outside_enabled_repos(tmp_path):
+    policy = canary_policy(tmp_path)
+    policy.write_text(json.dumps({"trustedOrgs": ["gisce"], "enabledRepos": ["gisce/other"]}))
+    payload = actionable_issue_comment_payload()
+    config = DashboardConfig(
+        db=tmp_path / "bridge.sqlite3", require_auth=False,
+        webhook_secrets=(SECRET,), webhook_mode="canary", webhook_policy=policy,
+    )
+
+    response = TestClient(create_app(config)).post(
+        "/api/webhooks/github", content=payload, headers=signed_headers(payload),
+    )
+
+    assert response.json()["enqueue_status"] == "outside_canary"
+    with sqlite3.connect(config.db) as con:
+        assert con.execute("SELECT COUNT(*) FROM jobs").fetchone()[0] == 0
+
+
+def test_webhook_canary_enqueues_failed_workflow_run(tmp_path):
+    payload = workflow_run_payload(conclusion="failure")
+    config = DashboardConfig(
+        db=tmp_path / "bridge.sqlite3", require_auth=False,
+        webhook_secrets=(SECRET,), webhook_mode="canary",
+        webhook_policy=canary_policy(tmp_path),
+    )
+
+    response = TestClient(create_app(config)).post(
+        "/api/webhooks/github",
+        content=payload,
+        headers=signed_headers(payload, delivery="workflow-failure", event="workflow_run"),
+    )
+
+    assert response.json()["enqueue_status"] == "enqueued"
+    with sqlite3.connect(config.db) as con:
+        assert con.execute("SELECT action,work_key FROM jobs").fetchone() == (
+            "workflow_run_failed",
+            "gisce/github-agent-bridge/actions/runs/33123456789",
+        )
+
+
+def test_webhook_canary_ignores_successful_workflow_run(tmp_path):
+    payload = workflow_run_payload(conclusion="success")
+    config = DashboardConfig(
+        db=tmp_path / "bridge.sqlite3", require_auth=False,
+        webhook_secrets=(SECRET,), webhook_mode="canary",
+        webhook_policy=canary_policy(tmp_path),
+    )
+
+    response = TestClient(create_app(config)).post(
+        "/api/webhooks/github",
+        content=payload,
+        headers=signed_headers(payload, delivery="workflow-success", event="workflow_run"),
+    )
+
+    assert response.json()["enqueue_status"] == "ignored"
+    with sqlite3.connect(config.db) as con:
+        assert con.execute("SELECT COUNT(*) FROM jobs").fetchone()[0] == 0
+
+
 def test_webhook_shadow_selects_secret_by_repository_owner(tmp_path):
     payload = issue_comment_payload()
     client = TestClient(create_app(DashboardConfig(