diff --git a/CHANGELOG.md b/CHANGELOG.md
index 54a320c..2953c94 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,9 +4,21 @@
 
 ### 보안
 - 리포트 출력 하드닝 — 생성된 markdown 리포트가 HTML로 렌더될 때 악성 finding 내용(예: 외부 엔진이 스캔한 코드의 `<script>`)이 주입되지 않도록, 프로즈 필드(message/remediation/verification)를 HTML 이스케이프하고 snippet의 code-fence 탈출을 무력화합니다(모든 리포트 타입). rule_id/category/context 등 제약된 식별자는 그대로 둡니다.
+- control plane API 하드닝: (1) 요청 본문을 10MiB로 캡하고 음수 Content-Length를 거부합니다(유효 키 소지자의 OOM/EOF-hang 방지). (2) `limit`/`offset` 쿼리 파라미터를 클램프합니다 — sqlite에서 `LIMIT -1`은 무제한이므로 음수를 그대로 전달하면 페이지네이션 캡이 우회됐습니다(list 1..1000, trend 1..365, offset ≥0).
 
 ### 추가
 - `appguardrail fix` 명령 — 안전하고 결정적인 자동 수정을 적용합니다(기본 dry-run diff, `--apply`로 기록). 의미를 바꾸지 않는 순수 additive 변환만 수행하며, 첫 변환으로 외부 `target="_blank"` 링크에 `rel="noopener noreferrer"`를 추가합니다(reverse tabnabbing 방지). 동작을 바꾸는 수정(시크릿→env 등)은 위험하므로 자동 적용하지 않고 fix-pack 프롬프트로 남깁니다. scan→fix→verify 루프를 안전하게 닫습니다.
+- `appguardrail serve` — 멀티테넌트 **control-plane API**(스캔 인제스트 + 히스토리). 일회성 CLI를 넘어, CI가 매 스캔의 `appguardrail.findings.v1`을 org별 API 키로 영속 저장하고 시간에 따른 추이를 조회할 수 있는 지속형 백본입니다. stdlib(sqlite3 + http.server)만 사용하며 org별 테넌트 격리를 강제합니다.
+  - 엔드포인트: `POST /api/v1/scans`(인제스트), `GET /api/v1/scans`(히스토리), `GET /api/v1/scans/{id}`(상세), `GET /api/v1/health`.
+  - 인증: `Authorization: Bearer <api_key>`. `--create-org <name>`으로 org·키 발급, 빈 DB면 기본 org를 부트스트랩합니다.
+  - **org console** — control-plane 서버가 `/`에서 서빙하는 단일 정적 페이지(`scanner/dashboard/console.html`). API 키로 연결해 스캔 히스토리, deploy-blocking 추이, 스캔 상세를 봅니다(프레임워크·빌드 단계 없음).
+  - **drift 감지** — 인제스트 시 같은 org+repo의 직전 스캔 대비 **신규 deploy-blocking** 수(`new_blocking`)를 계산합니다(line-독립 지문). console과 API 응답에 노출됩니다.
+  - **API 페이지네이션 + trend** — `GET /api/v1/scans?limit=&offset=`로 스캔 히스토리를 페이징하고, `GET /api/v1/scans/trend?limit=`로 시간순(오래된→최신) deploy_blocking·new_blocking 시계열을 얻습니다(차트용).
+  - **RBAC / 멀티유저** — org별 다중 API 키에 역할(viewer/member/owner)을 부여합니다. viewer=읽기, member=스캔 인제스트, owner=webhook·키 발급 포함 전체. `POST /api/v1/keys`(owner)로 역할 지정 키를 발급합니다. 부트스트랩 키는 owner입니다.
+  - **drift 알림 webhook** — org에 webhook URL을 설정하면(`POST /api/v1/webhook`) `new_blocking > 0`인 스캔에서 알림을 POST합니다(best-effort, 인제스트 실패 안 함). detect→alert 루프를 닫습니다.
+  - **Slack 포맷 drift 알림** — webhook 호스트가 `hooks.slack.com`이면 payload를 Slack Block Kit 메시지(헤더 + org·신규 blocker 수·repo·scan, 상위 5개 `rule_id`/파일 목록과 `+N more` 오버플로)로 자동 렌더링해 Slack Incoming Webhook이 읽기 좋은 카드로 표시합니다. 그 외 URL은 기존 generic JSON payload를 그대로 받습니다(하위 호환). 무의존성 유지를 위해 stdlib만 사용하며 텍스트는 이스케이프·트림합니다.
+  - `appguardrail scan --push <url>` — 스캔 후 findings를 control-plane에 POST합니다(키는 `APPGUARDRAIL_API_KEY`, repo/commit은 `GITHUB_REPOSITORY`/`GITHUB_SHA`에서 자동). CI가 매 스캔을 플랫폼에 밀어넣어 continuous-monitoring 루프를 닫습니다.
+  - `appguardrail monitor` 워크플로가 `APPGUARDRAIL_CONTROL_PLANE_URL` secret이 설정된 경우 스캔을 control-plane에 자동 push합니다(`APPGUARDRAIL_API_KEY` secret 사용). 미설정 시 기존 SARIF+게이트 동작 그대로.
 - `appguardrail sbom` — 의존성 매니페스트(npm `package-lock.json`/`package.json`, Python `requirements.txt`)에서 CycloneDX 1.5 SBOM을 생성합니다. 무의존성(stdlib)으로 동작하며, lockfile이 있으면 resolved 버전을, 없으면 매니페스트의 declared 범위를 사용하고 컴포넌트 properties에 출처를 기록합니다. 공급망 실사(due diligence)의 기본 산출물입니다.
 - `appguardrail sbom`의 lockfile 파서를 확장했습니다 — `poetry.lock`(pypi), `pnpm-lock.yaml`·`yarn.lock`(npm)을 추가로 인식합니다. 서드파티 toml/yaml 라이브러리 없이 stdlib만으로 손수 파싱하며(정규식·라인 스캔), scoped npm 패키지(`@scope/name`)·pnpm peer-dependency 접미사·yarn 다중 spec 헤더를 처리하고 resolved 버전으로 기록합니다. npm 측은 `package-lock.json` > `pnpm-lock.yaml` > `yarn.lock` > `package.json` 순으로 우선하고, `poetry.lock`은 Python 측에 additive로 더해집니다.
 
diff --git a/README.md b/README.md
index 001cf25..361284d 100644
--- a/README.md
+++ b/README.md
@@ -110,7 +110,9 @@ The SARIF output feeds GitHub code scanning
 SARIF consumer — findings appear in the GitHub **Security** tab and as inline PR
 annotations, ranked by `security-severity`. `appguardrail monitor` installs a
 workflow that emits and uploads SARIF automatically while preserving the deploy
-gate.
+gate. Set the `APPGUARDRAIL_CONTROL_PLANE_URL` and `APPGUARDRAIL_API_KEY`
+repository secrets and that workflow also pushes every scan to your control
+plane for history and drift tracking.
 
 Detects:
 - Hardcoded secrets (`SUPABASE_SERVICE_ROLE_KEY`, `STRIPE_SECRET_KEY`, etc.)
@@ -165,6 +167,53 @@ transform adds `rel="noopener noreferrer"` to external `target="_blank"` links
 (reverse-tabnabbing). Behavior-changing fixes (moving a secret to an env var,
 flipping TLS verification) stay as reviewable prompts — see
 `appguardrail report fix-pack`. This closes the scan → fix → verify loop safely.
+
+### Run the control-plane API (scan history)
+
+```bash
+# Provision an org + API key
+appguardrail serve --db cp.db --create-org "Acme"
+
+# Run the API (bootstraps a default org + key on an empty DB)
+appguardrail serve --db cp.db --port 8788
+```
+
+`appguardrail serve` turns AppGuardrail from a one-shot CLI into a persistent,
+multi-tenant surface: CI pushes each scan and the org queries its history.
+
+```bash
+# From CI, after `appguardrail scan --findings-json findings.json .`
+curl -X POST http://localhost:8788/api/v1/scans \
+  -H "Authorization: Bearer $APPGUARDRAIL_API_KEY" \
+  -H "Content-Type: application/json" \
+  -d @findings.json     # or {"repo":"...","commit":"...","findings":[...]}
+
+curl http://localhost:8788/api/v1/scans -H "Authorization: Bearer $APPGUARDRAIL_API_KEY"
+```
+
+Push scans from CI with `appguardrail scan --push http://your-control-plane .`
+(key from `APPGUARDRAIL_API_KEY`). The control plane computes **drift** — the
+number of deploy-blocking findings newly introduced since the repo's previous
+scan. Open the **org console** at `http://localhost:8788/` — a single static page that
+connects with your API key and shows scan history, the deploy-blocking trend,
+and per-scan detail.
+
+Issue role-scoped API keys with `POST /api/v1/keys` (owner only): `viewer`
+(read), `member` (ingest scans), or `owner` (full). The bootstrap key is an
+owner.
+
+Set a drift-alert webhook (`POST /api/v1/webhook` with `{"url":"…"}`) and the
+control plane notifies it whenever a scan introduces new deploy-blocking
+findings. If the URL is a Slack Incoming Webhook (`hooks.slack.com`), the alert
+is automatically formatted as a Slack Block Kit message — a header with the new
+blocker count plus the org, repo, scan id, and the top offending rule ids and
+files — so Slack renders a readable card. Any other URL receives the generic
+JSON payload unchanged.
+
+Endpoints: `POST /api/v1/scans`, `GET /api/v1/scans`, `GET /api/v1/scans/{id}`,
+`POST /api/v1/webhook`, `GET /api/v1/health`. Tenant-isolated by API key. Stdlib + SQLite (swap for a
+managed database behind the same functions at scale).
+
 ### Generate a CycloneDX SBOM
 
 ```bash
diff --git a/appguardrail_core/controlplane.py b/appguardrail_core/controlplane.py
new file mode 100644
index 0000000..0824733
--- /dev/null
+++ b/appguardrail_core/controlplane.py
@@ -0,0 +1,578 @@
+"""Multi-tenant control-plane store for AppGuardrail scan history.
+
+This is the seed of the hosted platform: instead of a one-shot CLI, an org can
+push every CI scan (`appguardrail.findings.v1`) to a persistent, API-key-scoped
+store and query its history/trend. That persistent, multi-tenant surface is the
+recurring-revenue backbone a CLI alone can't be.
+
+Stdlib only (sqlite3 + hashlib + secrets), so it ships in the same wheel and
+adds no dependency. SQLite is the ``ponytail`` starting point — swap the store
+for Postgres behind the same functions when scale demands it.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import secrets
+import sqlite3
+from importlib import resources
+from urllib.parse import urlparse, parse_qs
+from datetime import datetime, timezone
+from typing import Any, Iterable
+
+from .findings import is_deploy_blocking, normalize_findings, severity_counts
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS orgs (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    name TEXT NOT NULL,
+    api_key_hash TEXT NOT NULL UNIQUE,
+    created_at TEXT NOT NULL,
+    webhook_url TEXT
+);
+CREATE TABLE IF NOT EXISTS scans (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    org_id INTEGER NOT NULL,
+    created_at TEXT NOT NULL,
+    repo TEXT,
+    commit_sha TEXT,
+    total INTEGER NOT NULL,
+    deploy_blocking INTEGER NOT NULL,
+    severity_counts TEXT NOT NULL,
+    new_blocking INTEGER NOT NULL DEFAULT 0,
+    findings TEXT NOT NULL,
+    FOREIGN KEY (org_id) REFERENCES orgs (id)
+);
+CREATE INDEX IF NOT EXISTS idx_scans_org ON scans (org_id, id DESC);
+CREATE TABLE IF NOT EXISTS keys (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    org_id INTEGER NOT NULL,
+    key_hash TEXT NOT NULL UNIQUE,
+    role TEXT NOT NULL DEFAULT 'member',
+    label TEXT,
+    created_at TEXT NOT NULL,
+    FOREIGN KEY (org_id) REFERENCES orgs (id)
+);
+"""
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+# scrypt work factors (stdlib ``hashlib.scrypt``). n must be a power of two;
+# these are the interactive-login reference parameters and stay well within the
+# default 32 MiB ``maxmem`` (n*r*128 ≈ 16 MiB).
+_SCRYPT_N = 2 ** 14
+_SCRYPT_R = 8
+_SCRYPT_P = 1
+# Fixed application salt (pepper). API-key hashes are looked up by equality
+# (``WHERE api_key_hash = ?``), so hashing must be deterministic — a per-call
+# random salt would make every stored key unfindable. A constant salt keeps the
+# lookup working while scrypt's memory-hard cost defeats offline brute-force.
+_KEY_SALT = b"appguardrail.controlplane.key.v1"
+
+
+def _hash_key(api_key: str) -> str:
+    """Derive a deterministic, brute-force-resistant hash of an API key.
+
+    Uses scrypt (a memory-hard KDF from the stdlib) instead of a fast hash such
+    as SHA-256: API keys are secrets, so if the store leaks, an attacker must
+    pay scrypt's tunable compute/memory cost per guess rather than hashing
+    billions of candidates per second. The fixed application salt keeps the
+    output deterministic so keys remain findable by an indexed equality lookup.
+    """
+    return hashlib.scrypt(
+        api_key.encode("utf-8"),
+        salt=_KEY_SALT,
+        n=_SCRYPT_N,
+        r=_SCRYPT_R,
+        p=_SCRYPT_P,
+    ).hex()
+
+
+def connect(db_path: str) -> sqlite3.Connection:
+    """Open (and initialize) the control-plane database."""
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    conn.executescript(_SCHEMA)
+    conn.commit()
+    return conn
+
+
+def create_org(conn: sqlite3.Connection, name: str) -> "tuple[int, str]":
+    """Create an org and return (org_id, api_key). The key is shown only here."""
+    api_key = "agk_" + secrets.token_urlsafe(32)
+    cur = conn.execute(
+        "INSERT INTO orgs (name, api_key_hash, created_at) VALUES (?, ?, ?)",
+        (name, _hash_key(api_key), _now()),
+    )
+    org_id = cur.lastrowid
+    conn.execute(
+        "INSERT INTO keys (org_id, key_hash, role, label, created_at) VALUES (?, ?, ?, ?, ?)",
+        (org_id, _hash_key(api_key), "owner", "owner (bootstrap)", _now()),
+    )
+    conn.commit()
+    return org_id, api_key
+
+
+def org_for_key(conn: sqlite3.Connection, api_key: str) -> "int | None":
+    """Return the org id for a presented API key, or None."""
+    if not api_key:
+        return None
+    row = conn.execute(
+        "SELECT id FROM orgs WHERE api_key_hash = ?", (_hash_key(api_key),)
+    ).fetchone()
+    return row["id"] if row else None
+
+
+def _drift_fp(finding: dict[str, Any]) -> str:
+    """Coarse identity for drift: rule + file + message head (line-independent)."""
+    return f"{finding.get('rule_id')}|{finding.get('file')}|{str(finding.get('message', ''))[:80]}"
+
+
+def set_webhook(conn: sqlite3.Connection, org_id: int, url: "str | None") -> None:
+    """Set (or clear) the org's drift-alert webhook URL."""
+    conn.execute("UPDATE orgs SET webhook_url = ? WHERE id = ?", (url or None, org_id))
+    conn.commit()
+
+
+def _is_slack_webhook(url: str) -> bool:
+    """True if ``url`` is a Slack Incoming Webhook (host under hooks.slack.com)."""
+    try:
+        host = (urlparse(url).hostname or "").lower()
+    except ValueError:
+        return False
+    return host == "hooks.slack.com" or host.endswith(".hooks.slack.com")
+
+
+def _slack_escape(text: str) -> str:
+    """Escape the three characters Slack treats specially in message text."""
+    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def _trim(text: str, limit: int) -> str:
+    """Trim ``text`` to ``limit`` chars (Slack block text has hard caps)."""
+    return text if len(text) <= limit else text[: max(0, limit - 1)] + "…"
+
+
+def _slack_blocks(
+    org_name: "str | None",
+    payload: dict[str, Any],
+    new_findings: "list[dict[str, Any]]",
+    top: int = 5,
+) -> dict[str, Any]:
+    """Render a drift alert as a Slack Block Kit message (header + summary).
+
+    Lists the org, the count of newly-introduced deploy blockers, and up to
+    ``top`` offending ``rule_id`` / ``file`` pairs with a ``+N more`` overflow
+    line. Returns the dict POSTed to a Slack Incoming Webhook.
+    """
+    org = org_name or f"org {payload.get('org_id')}"
+    n = payload.get("new_blocking", 0)
+    repo = payload.get("repo") or "—"
+    scan_id = payload.get("scan_id")
+
+    shown = new_findings[:top]
+    lines = [
+        "• `{rule}` — {file}".format(
+            rule=_slack_escape(str(f.get("rule_id") or "?")),
+            file=_slack_escape(str(f.get("file") or "?")),
+        )
+        for f in shown
+    ]
+    remaining = len(new_findings) - len(shown)
+    if remaining > 0:
+        lines.append(f"• +{remaining} more")
+    detail = "\n".join(lines) if lines else "_no finding details available_"
+
+    header = f"{n} new deploy-blocking finding{'s' if n != 1 else ''}"
+    blocks: list[dict[str, Any]] = [
+        {"type": "header", "text": {"type": "plain_text", "text": _trim(header, 150)}},
+        {
+            "type": "section",
+            "fields": [
+                {"type": "mrkdwn", "text": _trim(f"*Org:*\n{_slack_escape(org)}", 2000)},
+                {"type": "mrkdwn", "text": f"*New blockers:*\n{n}"},
+                {"type": "mrkdwn", "text": _trim(f"*Repo:*\n{_slack_escape(str(repo))}", 2000)},
+                {"type": "mrkdwn", "text": f"*Scan:*\n#{scan_id}"},
+            ],
+        },
+        {"type": "section", "text": {"type": "mrkdwn", "text": _trim(detail, 3000)}},
+    ]
+    return {
+        "text": _trim(f"{header} in {_slack_escape(org)}", 3000),
+        "blocks": blocks,
+    }
+
+
+def _send_alert(
+    url: str,
+    payload: dict[str, Any],
+    *,
+    org_name: "str | None" = None,
+    new_findings: "list[dict[str, Any]] | None" = None,
+) -> bool:
+    """Best-effort POST of a drift alert. Never raises; returns delivery success.
+
+    For Slack Incoming Webhook URLs (host ``hooks.slack.com``) the alert is
+    rendered as a Block Kit message so Slack shows a readable card; every other
+    URL receives the generic JSON ``payload`` unchanged (backward compatible).
+    """
+    import urllib.error
+    import urllib.request
+
+    if _is_slack_webhook(url):
+        body = _slack_blocks(org_name, payload, new_findings or [])
+    else:
+        body = payload
+
+    try:
+        req = urllib.request.Request(
+            url,
+            data=json.dumps(body).encode("utf-8"),
+            method="POST",
+            headers={"Content-Type": "application/json"},
+        )
+        urllib.request.urlopen(req, timeout=10)
+        return True
+    except (urllib.error.URLError, OSError, ValueError):
+        return False
+
+
+ROLES = ("viewer", "member", "owner")
+_ROLE_RANK = {role: rank for rank, role in enumerate(ROLES)}
+
+
+def has_role(role: "str | None", minimum: str) -> bool:
+    """True if ``role`` is at or above ``minimum`` in the viewer<member<owner order."""
+    return _ROLE_RANK.get(role or "", -1) >= _ROLE_RANK.get(minimum, 99)
+
+
+def create_key(
+    conn: sqlite3.Connection, org_id: int, role: str = "member", label: "str | None" = None
+) -> "tuple[int, str]":
+    """Issue a new API key for an org with a role. Returns (key_id, api_key)."""
+    role = role if role in _ROLE_RANK else "member"
+    api_key = "agk_" + secrets.token_urlsafe(32)
+    cur = conn.execute(
+        "INSERT INTO keys (org_id, key_hash, role, label, created_at) VALUES (?, ?, ?, ?, ?)",
+        (org_id, _hash_key(api_key), role, label, _now()),
+    )
+    conn.commit()
+    return cur.lastrowid, api_key
+
+
+def role_for_key(conn: sqlite3.Connection, api_key: str) -> "tuple[int, str] | None":
+    """Return (org_id, role) for a presented key, or None."""
+    if not api_key:
+        return None
+    row = conn.execute(
+        "SELECT org_id, role FROM keys WHERE key_hash = ?", (_hash_key(api_key),)
+    ).fetchone()
+    if row:
+        return (row["org_id"], row["role"])
+    # legacy/bootstrap key stored on orgs is an owner key
+    row = conn.execute(
+        "SELECT id FROM orgs WHERE api_key_hash = ?", (_hash_key(api_key),)
+    ).fetchone()
+    return (row["id"], "owner") if row else None
+
+
+def add_scan(
+    conn: sqlite3.Connection,
+    org_id: int,
+    findings: Iterable[dict[str, Any]],
+    repo: "str | None" = None,
+    commit_sha: "str | None" = None,
+) -> dict[str, Any]:
+    """Store a scan for an org, computing counts from the findings."""
+    normalized = list(normalize_findings(findings))
+    counts = severity_counts(normalized)
+    blocking_findings = [f for f in normalized if is_deploy_blocking(f)]
+    blocking = len(blocking_findings)
+
+    # Drift: deploy-blocking findings new since this org+repo's previous scan.
+    prev = conn.execute(
+        "SELECT findings FROM scans WHERE org_id = ? AND IFNULL(repo, '') = IFNULL(?, '') "
+        "ORDER BY id DESC LIMIT 1",
+        (org_id, repo),
+    ).fetchone()
+    prev_fps = set()
+    if prev:
+        prev_fps = {
+            _drift_fp(f) for f in json.loads(prev["findings"]) if is_deploy_blocking(f)
+        }
+    new_findings = [f for f in blocking_findings if _drift_fp(f) not in prev_fps]
+    new_blocking = len(new_findings)
+
+    created_at = _now()
+    cur = conn.execute(
+        "INSERT INTO scans (org_id, created_at, repo, commit_sha, total, "
+        "deploy_blocking, severity_counts, new_blocking, findings) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            org_id,
+            created_at,
+            repo,
+            commit_sha,
+            len(normalized),
+            blocking,
+            json.dumps(counts),
+            new_blocking,
+            json.dumps(normalized),
+        ),
+    )
+    conn.commit()
+
+    # Drift alert: notify the org's webhook when new blockers were introduced.
+    if new_blocking > 0:
+        row = conn.execute(
+            "SELECT name, webhook_url FROM orgs WHERE id = ?", (org_id,)
+        ).fetchone()
+        hook = row["webhook_url"] if row else None
+        if hook:
+            _send_alert(
+                hook,
+                {
+                    "event": "drift.new_blocking",
+                    "org_id": org_id,
+                    "scan_id": cur.lastrowid,
+                    "repo": repo,
+                    "commit": commit_sha,
+                    "new_blocking": new_blocking,
+                    "deploy_blocking": blocking,
+                    "created_at": created_at,
+                },
+                org_name=row["name"] if row else None,
+                new_findings=new_findings,
+            )
+
+    return {
+        "id": cur.lastrowid,
+        "created_at": created_at,
+        "total": len(normalized),
+        "deploy_blocking": blocking,
+        "new_blocking": new_blocking,
+        "severity_counts": counts,
+    }
+
+
+def list_scans(conn: sqlite3.Connection, org_id: int, limit: int = 100, offset: int = 0) -> list[dict[str, Any]]:
+    """Return scan summaries for an org, newest first."""
+    rows = conn.execute(
+        "SELECT id, created_at, repo, commit_sha, total, deploy_blocking, new_blocking, severity_counts "
+        "FROM scans WHERE org_id = ? ORDER BY id DESC LIMIT ? OFFSET ?",
+        (org_id, limit, max(0, offset)),
+    ).fetchall()
+    return [
+        {
+            "id": r["id"],
+            "created_at": r["created_at"],
+            "repo": r["repo"],
+            "commit": r["commit_sha"],
+            "total": r["total"],
+            "deploy_blocking": r["deploy_blocking"],
+            "new_blocking": r["new_blocking"],
+            "severity_counts": json.loads(r["severity_counts"]),
+        }
+        for r in rows
+    ]
+
+
+def scan_trend(conn: sqlite3.Connection, org_id: int, limit: int = 30) -> list[dict[str, Any]]:
+    """Oldest->newest deploy_blocking/new_blocking series for charting."""
+    rows = conn.execute(
+        "SELECT created_at, deploy_blocking, new_blocking FROM scans "
+        "WHERE org_id = ? ORDER BY id DESC LIMIT ?",
+        (org_id, max(1, limit)),
+    ).fetchall()
+    return [
+        {"created_at": r["created_at"], "deploy_blocking": r["deploy_blocking"],
+         "new_blocking": r["new_blocking"]}
+        for r in reversed(rows)
+    ]
+
+
+def get_scan(conn: sqlite3.Connection, org_id: int, scan_id: int) -> "dict[str, Any] | None":
+    """Return a full scan (with findings) scoped to the org, or None."""
+    r = conn.execute(
+        "SELECT * FROM scans WHERE id = ? AND org_id = ?", (scan_id, org_id)
+    ).fetchone()
+    if r is None:
+        return None
+    return {
+        "id": r["id"],
+        "created_at": r["created_at"],
+        "repo": r["repo"],
+        "commit": r["commit_sha"],
+        "total": r["total"],
+        "deploy_blocking": r["deploy_blocking"],
+        "new_blocking": r["new_blocking"],
+        "severity_counts": json.loads(r["severity_counts"]),
+        "findings": json.loads(r["findings"]),
+    }
+
+
+
+def console_html() -> bytes:
+    """Return the packaged org-console HTML, or a minimal fallback."""
+    try:
+        path = resources.files("scanner").joinpath("dashboard", "console.html")
+        return path.read_bytes()
+    except (FileNotFoundError, ModuleNotFoundError, OSError):
+        return b"<!doctype html><title>AppGuardrail Console</title><p>Console asset missing.</p>"
+
+
+def make_control_plane_server(host: str, port: int, db_path: str):
+    """Build an HTTP API for scan ingest + history, scoped by API key.
+
+    Endpoints (JSON):
+      GET  /api/v1/health         -> {"status":"ok"} (no auth)
+      POST /api/v1/scans          -> ingest {findings:[...], repo?, commit?}
+      GET  /api/v1/scans          -> list this org's scans (summaries)
+      GET  /api/v1/scans/{id}     -> full scan with findings
+    Auth: Authorization: Bearer <api_key>.
+    """
+    import http.server
+
+    conn = sqlite3.connect(db_path, check_same_thread=False)
+    conn.row_factory = sqlite3.Row
+    conn.executescript(_SCHEMA)
+    conn.commit()
+    console = console_html()
+
+    class _Handler(http.server.BaseHTTPRequestHandler):
+        def _json(self, code, obj):
+            body = json.dumps(obj).encode("utf-8")
+            self.send_response(code)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def _auth(self):
+            hdr = self.headers.get("Authorization", "")
+            key = hdr[7:] if hdr.startswith("Bearer ") else ""
+            return role_for_key(conn, key)
+
+        def do_GET(self):
+            parsed = urlparse(self.path)
+            path = parsed.path
+            qs = parse_qs(parsed.query)
+
+            def _qint(name, default, lo, hi):
+                # Clamp: sqlite treats LIMIT -1 as "no limit", so never pass
+                # negatives through; hi keeps a single request bounded.
+                try:
+                    value = int(qs.get(name, [default])[0])
+                except (ValueError, TypeError):
+                    return default
+                return max(lo, min(hi, value))
+
+            if path in ("/", "/console", "/index.html"):
+                self.send_response(200)
+                self.send_header("Content-Type", "text/html; charset=utf-8")
+                self.send_header("Content-Length", str(len(console)))
+                self.end_headers()
+                self.wfile.write(console)
+                return
+            if path == "/api/v1/health":
+                return self._json(200, {"status": "ok"})
+            auth = self._auth()
+            if auth is None:
+                return self._json(401, {"error": "invalid or missing API key"})
+            org, _role = auth
+            if path == "/api/v1/scans":
+                return self._json(200, {"scans": list_scans(conn, org, _qint("limit", 100, 1, 1000), _qint("offset", 0, 0, 10**9))})
+            if path == "/api/v1/scans/trend":
+                return self._json(200, {"trend": scan_trend(conn, org, _qint("limit", 30, 1, 365))})
+            m = re.match(r"^/api/v1/scans/(\d+)$", path)
+            if m:
+                scan = get_scan(conn, org, int(m.group(1)))
+                return self._json(200, scan) if scan else self._json(404, {"error": "not found"})
+            return self._json(404, {"error": "not found"})
+
+        _MAX_BODY = 10 * 1024 * 1024  # 10 MiB — plenty for findings, blocks OOM posts
+
+        def _body(self):
+            try:
+                length = int(self.headers.get("Content-Length", 0))
+            except (ValueError, TypeError):
+                return None
+            if length < 0 or length > self._MAX_BODY:
+                # Negative reads until EOF; oversized bodies exhaust memory.
+                return None
+            try:
+                return json.loads(self.rfile.read(length) or b"{}")
+            except (ValueError, TypeError):
+                return None
+
+        def do_POST(self):
+            path = self.path.split("?", 1)[0]
+            if path not in ("/api/v1/scans", "/api/v1/webhook", "/api/v1/keys"):
+                return self._json(404, {"error": "not found"})
+            auth = self._auth()
+            if auth is None:
+                return self._json(401, {"error": "invalid or missing API key"})
+            org, role = auth
+
+            if path == "/api/v1/webhook":
+                if not has_role(role, "owner"):
+                    return self._json(403, {"error": "owner role required"})
+                body = self._body()
+                if body is None:
+                    return self._json(400, {"error": "invalid JSON body"})
+                set_webhook(conn, org, (body or {}).get("url"))
+                return self._json(200, {"webhook_url": (body or {}).get("url")})
+
+            if path == "/api/v1/keys":
+                if not has_role(role, "owner"):
+                    return self._json(403, {"error": "owner role required"})
+                body = self._body()
+                if body is None:
+                    return self._json(400, {"error": "invalid JSON body"})
+                new_role = (body or {}).get("role", "member")
+                _kid, new_key = create_key(conn, org, new_role, (body or {}).get("label"))
+                return self._json(201, {"api_key": new_key, "role": new_role if new_role in ROLES else "member"})
+            if not has_role(role, "member"):
+                return self._json(403, {"error": "member role required to ingest scans"})
+            data = self._body()
+            if data is None:
+                return self._json(400, {"error": "invalid JSON body"})
+            findings = data.get("findings") if isinstance(data, dict) else data
+            if not isinstance(findings, list):
+                return self._json(400, {"error": "expected a findings array or {\"findings\":[...]}"})
+            meta = data if isinstance(data, dict) else {}
+            summary = add_scan(conn, org, findings, meta.get("repo"), meta.get("commit"))
+            return self._json(201, summary)
+
+        def log_message(self, *_args):
+            pass
+
+    return http.server.HTTPServer((host, port), _Handler)
+
+
+if __name__ == "__main__":  # pragma: no cover - self-check
+    conn = connect(":memory:")
+    oid, key = create_org(conn, "Acme")
+    assert org_for_key(conn, key) == oid
+    assert org_for_key(conn, "agk_wrong") is None
+    s = add_scan(
+        conn, oid,
+        [{"severity": "CRITICAL", "rule_id": "x", "context": "app-code"},
+         {"severity": "INFO", "rule_id": "y", "context": "doc"}],
+        repo="acme/app", commit_sha="abc123",
+    )
+    assert s["total"] == 2 and s["deploy_blocking"] == 1, s
+    listed = list_scans(conn, oid)
+    assert len(listed) == 1 and listed[0]["repo"] == "acme/app"
+    full = get_scan(conn, oid, s["id"])
+    assert full and len(full["findings"]) == 2
+    # tenant isolation: another org can't read the first org's scan
+    oid2, _ = create_org(conn, "Beta")
+    assert get_scan(conn, oid2, s["id"]) is None
+    assert list_scans(conn, oid2) == []
+    print("controlplane self-check OK")
diff --git a/scanner/cli/appguardrail.py b/scanner/cli/appguardrail.py
index 4fd640d..80e8f6f 100644
--- a/scanner/cli/appguardrail.py
+++ b/scanner/cli/appguardrail.py
@@ -246,6 +246,8 @@
 jobs:
   scan:
     runs-on: ubuntu-latest
+    env:
+      CP_URL: ${{ secrets.APPGUARDRAIL_CONTROL_PLANE_URL }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
@@ -258,10 +260,15 @@
       - name: Install AppGuardrail
         run: python -m pip install --disable-pip-version-check appguardrail
 
-      - name: Run AppGuardrail (SARIF + deploy gate)
+      - name: Run AppGuardrail (SARIF + deploy gate; push to control plane if configured)
         id: scan
         continue-on-error: true
-        run: appguardrail scan --sarif appguardrail.sarif .
+        env:
+          APPGUARDRAIL_API_KEY: ${{ secrets.APPGUARDRAIL_API_KEY }}
+        run: |
+          PUSH=""
+          if [ -n "$CP_URL" ]; then PUSH="--push $CP_URL"; fi
+          appguardrail scan --sarif appguardrail.sarif $PUSH .
 
       - name: Upload results to GitHub code scanning
         if: always()
@@ -1521,6 +1528,10 @@ def cmd_scan(args):
             )
             return 1
 
+    push_url = getattr(args, "push", None)
+    if push_url:
+        _push_findings(push_url, findings)
+
     _print_scan_results(findings, files_scanned)
     if files_scanned == 0:
         return 1
@@ -1571,6 +1582,39 @@ def _write_findings_json(findings, output_path: Path):
     print(f"🧾 Findings JSON written: {output_path}")
 
 
+def _push_findings(url, findings):
+    """POST normalized findings to a control-plane /api/v1/scans endpoint."""
+    import urllib.error
+    import urllib.request
+
+    api_key = os.environ.get("APPGUARDRAIL_API_KEY", "")
+    if not api_key:
+        print("⚠️  --push set but APPGUARDRAIL_API_KEY is empty; skipping push.", file=sys.stderr)
+        return
+    payload = {
+        "findings": list(normalize_findings(findings)),
+        "repo": os.environ.get("GITHUB_REPOSITORY"),
+        "commit": os.environ.get("GITHUB_SHA"),
+    }
+    endpoint = url.rstrip("/") + "/api/v1/scans"
+    req = urllib.request.Request(
+        endpoint,
+        data=json.dumps(payload).encode("utf-8"),
+        method="POST",
+        headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"},
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            body = json.loads(resp.read() or b"{}")
+        drift = body.get("new_blocking")
+        extra = f", {drift} newly deploy-blocking" if drift else ""
+        print(f"📡 Pushed scan #{body.get('id')} to control plane{extra}.")
+    except urllib.error.HTTPError as exc:
+        print(f"⚠️  Control-plane push failed ({exc.code}); scan still completed.", file=sys.stderr)
+    except (urllib.error.URLError, OSError, ValueError) as exc:
+        print(f"⚠️  Control-plane push failed ({exc}); scan still completed.", file=sys.stderr)
+
+
 def _write_sarif(findings, output_path: Path):
     """Write SARIF 2.1.0 for GitHub code scanning and other SARIF consumers."""
     from appguardrail_core.sarif import findings_to_sarif
@@ -2971,6 +3015,46 @@ def log_message(self, *_args):  # keep the console quiet
     return http.server.HTTPServer((host, port), _Handler)
 
 
+def cmd_serve(args):
+    """Run the AppGuardrail control-plane API (scan ingest + history)."""
+    from appguardrail_core import controlplane as cp
+
+    db = getattr(args, "db", None) or "appguardrail-control-plane.db"
+    conn = cp.connect(db)
+    create = getattr(args, "create_org", None)
+    if create:
+        oid, key = cp.create_org(conn, create)
+        conn.close()
+        print(f"✅ Created org '{create}' (id {oid}).")
+        print(f"🔑 API key (store it now — shown only once): {key}")
+        return 0
+    if conn.execute("SELECT COUNT(*) AS c FROM orgs").fetchone()["c"] == 0:
+        _oid, key = cp.create_org(conn, "default")
+        print("ℹ️  No orgs yet — created 'default'.")
+        print(f"🔑 API key (store it now — shown only once): {key}\n")
+    conn.close()
+
+    host = getattr(args, "host", "127.0.0.1")
+    port = getattr(args, "port", 8788)
+    try:
+        server = cp.make_control_plane_server(host, port, db)
+    except OSError as exc:
+        print(f"❌ Cannot start control plane on {host}:{port} ({exc}).", file=sys.stderr)
+        print("💡 Pass a free port with --port.", file=sys.stderr)
+        return 1
+    actual = server.server_address[1]
+    print(f"🛰️  AppGuardrail control plane on http://{host}:{actual}")
+    print("   POST /api/v1/scans · GET /api/v1/scans · GET /api/v1/scans/{id} · GET /api/v1/health")
+    print("   Auth: Authorization: Bearer <api_key>. Ctrl+C to stop.")
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        print("\n👋 Control plane stopped.")
+    finally:
+        server.server_close()
+    return 0
+
+
 def cmd_sbom(args):
     """Generate a CycloneDX SBOM from dependency manifests."""
     from appguardrail_core.sbom import build_sbom, collect_components
@@ -3169,6 +3253,12 @@ def main():
         default=None,
         help="Write SARIF 2.1.0 for GitHub code scanning, VS Code, and other tools",
     )
+    scan_parser.add_argument(
+        "--push",
+        default=None,
+        metavar="URL",
+        help="POST findings to a control-plane URL (key from APPGUARDRAIL_API_KEY)",
+    )
     scan_parser.add_argument(
         "--codegraph",
         action="store_true",
@@ -3312,6 +3402,13 @@ def add_report_arguments(parser):
         action="store_true",
         help="Write fixes to disk (default: show a dry-run diff)",
     )
+    serve_parser = subparsers.add_parser(
+        "serve", help="Run the control-plane API (multi-tenant scan ingest + history)"
+    )
+    serve_parser.add_argument("--db", default=None, help="SQLite database path (default: appguardrail-control-plane.db)")
+    serve_parser.add_argument("--host", default="127.0.0.1", help="Bind host")
+    serve_parser.add_argument("--port", type=int, default=8788, help="Bind port")
+    serve_parser.add_argument("--create-org", default=None, metavar="NAME", help="Create an org, print its API key, and exit")
     sbom_parser = subparsers.add_parser(
         "sbom", help="Generate a CycloneDX SBOM from dependency manifests"
     )
@@ -3358,6 +3455,8 @@ def add_report_arguments(parser):
         sys.exit(cmd_hook(args))
     elif args.command == "fix":
         sys.exit(cmd_fix(args))
+    elif args.command == "serve":
+        sys.exit(cmd_serve(args))
     elif args.command == "sbom":
         sys.exit(cmd_sbom(args))
     elif args.command == "dashboard":
diff --git a/scanner/dashboard/console.html b/scanner/dashboard/console.html
new file mode 100644
index 0000000..29593aa
--- /dev/null
+++ b/scanner/dashboard/console.html
@@ -0,0 +1,132 @@
+<!doctype html>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>AppGuardrail Console</title>
+<!--
+  Org console for the AppGuardrail control plane (`appguardrail serve`).
+  Paste an org API key, then it calls GET /api/v1/scans (same origin) to show
+  scan history, the deploy-blocking trend, and per-scan detail.
+  ponytail: no framework, no build step. Consumes the API in controlplane.py.
+-->
+<style>
+  :root{
+    --bg:#F4F5F7; --surface:#FFFFFF; --text:#1A1D24; --muted:#5B6472;
+    --border:#D6DAE0; --divider:#E7EAEF; --primary:#256EF4; --on-primary:#fff;
+    --crit:#D93B3B; --high:#E06C00; --warn:#B7791F; --info:#5B6472; --ok:#1E874B;
+    --radius:12px;
+  }
+  *{box-sizing:border-box}
+  body{margin:0;font:14px/1.5 -apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif;background:var(--bg);color:var(--text)}
+  header{display:flex;gap:12px;align-items:center;padding:14px 20px;background:var(--surface);border-bottom:1px solid var(--border);flex-wrap:wrap}
+  .logo{width:26px;height:26px;border-radius:7px;background:var(--primary)}
+  h1{font-size:16px;margin:0;font-weight:700}
+  main{max-width:1000px;margin:0 auto;padding:20px}
+  input{font:inherit;padding:8px 10px;border:1px solid var(--border);border-radius:8px;min-width:280px}
+  button{font:inherit;font-weight:600;padding:8px 14px;border:0;border-radius:8px;background:var(--primary);color:var(--on-primary);cursor:pointer}
+  button.ghost{background:var(--surface);color:var(--primary);border:1px solid var(--border)}
+  .card{background:var(--surface);border:1px solid var(--border);border-radius:var(--radius);padding:16px 18px;margin-bottom:16px}
+  .stats{display:flex;gap:12px;flex-wrap:wrap}
+  .stat{flex:1;min-width:130px;background:var(--surface);border:1px solid var(--border);border-radius:var(--radius);padding:12px 14px}
+  .stat .n{font-size:26px;font-weight:800}
+  .stat .l{font-size:11px;font-weight:600;color:var(--muted);text-transform:uppercase;letter-spacing:.03em}
+  table{width:100%;border-collapse:collapse}
+  th,td{text-align:left;padding:8px 10px;border-bottom:1px solid var(--divider);font-size:13px}
+  th{font-size:11px;color:var(--muted);text-transform:uppercase}
+  tr.scan{cursor:pointer}
+  tr.scan:hover{background:var(--bg)}
+  .pill{display:inline-block;padding:1px 8px;border-radius:999px;font-size:11px;font-weight:700;color:#fff}
+  .trend{display:flex;align-items:flex-end;gap:4px;height:60px;margin-top:8px}
+  .bar{flex:1;min-width:6px;border-radius:3px 3px 0 0}
+  .muted{color:var(--muted)}
+  .err{color:var(--crit);font-weight:600}
+  code{background:var(--bg);padding:1px 5px;border-radius:4px}
+  .hidden{display:none}
+</style>
+</head>
+<body>
+<header>
+  <div class="logo"></div>
+  <h1>AppGuardrail Console</h1>
+  <span class="muted" id="conn"></span>
+  <span style="flex:1"></span>
+  <input id="key" type="password" placeholder="Org API key (agk_…)" autocomplete="off">
+  <button id="connect">Connect</button>
+  <button id="logout" class="ghost hidden">Sign out</button>
+</header>
+<main>
+  <div id="msg" class="card muted">Paste your org API key and connect to view scan history.</div>
+  <div id="app" class="hidden">
+    <div class="stats" id="stats"></div>
+    <div class="card">
+      <strong>Deploy-blocking trend</strong> <span class="muted">(oldest → newest)</span>
+      <div class="trend" id="trend"></div>
+    </div>
+    <div class="card">
+      <strong>Scan history</strong>
+      <table id="history"><thead><tr><th>When</th><th>Repo</th><th>Commit</th><th>Total</th><th>Blocking</th><th>New</th></tr></thead><tbody></tbody></table>
+    </div>
+    <div class="card hidden" id="detail"></div>
+  </div>
+</main>
+<script>
+const $=s=>document.querySelector(s);
+const esc=s=>String(s==null?"":s).replace(/[&<>"']/g,c=>({"&":"&amp;","<":"&lt;",">":"&gt;",'"':"&quot;","'":"&#39;"}[c]));
+const SEV={CRITICAL:"var(--crit)",HIGH:"var(--high)",WARNING:"var(--warn)",INFO:"var(--info)"};
+let KEY=sessionStorage.getItem("ag_key")||"";
+
+async function api(path){
+  const r=await fetch(path,{headers:{Authorization:"Bearer "+KEY}});
+  if(r.status===401)throw new Error("Invalid API key.");
+  if(!r.ok)throw new Error("Request failed ("+r.status+").");
+  return r.json();
+}
+function pill(n,color){return n>0?`<span class="pill" style="background:${color}">${n}</span>`:`<span class="muted">0</span>`;}
+
+async function load(){
+  try{
+    const {scans}=await api("/api/v1/scans");
+    $("#msg").classList.add("hidden");$("#app").classList.remove("hidden");
+    $("#logout").classList.remove("hidden");
+    $("#conn").textContent=scans.length+" scan(s)";
+    const latest=scans[0]||{severity_counts:{}};
+    const c=latest.severity_counts||{};
+    $("#stats").innerHTML=[
+      ["Latest deploy-blocking",latest.deploy_blocking||0],
+      ["New since last scan",latest.new_blocking||0],
+      ["Critical",c.CRITICAL||0],
+      ["Scans stored",scans.length],
+    ].map(([l,n])=>`<div class="stat"><div class="l">${l}</div><div class="n">${n}</div></div>`).join("");
+    // trend: oldest->newest deploy-blocking
+    const ord=[...scans].reverse();
+    const max=Math.max(1,...ord.map(s=>s.deploy_blocking||0));
+    $("#trend").innerHTML=ord.map(s=>{const h=Math.round(6+((s.deploy_blocking||0)/max)*54);
+      const col=(s.deploy_blocking||0)>0?"var(--crit)":"var(--ok)";
+      return `<div class="bar" title="${esc(s.created_at)}: ${s.deploy_blocking||0} blocking" style="height:${h}px;background:${col}"></div>`;}).join("")||'<span class="muted">No scans yet.</span>';
+    // table
+    $("#history tbody").innerHTML=scans.map(s=>`<tr class="scan" data-id="${s.id}">
+      <td>${esc(s.created_at)}</td><td>${esc(s.repo||"—")}</td><td><code>${esc((s.commit||"—").slice(0,10))}</code></td>
+      <td>${s.total}</td><td>${pill(s.deploy_blocking,"var(--crit)")}</td><td>${pill(s.new_blocking,"var(--high)")}</td></tr>`).join("")||'<tr><td colspan="6" class="muted">No scans. POST to /api/v1/scans from CI.</td></tr>';
+    document.querySelectorAll("tr.scan").forEach(tr=>tr.onclick=()=>detail(tr.dataset.id));
+  }catch(e){ $("#msg").classList.remove("hidden");$("#app").classList.add("hidden");
+    $("#msg").innerHTML=`<span class="err">${esc(e.message)}</span>`; }
+}
+async function detail(id){
+  const s=await api("/api/v1/scans/"+id);
+  const d=$("#detail");d.classList.remove("hidden");
+  const rows=(s.findings||[]).map(f=>`<tr><td><span class="pill" style="background:${SEV[f.severity]||'var(--info)'}">${esc(f.severity)}</span></td>
+    <td><code>${esc(f.rule_id)}</code></td><td>${esc((f.message||"").split("\n")[0].slice(0,120))}</td>
+    <td><code>${esc(f.file)}:${esc(f.line)}</code></td></tr>`).join("");
+  d.innerHTML=`<strong>Scan #${esc(s.id)}</strong> <span class="muted">${esc(s.created_at)} · ${esc(s.repo||"—")}</span>
+    <table style="margin-top:8px"><thead><tr><th>Severity</th><th>Rule</th><th>Finding</th><th>Location</th></tr></thead>
+    <tbody>${rows||'<tr><td colspan="4" class="muted">No findings.</td></tr>'}</tbody></table>`;
+  d.scrollIntoView({behavior:"smooth"});
+}
+$("#connect").onclick=()=>{KEY=$("#key").value.trim();if(!KEY)return;sessionStorage.setItem("ag_key",KEY);$("#key").value="";load();};
+$("#key").addEventListener("keydown",e=>{if(e.key==="Enter")$("#connect").click();});
+$("#logout").onclick=()=>{sessionStorage.removeItem("ag_key");location.reload();};
+if(KEY)load();
+</script>
+</body>
+</html>
diff --git a/tests/test_appguardrail.py b/tests/test_appguardrail.py
index ec00d70..ce10582 100644
--- a/tests/test_appguardrail.py
+++ b/tests/test_appguardrail.py
@@ -1471,9 +1471,12 @@ def test_cmd_monitor_installs_github_actions_workflow(tmp_path, monkeypatch, cap
     workflow_text = workflow.read_text()
     assert workflow.exists()
     assert "name: AppGuardrail Monitor" in workflow_text
-    assert "appguardrail scan --sarif appguardrail.sarif ." in workflow_text
+    assert "appguardrail scan --sarif appguardrail.sarif $PUSH ." in workflow_text
     assert "github/codeql-action/upload-sarif" in workflow_text
     assert "security-events: write" in workflow_text
+    # optional control-plane push wiring
+    assert "APPGUARDRAIL_CONTROL_PLANE_URL" in workflow_text
+    assert "APPGUARDRAIL_API_KEY" in workflow_text
     assert "appguardrail-monitor.yml" in capsys.readouterr().out
 
 
diff --git a/tests/test_controlplane.py b/tests/test_controlplane.py
new file mode 100644
index 0000000..a553100
--- /dev/null
+++ b/tests/test_controlplane.py
@@ -0,0 +1,367 @@
+"""Tests for the multi-tenant control-plane store + API."""
+
+import json
+import threading
+import urllib.error
+import urllib.request
+from contextlib import closing
+
+import pytest
+
+from appguardrail_core.controlplane import (
+    _is_slack_webhook,
+    _send_alert,
+    _slack_blocks,
+    add_scan,
+    connect,
+    create_key,
+    create_org,
+    get_scan,
+    has_role,
+    list_scans,
+    make_control_plane_server,
+    org_for_key,
+    role_for_key,
+    scan_trend,
+    set_webhook,
+)
+
+FINDINGS = [
+    {"severity": "CRITICAL", "rule_id": "x", "context": "app-code"},
+    {"severity": "INFO", "rule_id": "y", "context": "doc"},
+]
+
+
+# ---- store ----
+
+def test_org_key_auth():
+    conn = connect(":memory:")
+    oid, key = create_org(conn, "Acme")
+    assert org_for_key(conn, key) == oid
+    assert org_for_key(conn, "agk_wrong") is None
+    assert org_for_key(conn, "") is None
+
+
+def test_add_and_list_scan_counts():
+    conn = connect(":memory:")
+    oid, _ = create_org(conn, "Acme")
+    s = add_scan(conn, oid, FINDINGS, repo="acme/app", commit_sha="abc")
+    assert s["total"] == 2 and s["deploy_blocking"] == 1
+    listed = list_scans(conn, oid)
+    assert len(listed) == 1 and listed[0]["repo"] == "acme/app"
+    full = get_scan(conn, oid, s["id"])
+    assert len(full["findings"]) == 2
+
+
+def test_tenant_isolation():
+    conn = connect(":memory:")
+    a, _ = create_org(conn, "Acme")
+    b, _ = create_org(conn, "Beta")
+    s = add_scan(conn, a, FINDINGS)
+    assert get_scan(conn, b, s["id"]) is None  # cross-tenant read blocked
+    assert list_scans(conn, b) == []
+
+
+# ---- API ----
+
+def _serve(server):
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+
+
+def _req(method, url, key=None, body=None):
+    data = json.dumps(body).encode() if body is not None else None
+    r = urllib.request.Request(url, data=data, method=method)
+    if key:
+        r.add_header("Authorization", f"Bearer {key}")
+    if data:
+        r.add_header("Content-Type", "application/json")
+    with closing(urllib.request.urlopen(r, timeout=5)) as resp:
+        return resp.status, json.loads(resp.read())
+
+
+@pytest.fixture()
+def server(tmp_path):
+    db = str(tmp_path / "cp.db")
+    conn = connect(db)
+    _oid, key = create_org(conn, "Acme")
+    conn.close()
+    srv = make_control_plane_server("127.0.0.1", 0, db)
+    _serve(srv)
+    port = srv.server_address[1]
+    yield f"http://127.0.0.1:{port}", key
+    srv.shutdown()
+    srv.server_close()
+
+
+def test_health_no_auth(server):
+    base, _ = server
+    status, body = _req("GET", f"{base}/api/v1/health")
+    assert status == 200 and body["status"] == "ok"
+
+
+def test_ingest_and_history(server):
+    base, key = server
+    status, summary = _req(
+        "POST", f"{base}/api/v1/scans", key,
+        {"repo": "acme/app", "commit": "abc", "findings": FINDINGS},
+    )
+    assert status == 201 and summary["deploy_blocking"] == 1
+    status, listing = _req("GET", f"{base}/api/v1/scans", key)
+    assert status == 200 and len(listing["scans"]) == 1
+    scan_id = listing["scans"][0]["id"]
+    status, full = _req("GET", f"{base}/api/v1/scans/{scan_id}", key)
+    assert status == 200 and len(full["findings"]) == 2
+
+
+def test_auth_required(server):
+    base, _ = server
+    for key in (None, "agk_wrong"):
+        with pytest.raises(urllib.error.HTTPError) as exc:
+            _req("GET", f"{base}/api/v1/scans", key)
+        assert exc.value.code == 401
+
+
+def test_bad_body_400(server):
+    base, key = server
+    with pytest.raises(urllib.error.HTTPError) as exc:
+        _req("POST", f"{base}/api/v1/scans", key, {"findings": "not-a-list"})
+    assert exc.value.code == 400
+
+
+def test_console_served_at_root(server):
+    base, _ = server
+    with closing(urllib.request.urlopen(base + "/", timeout=5)) as resp:
+        body = resp.read()
+    assert resp.status == 200
+    assert b"AppGuardrail Console" in body  # served the org console HTML
+
+
+def test_drift_new_blocking():
+    conn = connect(":memory:")
+    oid, _ = create_org(conn, "Acme")
+    crit = {"severity": "CRITICAL", "rule_id": "secret", "file": "a.ts", "line": 3,
+            "message": "hardcoded key", "context": "app-code"}
+    other = {"severity": "HIGH", "rule_id": "rls", "file": "b.sql", "line": 1,
+             "message": "RLS off", "context": "app-code"}
+    s1 = add_scan(conn, oid, [crit], repo="acme/app")
+    assert s1["new_blocking"] == 1  # first scan: all blocking are new
+    s2 = add_scan(conn, oid, [{**crit, "line": 9}], repo="acme/app")
+    assert s2["new_blocking"] == 0  # same finding, line moved -> not new
+    s3 = add_scan(conn, oid, [crit, other], repo="acme/app")
+    assert s3["deploy_blocking"] == 2 and s3["new_blocking"] == 1  # one newly introduced
+    s4 = add_scan(conn, oid, [crit], repo="acme/other")
+    assert s4["new_blocking"] == 1  # different repo -> independent baseline
+
+
+def test_webhook_alerts_on_drift(monkeypatch):
+    sent = []
+    monkeypatch.setattr("appguardrail_core.controlplane._send_alert", lambda url, payload, **kw: sent.append((url, payload, kw)) or True)
+    conn = connect(":memory:")
+    oid, _ = create_org(conn, "Acme")
+    set_webhook(conn, oid, "http://hook.example/x")
+    crit = {"severity": "CRITICAL", "rule_id": "s", "file": "a.ts", "line": 1,
+            "message": "k", "context": "app-code"}
+    add_scan(conn, oid, [crit], repo="acme/app")             # 1 new -> alert
+    add_scan(conn, oid, [{**crit, "line": 9}], repo="acme/app")  # 0 new -> no alert
+    assert len(sent) == 1
+    url, payload, kw = sent[0]
+    assert url == "http://hook.example/x"
+    assert payload["event"] == "drift.new_blocking" and payload["new_blocking"] == 1
+    # add_scan hands the Slack renderer the org name + the new findings.
+    assert kw["org_name"] == "Acme"
+    assert [f["rule_id"] for f in kw["new_findings"]] == ["s"]
+
+
+def test_no_webhook_no_alert(monkeypatch):
+    sent = []
+    monkeypatch.setattr("appguardrail_core.controlplane._send_alert", lambda url, payload, **kw: sent.append(1))
+    conn = connect(":memory:")
+    oid, _ = create_org(conn, "Acme")  # no webhook set
+    add_scan(conn, oid, [{"severity": "CRITICAL", "rule_id": "s", "file": "a", "line": 1, "context": "app-code"}])
+    assert sent == []  # no webhook -> no delivery attempt
+
+
+def test_api_set_webhook(server):
+    base, key = server
+    status, body = _req("POST", f"{base}/api/v1/webhook", key, {"url": "http://hook.example/y"})
+    assert status == 200 and body["webhook_url"] == "http://hook.example/y"
+
+
+def test_roles_and_key_scoping():
+    conn = connect(":memory:")
+    oid, owner_key = create_org(conn, "Acme")
+    # bootstrap key is an owner
+    assert role_for_key(conn, owner_key) == (oid, "owner")
+    _, mk = create_key(conn, oid, "member", "ci")
+    _, vk = create_key(conn, oid, "viewer")
+    assert role_for_key(conn, mk) == (oid, "member")
+    assert role_for_key(conn, vk) == (oid, "viewer")
+    assert role_for_key(conn, "agk_bad") is None
+    assert has_role("owner", "member") and has_role("member", "member")
+    assert not has_role("viewer", "member") and not has_role("member", "owner")
+
+
+def test_api_role_enforcement(server):
+    base, owner_key = server
+    # forge member + viewer keys directly in the fixture DB via a fresh connect
+    # (the server shares the same file); simplest: create via the owner API.
+    status, mk = _req("POST", f"{base}/api/v1/keys", owner_key, {"role": "member"})
+    assert status == 201 and mk["role"] == "member"
+    status, vk = _req("POST", f"{base}/api/v1/keys", owner_key, {"role": "viewer"})
+    assert status == 201 and vk["role"] == "viewer"
+    member, viewer = mk["api_key"], vk["api_key"]
+
+    F = {"findings": [{"severity": "CRITICAL", "rule_id": "x", "context": "app-code"}]}
+    # viewer: read yes, ingest no, keys no
+    assert _req("GET", f"{base}/api/v1/scans", viewer)[0] == 200
+    for method, path, body in [("POST", "/api/v1/scans", F),
+                               ("POST", "/api/v1/keys", {"role": "member"}),
+                               ("POST", "/api/v1/webhook", {"url": "http://x"})]:
+        with pytest.raises(urllib.error.HTTPError) as e:
+            _req(method, f"{base}{path}", viewer, body)
+        assert e.value.code == 403
+    # member: ingest yes, keys/webhook no
+    assert _req("POST", f"{base}/api/v1/scans", member, F)[0] == 201
+    with pytest.raises(urllib.error.HTTPError) as e:
+        _req("POST", f"{base}/api/v1/webhook", member, {"url": "http://x"})
+    assert e.value.code == 403
+    # owner: all yes
+    assert _req("POST", f"{base}/api/v1/webhook", owner_key, {"url": "http://x"})[0] == 200
+
+
+def test_pagination_and_trend():
+    conn = connect(":memory:")
+    oid, _ = create_org(conn, "Acme")
+    for i in range(5):
+        add_scan(conn, oid, [{"severity": "CRITICAL", "rule_id": f"r{i}",
+                              "file": "a", "line": i, "message": "m", "context": "app-code"}],
+                 repo="acme/app")
+    # list is newest-first; offset skips
+    p1 = list_scans(conn, oid, limit=2, offset=0)
+    p2 = list_scans(conn, oid, limit=2, offset=2)
+    assert [s["id"] for s in p1] == [5, 4]
+    assert [s["id"] for s in p2] == [3, 2]
+    # trend is oldest-first with the right length
+    t = scan_trend(conn, oid, limit=10)
+    assert len(t) == 5
+    assert t[0]["created_at"] <= t[-1]["created_at"]
+    assert all("deploy_blocking" in x and "new_blocking" in x for x in t)
+
+
+def test_api_pagination_and_trend(server):
+    base, key = server
+    for i in range(4):
+        _req("POST", f"{base}/api/v1/scans", key, {"repo": "r", "findings": []})
+    _, page = _req("GET", f"{base}/api/v1/scans?limit=2&offset=1", key)
+    assert len(page["scans"]) == 2
+    _, tr = _req("GET", f"{base}/api/v1/scans/trend?limit=3", key)
+    assert len(tr["trend"]) == 3
+
+
+# ---- Slack-formatted drift alert ----
+
+def test_is_slack_webhook():
+    assert _is_slack_webhook("https://hooks.slack.com/services/T/B/xyz")
+    assert not _is_slack_webhook("https://hooks.slack.com.evil.example/x")  # host must match
+    assert not _is_slack_webhook("https://hook.example/x")
+    assert not _is_slack_webhook("not a url")
+
+
+def test_slack_blocks_shape_and_content():
+    payload = {"org_id": 7, "scan_id": 42, "repo": "acme/app", "new_blocking": 2}
+    findings = [
+        {"rule_id": "hardcoded-secret", "file": "src/a.ts"},
+        {"rule_id": "sql-injection", "file": "src/b.py"},
+    ]
+    body = _slack_blocks("Acme", payload, findings)
+    assert "blocks" in body and body["blocks"][0]["type"] == "header"
+    text = json.dumps(body)
+    assert "Acme" in text  # org name rendered
+    assert "2 new deploy-blocking findings" in body["blocks"][0]["text"]["text"]  # count in header
+    assert "hardcoded-secret" in text and "src/a.ts" in text  # top rule_id + file listed
+    assert "#42" in text  # scan id
+
+
+def test_slack_blocks_caps_and_escapes():
+    findings = [{"rule_id": f"r{i}", "file": f"f{i}"} for i in range(8)]
+    payload = {"org_id": 1, "scan_id": 1, "repo": "r", "new_blocking": 8}
+    body = _slack_blocks("Ben & <Co>", payload, findings, top=5)
+    detail = body["blocks"][-1]["text"]["text"]
+    assert detail.count("• `r") == 5  # only top 5 rules listed
+    assert "+3 more" in detail       # overflow line
+    assert "Ben &amp; &lt;Co&gt;" in json.dumps(body)  # org name escaped for Slack
+
+
+def test_send_alert_slack_vs_generic(monkeypatch):
+    posted = {}
+
+    def _fake_urlopen(req, timeout=None):
+        posted["url"] = req.full_url
+        posted["body"] = json.loads(req.data.decode())
+        class _R:  # minimal stand-in, urlopen result is ignored
+            pass
+        return _R()
+
+    monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
+    generic = {"event": "drift.new_blocking", "org_id": 3, "scan_id": 9,
+               "repo": "acme/app", "new_blocking": 1, "deploy_blocking": 1}
+    findings = [{"rule_id": "secret", "file": "a.ts"}]
+
+    # Slack URL -> Block Kit payload
+    assert _send_alert("https://hooks.slack.com/services/x", generic,
+                       org_name="Acme", new_findings=findings) is True
+    assert "blocks" in posted["body"]
+    assert "Acme" in json.dumps(posted["body"])
+    assert "1 new deploy-blocking finding" in posted["body"]["blocks"][0]["text"]["text"]
+
+    # Generic URL -> untouched original payload (backward compatible)
+    assert _send_alert("https://hook.example/x", generic,
+                       org_name="Acme", new_findings=findings) is True
+    assert posted["body"] == generic
+    assert "blocks" not in posted["body"]
+
+
+# ---- API hardening: body cap + query clamps ----
+
+def test_negative_and_huge_limit_clamped(server):
+    base, key = server
+    for _ in range(3):
+        _req("POST", f"{base}/api/v1/scans", key, {"repo": "r", "findings": []})
+    # limit=-1 means UNBOUNDED in sqlite — must be clamped, not passed through
+    _, page = _req("GET", f"{base}/api/v1/scans?limit=-1", key)
+    assert 1 <= len(page["scans"]) <= 1000
+    _, page2 = _req("GET", f"{base}/api/v1/scans?limit=999999&offset=-5", key)
+    assert len(page2["scans"]) <= 1000  # hi clamp; negative offset -> 0
+
+
+def test_oversized_body_rejected(server):
+    import http.client
+    from urllib.parse import urlparse as _u
+    base, key = server
+    u = _u(base)
+    conn = http.client.HTTPConnection(u.hostname, u.port, timeout=10)
+    # Claim an over-cap body; server must 400 without reading it all.
+    conn.putrequest("POST", "/api/v1/scans")
+    conn.putheader("Authorization", f"Bearer {key}")
+    conn.putheader("Content-Type", "application/json")
+    conn.putheader("Content-Length", str(50 * 1024 * 1024))
+    conn.endheaders()
+    conn.send(b"{")  # send a byte so the server can respond
+    resp = conn.getresponse()
+    assert resp.status == 400
+    conn.close()
+
+
+def test_negative_content_length_rejected(server):
+    import http.client
+    from urllib.parse import urlparse as _u
+    base, key = server
+    u = _u(base)
+    conn = http.client.HTTPConnection(u.hostname, u.port, timeout=10)
+    conn.putrequest("POST", "/api/v1/scans")
+    conn.putheader("Authorization", f"Bearer {key}")
+    conn.putheader("Content-Length", "-1")
+    conn.endheaders()
+    resp = conn.getresponse()
+    assert resp.status == 400
+    conn.close()