diff --git a/pipeline/tool_registry.py b/pipeline/tool_registry.py
index 694a002..e568768 100644
--- a/pipeline/tool_registry.py
+++ b/pipeline/tool_registry.py
@@ -55,6 +55,13 @@ def _github_check() -> bool:
     return is_authenticated()
 
 
+def _whatsapp_check() -> bool:
+    # No OAuth — "connected" means the local GoWA REST server has a paired device.
+    # Short timeout so an absent/idle server doesn't stall schema building.
+    from pipeline.tools_whatsapp import has_paired_device
+    return has_paired_device(timeout=3.0)
+
+
 # ── toolset 정의 ──────────────────────────────────────────────────────────────
 # hermes-agent toolsets.py의 TOOLSETS 패턴 — 서비스(toolset) 단위로 도구를 묶고
 # check_fn 하나로 가용성을 판정한다.
@@ -118,6 +125,14 @@ def _github_check() -> bool:
         "check_fn": _github_check,
         "connect_hint": "설정 → 워크스페이스에서 GitHub PAT 를 연결하라 (GITHUB_PERSONAL_ACCESS_TOKEN).",
     },
+    "whatsapp": {
+        "description": "WhatsApp (채팅 목록/메시지 읽기·전송, 로컬 GoWA 서버 경유)",
+        "tools": [
+            "whatsapp_list_chats", "whatsapp_read_messages", "whatsapp_send_message",
+        ],
+        "check_fn": _whatsapp_check,
+        "connect_hint": "GoWA 서버 기동 + QR 페어링 필요 (whatsapp rest --port 3777, VEGA_WHATSAPP_GOWA_URL).",
+    },
 }
 
 _TOOL_TO_TOOLSET: dict[str, str] = {
diff --git a/pipeline/tools.py b/pipeline/tools.py
index 352acf4..7cfaa92 100644
--- a/pipeline/tools.py
+++ b/pipeline/tools.py
@@ -92,6 +92,8 @@ def set_ce_mode(active: bool) -> None:
     # GitHub
     "github_list_issues", "github_get_issue", "github_list_pulls",
     "github_get_pull", "github_search_code", "github_read_file",
+    # WhatsApp (read only — send is write, excluded)
+    "whatsapp_list_chats", "whatsapp_read_messages",
 })
 
 
@@ -1098,6 +1100,12 @@ def get_schemas_for_mode(
 from pipeline.tools_superthread import SUPERTHREAD_TOOL_SCHEMAS, SUPERTHREAD_TOOL_FUNCTIONS
 from pipeline.tools_airtable import AIRTABLE_TOOL_SCHEMAS, AIRTABLE_TOOL_FUNCTIONS
 from pipeline.tools_github import GITHUB_TOOL_SCHEMAS, GITHUB_TOOL_FUNCTIONS
+# WhatsApp wraps a local GoWA REST server — optional (module skipped if unavailable).
+try:
+    from pipeline.tools_whatsapp import WHATSAPP_TOOL_SCHEMAS, WHATSAPP_TOOL_FUNCTIONS
+except Exception:
+    WHATSAPP_TOOL_SCHEMAS: list[dict] = []
+    WHATSAPP_TOOL_FUNCTIONS: dict[str, Any] = {}
 
 TOOL_SCHEMAS.extend(MEMORY_TOOL_SCHEMAS)
 TOOL_SCHEMAS.extend(SESSION_TOOL_SCHEMAS)
@@ -1107,6 +1115,7 @@ def get_schemas_for_mode(
 TOOL_SCHEMAS.extend(SUPERTHREAD_TOOL_SCHEMAS)
 TOOL_SCHEMAS.extend(AIRTABLE_TOOL_SCHEMAS)
 TOOL_SCHEMAS.extend(GITHUB_TOOL_SCHEMAS)
+TOOL_SCHEMAS.extend(WHATSAPP_TOOL_SCHEMAS)
 
 # vega-agent: 네이티브 linear_* 도구는 pipeline.linear_client(개인 VEGA 전용, 여기 없음)에
 # 의존한다. 모듈이 없으면 호출 시 무조건 실패하고 self_improve 가 폭주하므로,
@@ -1909,6 +1918,7 @@ def mcp_reload() -> dict:
 TOOL_FUNCTIONS.update(SUPERTHREAD_TOOL_FUNCTIONS)
 TOOL_FUNCTIONS.update(AIRTABLE_TOOL_FUNCTIONS)
 TOOL_FUNCTIONS.update(GITHUB_TOOL_FUNCTIONS)
+TOOL_FUNCTIONS.update(WHATSAPP_TOOL_FUNCTIONS)
 
 
 def dispatch_tool(name: str, arguments: dict) -> str:
diff --git a/pipeline/tools_whatsapp.py b/pipeline/tools_whatsapp.py
new file mode 100644
index 0000000..c4bfdbd
--- /dev/null
+++ b/pipeline/tools_whatsapp.py
@@ -0,0 +1,220 @@
+# Created: 2026-07-02
+# Purpose: WhatsApp native tools — wraps a local GoWA (go-whatsapp-web-multidevice) REST server (INT-2323)
+# Dependencies: stdlib only (urllib). No OAuth — auth is the locally paired GoWA device.
+# Test Status: tests/test_whatsapp_int2323.py
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.parse
+import urllib.request
+from typing import Any
+
+_DEFAULT_URL = "http://localhost:3777"
+
+# Resolved GoWA device_id cache (process lifetime). v8 multi-device requires X-Device-Id
+# on every request. Reset via reset_device_cache() (used by tests / after re-pairing).
+_device_cache: str | None = None
+
+
+def _base_url() -> str:
+    return (os.getenv("VEGA_WHATSAPP_GOWA_URL") or _DEFAULT_URL).rstrip("/")
+
+
+def reset_device_cache() -> None:
+    """Drop the cached device_id — forces re-resolution on next call."""
+    global _device_cache
+    _device_cache = None
+
+
+# ── HTTP (stdlib urllib) ──────────────────────────────────────────────────────
+
+def _get(path: str, params: dict | None = None, *, device_id: str | None = None,
+         timeout: float = 30.0) -> dict:
+    url = _base_url() + path
+    if params:
+        url += "?" + urllib.parse.urlencode(
+            {k: v for k, v in params.items() if v not in (None, "")}
+        )
+    headers = {}
+    if device_id:
+        headers["X-Device-Id"] = device_id
+    req = urllib.request.Request(url, headers=headers)
+    with urllib.request.urlopen(req, timeout=timeout) as r:
+        return json.loads(r.read().decode())
+
+
+def _post_json(path: str, body: dict, *, device_id: str | None = None,
+               timeout: float = 30.0) -> dict:
+    data = json.dumps(body).encode()
+    headers = {"Content-Type": "application/json"}
+    if device_id:
+        headers["X-Device-Id"] = device_id
+    req = urllib.request.Request(_base_url() + path, data=data, method="POST", headers=headers)
+    with urllib.request.urlopen(req, timeout=timeout) as r:
+        return json.loads(r.read().decode())
+
+
+# ── Defensive parsing ─────────────────────────────────────────────────────────
+# GoWA responses are inconsistent: `results` is a dict for chats (results.data)
+# but a plain list for /app/devices. Handle both shapes.
+
+def _rows(results: Any) -> list:
+    if isinstance(results, dict):
+        data = results.get("data")
+        return data if isinstance(data, list) else []
+    if isinstance(results, list):
+        return results
+    return []
+
+
+def _devices() -> list:
+    data = _get("/app/devices", timeout=30.0)
+    return _rows(data.get("results"))
+
+
+def _resolve_device() -> str:
+    """Return the GoWA device_id (cached). Raise if the server is unreachable or unpaired."""
+    global _device_cache
+    if _device_cache:
+        return _device_cache
+    try:
+        devices = _devices()
+    except Exception as e:
+        raise RuntimeError(
+            "WhatsApp GoWA 서버에 연결할 수 없습니다 — GoWA REST 서버를 기동하세요 "
+            f"(whatsapp rest --port …, {_base_url()}). 원인: {e}"
+        ) from e
+    if not devices:
+        raise RuntimeError("WhatsApp GoWA 미연결 — QR 페어링 필요")
+    dev = devices[0].get("device") or devices[0].get("jid")
+    if not dev:
+        raise RuntimeError("WhatsApp GoWA 미연결 — QR 페어링 필요")
+    _device_cache = str(dev)
+    return _device_cache
+
+
+def has_paired_device(timeout: float = 3.0) -> bool:
+    """Connectivity/pairing probe for the toolset gate. Never raises."""
+    try:
+        data = _get("/app/devices", timeout=timeout)
+    except Exception:
+        return False
+    return bool(_rows(data.get("results")))
+
+
+def _msg_text(m: dict) -> str:
+    """Best-effort text extraction — GoWA message shape varies by message type."""
+    for k in ("text", "message", "body", "content", "caption"):
+        v = m.get(k)
+        if isinstance(v, str) and v:
+            return v[:4000]
+        if isinstance(v, dict):
+            for kk in ("text", "conversation", "caption", "body"):
+                vv = v.get(kk)
+                if isinstance(vv, str) and vv:
+                    return vv[:4000]
+    return ""
+
+
+# ── Tools ─────────────────────────────────────────────────────────────────────
+
+def whatsapp_list_chats(limit: int = 20) -> list[dict]:
+    """List recent WhatsApp chats (jid/name/last_message_time)."""
+    dev = _resolve_device()
+    data = _get("/chats", {"limit": int(limit)}, device_id=dev)
+    out = []
+    for c in _rows(data.get("results")):
+        out.append({
+            "jid": c.get("jid"),
+            "name": c.get("name"),
+            "last_message_time": c.get("last_message_time"),
+            "archived": c.get("archived"),
+        })
+    return out
+
+
+def whatsapp_read_messages(chat_jid: str, limit: int = 20) -> list[dict]:
+    """Read recent messages of a specific chat. chat_jid from whatsapp_list_chats."""
+    dev = _resolve_device()
+    data = _get(f"/chat/{urllib.parse.quote(chat_jid, safe='@.')}/messages",
+                {"limit": int(limit)}, device_id=dev)
+    out = []
+    for m in _rows(data.get("results")):
+        out.append({
+            "id": m.get("id") or m.get("message_id"),
+            "sender": m.get("sender") or m.get("from") or m.get("pushname"),
+            "timestamp": m.get("timestamp") or m.get("message_time"),
+            "from_me": m.get("from_me", m.get("is_from_me")),
+            "text": _msg_text(m),
+        })
+    return out
+
+
+def whatsapp_send_message(phone: str, message: str) -> dict:
+    """Send a WhatsApp message. phone is a number (auto-suffixed @s.whatsapp.net)
+    or a full jid (individual @s.whatsapp.net / group @g.us — kept as-is)."""
+    dev = _resolve_device()
+    target = phone if "@" in phone else f"{phone}@s.whatsapp.net"
+    data = _post_json("/send/message", {"phone": target, "message": message}, device_id=dev)
+    return {
+        "ok": data.get("code") == "SUCCESS",
+        "code": data.get("code"),
+        "phone": target,
+        "results": data.get("results"),
+    }
+
+
+WHATSAPP_TOOL_SCHEMAS: list[dict] = [
+    {
+        "type": "function",
+        "name": "whatsapp_send_message",
+        "description": (
+            "WhatsApp 메시지를 직접 전송한다(사용자 명의, 로컬 GoWA 서버 경유). "
+            "정리한 내용을 복붙 안내하지 말고 이 도구로 바로 보낸다. "
+            "phone 은 국가코드 포함 번호(예: 821012345678) — @s.whatsapp.net 은 자동 부착된다. "
+            "그룹은 whatsapp_list_chats 의 jid(…@g.us)를 그대로 넣는다. "
+            "(GoWA 서버 미연결/미페어링 시 재연결 안내.)"
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "phone": {"type": "string", "description": "수신 번호(국가코드 포함) 또는 전체 jid(개인 …@s.whatsapp.net / 그룹 …@g.us)"},
+                "message": {"type": "string", "description": "보낼 메시지 본문"},
+            },
+            "required": ["phone", "message"],
+        },
+    },
+    {
+        "type": "function",
+        "name": "whatsapp_list_chats",
+        "description": "WhatsApp 최근 채팅 목록을 조회한다. 각 항목의 jid 로 whatsapp_read_messages·whatsapp_send_message 를 호출한다.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "limit": {"type": "integer", "default": 20, "description": "최대 채팅 수"},
+            },
+            "required": [],
+        },
+    },
+    {
+        "type": "function",
+        "name": "whatsapp_read_messages",
+        "description": "특정 WhatsApp 채팅의 최근 메시지를 읽는다. chat_jid 는 whatsapp_list_chats 결과의 jid.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "chat_jid": {"type": "string", "description": "채팅 jid (…@s.whatsapp.net / …@g.us)"},
+                "limit": {"type": "integer", "default": 20, "description": "메시지 수"},
+            },
+            "required": ["chat_jid"],
+        },
+    },
+]
+
+WHATSAPP_TOOL_FUNCTIONS: dict[str, Any] = {
+    "whatsapp_list_chats": whatsapp_list_chats,
+    "whatsapp_read_messages": whatsapp_read_messages,
+    "whatsapp_send_message": whatsapp_send_message,
+}
diff --git a/pipeline/whatsapp_sidecar.py b/pipeline/whatsapp_sidecar.py
new file mode 100644
index 0000000..752b36c
--- /dev/null
+++ b/pipeline/whatsapp_sidecar.py
@@ -0,0 +1,103 @@
+# Created: 2026-07-02
+# Purpose: Manage the GoWA (go-whatsapp-web-multidevice) REST server as an opt-in
+#   VEGA-managed sidecar process. Started/stopped from web/server.py lifespan.
+#   The whatsapp_* tools (pipeline/tools_whatsapp.py) talk to this server over REST.
+# Dependencies: stdlib only (subprocess, urllib), pipeline.data_paths
+# Test Status: tests/test_whatsapp_sidecar.py
+
+from __future__ import annotations
+
+import os
+import subprocess
+import urllib.request
+from pathlib import Path
+
+# Opt-in: sidecar only starts when this env flag is truthy. Off by default so the
+# backend never spawns an unofficial-protocol WhatsApp server unless the user asks.
+_OPT_IN_ENV = "VEGA_WHATSAPP_SIDECAR"
+# Port must match tools_whatsapp default (VEGA_WHATSAPP_GOWA_URL) — keep in sync.
+_DEFAULT_PORT = 3777
+
+_proc: subprocess.Popen | None = None
+
+
+def is_enabled() -> bool:
+    """True if the GoWA sidecar is opted in via env (1/true/yes/on)."""
+    return os.getenv(_OPT_IN_ENV, "").strip().lower() in ("1", "true", "yes", "on")
+
+
+def _binary_path() -> Path:
+    """GoWA binary location. Override with VEGA_WHATSAPP_GOWA_BIN; else data_dir default."""
+    override = os.getenv("VEGA_WHATSAPP_GOWA_BIN")
+    if override:
+        return Path(override).expanduser()
+    try:
+        from pipeline.data_paths import data_dir
+        base = Path(data_dir())
+    except Exception:
+        base = Path(os.getenv("VEGA_DATA_DIR", ".vega"))
+    return base / "whatsapp-gowa" / "whatsapp"
+
+
+def _port() -> int:
+    try:
+        return int(os.getenv("VEGA_WHATSAPP_GOWA_PORT", str(_DEFAULT_PORT)))
+    except ValueError:
+        return _DEFAULT_PORT
+
+
+def _already_serving(port: int, timeout: float = 1.5) -> bool:
+    """True if something already answers on the GoWA port (manual start / prior run)."""
+    try:
+        with urllib.request.urlopen(f"http://localhost:{port}/", timeout=timeout):
+            return True
+    except Exception:
+        return False
+
+
+def start() -> dict:
+    """Start the GoWA sidecar if opted in and not already running.
+
+    Returns a status dict — never raises (lifespan must not fail on sidecar issues).
+    """
+    global _proc
+    if not is_enabled():
+        return {"started": False, "reason": "not opted in"}
+    port = _port()
+    if _already_serving(port):
+        return {"started": False, "reason": f"already serving on {port}"}
+    binary = _binary_path()
+    if not binary.is_file():
+        return {"started": False, "reason": f"binary not found: {binary}"}
+    if not os.access(binary, os.X_OK):
+        return {"started": False, "reason": f"binary not executable: {binary}"}
+    try:
+        # rest mode on the shared port; own process group so we can clean it up.
+        _proc = subprocess.Popen(
+            [str(binary), "rest", "--port", str(port)],
+            cwd=str(binary.parent),
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            start_new_session=True,
+        )
+        return {"started": True, "pid": _proc.pid, "port": port}
+    except Exception as e:
+        _proc = None
+        return {"started": False, "reason": f"spawn failed: {e}"}
+
+
+def stop() -> None:
+    """Terminate the sidecar if VEGA started it (no-op for manually-run servers)."""
+    global _proc
+    if _proc is None:
+        return
+    try:
+        _proc.terminate()
+        try:
+            _proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            _proc.kill()
+    except Exception:
+        pass
+    finally:
+        _proc = None
diff --git a/tests/test_tools_workspace.py b/tests/test_tools_workspace.py
index c8e94e9..37289d9 100644
--- a/tests/test_tools_workspace.py
+++ b/tests/test_tools_workspace.py
@@ -329,12 +329,14 @@ def test_new_groups_respect_explicit_disable(tmp_path, monkeypatch):
     "linear_search_issues", "linear_list_issues",
     "airtable_list_records", "airtable_get_records",
     "github_search_code", "github_list_issues",
+    "whatsapp_list_chats", "whatsapp_read_messages",
 ]
 # light 에서 절대 노출되면 안 되는 워크스페이스 *쓰기/외부전송* 도구.
 _WORKSPACE_WRITE_NOT_IN_LIGHT = [
     "slack_send_message", "superthread_create_card", "gmail_send",
     "calendar_create_event", "linear_create_issue", "airtable_create_record",
     "github_create_issue", "docs_create", "slides_create",
+    "whatsapp_send_message",
 ]
 
 
diff --git a/tests/test_whatsapp_int2323.py b/tests/test_whatsapp_int2323.py
new file mode 100644
index 0000000..4639bb3
--- /dev/null
+++ b/tests/test_whatsapp_int2323.py
@@ -0,0 +1,199 @@
+# Created: 2026-07-02
+# Purpose: WhatsApp native tools (GoWA REST wrapper) — unit tests (INT-2323)
+# Dependencies: pipeline/tools_whatsapp.py, pipeline/tool_registry.py
+# Test Status: green (2026-07-02)
+
+from __future__ import annotations
+
+import io
+import json
+
+import pytest
+
+import pipeline.tools_whatsapp as wa
+
+
+class _FakeResp:
+    """Minimal context-manager stand-in for urllib.request.urlopen return value."""
+
+    def __init__(self, payload: dict):
+        self._buf = io.BytesIO(json.dumps(payload).encode())
+
+    def read(self):
+        return self._buf.read()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        return False
+
+
+def _install_urlopen(monkeypatch, handler):
+    """handler(req) -> dict payload. Also records the captured Request objects."""
+    calls: list = []
+
+    def _fake_urlopen(req, timeout=None, **kw):
+        calls.append(req)
+        return _FakeResp(handler(req))
+
+    monkeypatch.setattr("urllib.request.urlopen", _fake_urlopen)
+    return calls
+
+
+@pytest.fixture(autouse=True)
+def _reset_cache():
+    wa.reset_device_cache()
+    yield
+    wa.reset_device_cache()
+
+
+_DEVICES = {"results": [{"name": "iPhone", "device": "dev-1", "jid": "111@s.whatsapp.net"}]}
+
+
+# ── device resolution ─────────────────────────────────────────────────────────
+
+class TestDeviceResolve:
+    def test_resolves_and_caches(self, monkeypatch):
+        hits = {"devices": 0}
+
+        def handler(req):
+            if req.full_url.endswith("/app/devices"):
+                hits["devices"] += 1
+                return _DEVICES
+            raise AssertionError("unexpected url")
+
+        _install_urlopen(monkeypatch, handler)
+        assert wa._resolve_device() == "dev-1"
+        # second call is served from cache — no extra /app/devices hit
+        assert wa._resolve_device() == "dev-1"
+        assert hits["devices"] == 1
+
+    def test_no_device_raises(self, monkeypatch):
+        _install_urlopen(monkeypatch, lambda req: {"results": []})
+        with pytest.raises(RuntimeError, match="페어링"):
+            wa._resolve_device()
+
+    def test_has_paired_device_false_on_connection_error(self, monkeypatch):
+        def _boom(req, timeout=None, **kw):
+            raise OSError("connection refused")
+        monkeypatch.setattr("urllib.request.urlopen", _boom)
+        assert wa.has_paired_device() is False
+
+
+# ── list_chats ────────────────────────────────────────────────────────────────
+
+class TestListChats:
+    def test_parses_results_data(self, monkeypatch):
+        def handler(req):
+            if req.full_url.endswith("/app/devices"):
+                return _DEVICES
+            if "/chats" in req.full_url:
+                assert req.get_header("X-device-id") == "dev-1"
+                return {"results": {"data": [
+                    {"jid": "aaa@s.whatsapp.net", "name": "Alice",
+                     "last_message_time": "2026-07-02T10:00:00Z", "archived": False},
+                    {"jid": "grp@g.us", "name": "Team", "last_message_time": "x", "archived": True},
+                ], "pagination": {"cursor": "c"}}}
+            raise AssertionError(req.full_url)
+
+        _install_urlopen(monkeypatch, handler)
+        chats = wa.whatsapp_list_chats(limit=5)
+        assert [c["jid"] for c in chats] == ["aaa@s.whatsapp.net", "grp@g.us"]
+        assert chats[0]["name"] == "Alice"
+
+    def test_defensive_direct_array_results(self, monkeypatch):
+        # GoWA sometimes returns results as a plain array — must not crash
+        def handler(req):
+            if req.full_url.endswith("/app/devices"):
+                return _DEVICES
+            return {"results": [{"jid": "aaa@s.whatsapp.net", "name": "Alice"}]}
+
+        _install_urlopen(monkeypatch, handler)
+        chats = wa.whatsapp_list_chats()
+        assert chats[0]["jid"] == "aaa@s.whatsapp.net"
+
+
+# ── read_messages ─────────────────────────────────────────────────────────────
+
+class TestReadMessages:
+    def test_parses_messages(self, monkeypatch):
+        def handler(req):
+            if req.full_url.endswith("/app/devices"):
+                return _DEVICES
+            if "/messages" in req.full_url:
+                assert "/chat/aaa@s.whatsapp.net/messages" in req.full_url
+                return {"results": {"data": [
+                    {"id": "m1", "sender": "Alice", "timestamp": "t1", "text": "hi"},
+                    {"id": "m2", "from": "me", "message": {"conversation": "yo"}},
+                ]}}
+            raise AssertionError(req.full_url)
+
+        _install_urlopen(monkeypatch, handler)
+        msgs = wa.whatsapp_read_messages("aaa@s.whatsapp.net", limit=10)
+        assert msgs[0]["id"] == "m1"
+        assert msgs[0]["text"] == "hi"
+        assert msgs[1]["text"] == "yo"  # nested dict extraction
+
+
+# ── send_message ──────────────────────────────────────────────────────────────
+
+class TestSendMessage:
+    def test_auto_suffix_and_post_body(self, monkeypatch):
+        captured = {}
+
+        def handler(req):
+            if req.full_url.endswith("/app/devices"):
+                return _DEVICES
+            if req.full_url.endswith("/send/message"):
+                assert req.get_method() == "POST"
+                assert req.get_header("X-device-id") == "dev-1"
+                assert req.get_header("Content-type") == "application/json"
+                captured["body"] = json.loads(req.data.decode())
+                return {"code": "SUCCESS", "results": {"message_id": "M1", "status": "sent"}}
+            raise AssertionError(req.full_url)
+
+        _install_urlopen(monkeypatch, handler)
+        res = wa.whatsapp_send_message("821012345678", "hello")
+        assert captured["body"] == {"phone": "821012345678@s.whatsapp.net", "message": "hello"}
+        assert res["ok"] is True
+        assert res["phone"] == "821012345678@s.whatsapp.net"
+
+    def test_group_jid_kept_as_is(self, monkeypatch):
+        captured = {}
+
+        def handler(req):
+            if req.full_url.endswith("/app/devices"):
+                return _DEVICES
+            captured["body"] = json.loads(req.data.decode())
+            return {"code": "SUCCESS", "results": {}}
+
+        _install_urlopen(monkeypatch, handler)
+        wa.whatsapp_send_message("grp@g.us", "hi team")
+        assert captured["body"]["phone"] == "grp@g.us"  # no double suffix
+
+
+# ── registry wiring ───────────────────────────────────────────────────────────
+
+class TestRegistryWiring:
+    def test_toolset_registered(self):
+        import pipeline.tool_registry as reg
+        assert "whatsapp" in reg.WORKSPACE_TOOLSETS
+        for t in ("whatsapp_list_chats", "whatsapp_read_messages", "whatsapp_send_message"):
+            assert reg.toolset_of(t) == "whatsapp"
+
+    def test_schemas_and_functions_exposed(self):
+        import pipeline.tools as tools
+        names = {s["name"] for s in tools.TOOL_SCHEMAS}
+        for t in ("whatsapp_list_chats", "whatsapp_read_messages", "whatsapp_send_message"):
+            assert t in names
+            assert t in tools.TOOL_FUNCTIONS
+
+    def test_check_fn_gates_on_pairing(self, monkeypatch):
+        import pipeline.tool_registry as reg
+        reg.invalidate_check_fn_cache()
+        monkeypatch.setattr(wa, "has_paired_device", lambda timeout=3.0: False)
+        assert reg.is_toolset_available("whatsapp") is False
+        reg.invalidate_check_fn_cache()
+        monkeypatch.setattr(wa, "has_paired_device", lambda timeout=3.0: True)
+        assert reg.is_toolset_available("whatsapp") is True
diff --git a/tests/test_whatsapp_sidecar.py b/tests/test_whatsapp_sidecar.py
new file mode 100644
index 0000000..61934de
--- /dev/null
+++ b/tests/test_whatsapp_sidecar.py
@@ -0,0 +1,84 @@
+# Created: 2026-07-02
+# Purpose: INT-2323 — GoWA sidecar opt-in lifecycle (start gate, already-serving skip,
+#          binary checks, stop). No real process spawned (Popen mocked).
+
+from __future__ import annotations
+
+import pipeline.whatsapp_sidecar as sc
+
+
+def _reset(monkeypatch, **env):
+    sc._proc = None
+    for k in ("VEGA_WHATSAPP_SIDECAR", "VEGA_WHATSAPP_GOWA_BIN", "VEGA_WHATSAPP_GOWA_PORT"):
+        monkeypatch.delenv(k, raising=False)
+    for k, v in env.items():
+        monkeypatch.setenv(k, v)
+
+
+def test_disabled_by_default(monkeypatch):
+    _reset(monkeypatch)
+    assert sc.is_enabled() is False
+    assert sc.start()["started"] is False
+    assert sc.start()["reason"] == "not opted in"
+
+
+def test_enabled_flag_variants(monkeypatch):
+    for val in ("1", "true", "YES", "on"):
+        _reset(monkeypatch, VEGA_WHATSAPP_SIDECAR=val)
+        assert sc.is_enabled() is True
+    for val in ("0", "false", "", "no"):
+        _reset(monkeypatch, VEGA_WHATSAPP_SIDECAR=val)
+        assert sc.is_enabled() is False
+
+
+def test_skip_when_already_serving(monkeypatch):
+    _reset(monkeypatch, VEGA_WHATSAPP_SIDECAR="1")
+    monkeypatch.setattr(sc, "_already_serving", lambda port, timeout=1.5: True)
+    r = sc.start()
+    assert r["started"] is False and "already serving" in r["reason"]
+
+
+def test_binary_missing(monkeypatch, tmp_path):
+    _reset(monkeypatch, VEGA_WHATSAPP_SIDECAR="1", VEGA_WHATSAPP_GOWA_BIN=str(tmp_path / "nope"))
+    monkeypatch.setattr(sc, "_already_serving", lambda port, timeout=1.5: False)
+    r = sc.start()
+    assert r["started"] is False and "binary not found" in r["reason"]
+
+
+def test_start_spawns_when_enabled(monkeypatch, tmp_path):
+    binary = tmp_path / "whatsapp"
+    binary.write_text("#!/bin/sh\n")
+    binary.chmod(0o755)
+    _reset(monkeypatch, VEGA_WHATSAPP_SIDECAR="1", VEGA_WHATSAPP_GOWA_BIN=str(binary),
+           VEGA_WHATSAPP_GOWA_PORT="3777")
+    monkeypatch.setattr(sc, "_already_serving", lambda port, timeout=1.5: False)
+
+    calls = {}
+
+    class _FakeProc:
+        pid = 4242
+        def terminate(self): calls["terminated"] = True
+        def wait(self, timeout=None): return 0
+        def kill(self): calls["killed"] = True
+
+    def fake_popen(args, **kw):
+        calls["args"] = args
+        return _FakeProc()
+
+    monkeypatch.setattr(sc.subprocess, "Popen", fake_popen)
+    r = sc.start()
+    assert r["started"] is True and r["pid"] == 4242 and r["port"] == 3777
+    # correct invocation: <binary> rest --port 3777
+    assert calls["args"][0] == str(binary)
+    assert calls["args"][1] == "rest" and "3777" in calls["args"]
+
+    # stop terminates the tracked process
+    sc.stop()
+    assert calls.get("terminated") is True
+    assert sc._proc is None
+
+
+def test_stop_noop_when_not_started(monkeypatch):
+    _reset(monkeypatch)
+    sc.stop()  # must not raise
+    assert sc._proc is None
diff --git a/web/server.py b/web/server.py
index 433fe94..1b4a33c 100644
--- a/web/server.py
+++ b/web/server.py
@@ -181,6 +181,19 @@ async def _warmup_local_llm():
     # Cron loop — run scheduled prompts at their due time in the background (INT-1407)
     _cron_task = asyncio.create_task(_cron_loop())
 
+    # WhatsApp GoWA sidecar — opt-in (VEGA_WHATSAPP_SIDECAR). Off by default;
+    # when enabled, VEGA manages the GoWA REST server lifecycle so whatsapp_*
+    # tools work without a manual start (INT-2323). Never fails startup.
+    try:
+        from pipeline import whatsapp_sidecar
+        _wa = whatsapp_sidecar.start()
+        if _wa.get("started"):
+            print(f"[WhatsApp] GoWA sidecar started (pid={_wa.get('pid')}, port={_wa.get('port')})")  # cxt-ignore: fake_execution
+        elif whatsapp_sidecar.is_enabled():
+            print(f"[WhatsApp] GoWA sidecar not started: {_wa.get('reason')}")
+    except Exception as e:
+        print(f"[WhatsApp] sidecar init warning: {e}")
+
     # 코드 실행은 호스트 동봉 인터프리터로 직접 동작(Docker 제거, INT-1870) — 샌드박스 warmup 불필요.
 
     # heartbeat은 이 repo(agent.db 분기)에서 테이블 사전생성 함수 없음 — 생략
@@ -198,6 +211,13 @@ async def _warmup_local_llm():
     _hb_task.cancel()
     _cron_task.cancel()
 
+    # Stop the WhatsApp GoWA sidecar if VEGA started it (no-op otherwise)
+    try:
+        from pipeline import whatsapp_sidecar
+        whatsapp_sidecar.stop()
+    except Exception:
+        pass
+
 
 app = FastAPI(title="VEGA", lifespan=lifespan)