diff --git a/connectors/README.md b/connectors/README.md
index cfab88a..0b30bdc 100644
--- a/connectors/README.md
+++ b/connectors/README.md
@@ -12,7 +12,7 @@ core sạch và không phụ thuộc vào bất kỳ nền tảng cụ thể nà
 ```
 connectors/
 ├── zalo/            # Zalo OA Connector — nhận/gửi tin nhắn qua Zalo Official Account
-├── google_sheet/    # (sắp có) đọc/ghi đơn hàng, khách hàng qua Google Sheet
+├── google_sheet/    # Google Sheet Connector — đọc/ghi đơn hàng, khách hàng
 ├── nhanh/           # (sắp có) tích hợp Nhanh.vn
 └── base/            # (sắp có) tích hợp Base.vn
 ```
@@ -37,5 +37,6 @@ connectors/
 ```bash
 pip install -e sdk/python[connectors,dev]
 pip install -r connectors/zalo/requirements.txt
-pytest connectors/zalo/tests
+pip install -r connectors/google_sheet/requirements.txt
+pytest connectors/zalo/tests connectors/google_sheet/tests
 ```
diff --git a/connectors/google_sheet/README.md b/connectors/google_sheet/README.md
new file mode 100644
index 0000000..d8b8b40
--- /dev/null
+++ b/connectors/google_sheet/README.md
@@ -0,0 +1,89 @@
+# Google Sheet Connector
+
+Đọc/ghi dữ liệu đơn hàng (`Order`) và khách hàng (`Customer`) từ Google Sheet —
+phù hợp cho shop nhỏ chưa dùng phần mềm quản lý bán hàng riêng.
+
+## 1. Chuẩn bị
+
+1. Tạo project trên [Google Cloud Console](https://console.cloud.google.com), bật **Google Sheets API**.
+2. Tạo **Service Account** → tải file JSON key.
+3. Mở Google Sheet cần dùng → **Share** với email của Service Account
+   (dạng `xxx@project-id.iam.gserviceaccount.com`), quyền **Editor**.
+4. Copy `config.example.yaml` → `config.yaml`, điền `spreadsheet_id` (lấy từ URL sheet)
+   và nội dung JSON key.
+
+> ⚠️ **`config.yaml` và file JSON key chứa secret** (private key của service account).
+> Không commit hai file này vào git (đã có trong `.gitignore` mẫu — kiểm tra lại trước khi
+> push). Ở production, ưu tiên `service_account_file` trỏ tới key nằm ngoài repo, hoặc
+> nạp key từ secret manager (Vault, AWS Secrets Manager...) thay vì nhúng thẳng vào YAML.
+
+## 2. Cài đặt
+
+Chạy từ thư mục gốc repo (`magic/`):
+
+```bash
+pip install -e 'sdk/python[connectors]'
+pip install -r connectors/google_sheet/requirements.txt
+```
+
+## 3. Sử dụng
+
+Sheet cần có dòng header ở hàng đầu tiên khớp với tên field, ví dụ tab `Customers`:
+
+| id | name | phone | email | address |
+|----|------|-------|-------|---------|
+| C001 | Nguyễn Văn A | 0901234567 | a@example.com | 12 Lê Lợi, Q1 |
+
+> **Quan trọng:** `map_from_common_schema` luôn ghi giá trị theo **thứ tự cột cố định**
+> (đúng thứ tự field trong ví dụ trên với Customer, và
+> `id, customer_id, status, total_amount, shipping_fee, created_at, items_json` với Order).
+> Nếu bạn đổi thứ tự cột trong sheet thật, giá trị sẽ bị ghi lệch cột — giữ đúng thứ tự
+> này, hoặc tự viết lại `map_from_common_schema` để ghi theo header thực tế của sheet.
+
+```python
+import asyncio
+import yaml
+from connectors.google_sheet.connector import GoogleSheetConnector
+
+async def main():
+    config = yaml.safe_load(open("connectors/google_sheet/config.yaml"))
+    async with GoogleSheetConnector(config) as conn:
+        # Đọc toàn bộ khách hàng
+        data = await conn.execute("read_range", {"range": "Customers!A1:E100"})
+        header, *rows = data.get("values", [])
+        customers = [conn.map_to_common_schema({"header": header, "row": r}, "customer") for r in rows]
+        print(customers)
+
+        # Thêm khách hàng mới — thứ tự cột phải khớp header (xem lưu ý ở trên)
+        row = conn.map_from_common_schema(
+            {"id": "C002", "name": "Trần Thị B", "phone": "0909999999"}, "customer"
+        )
+        await conn.execute("append_row", {"range": "Customers!A1:E1", "values": [row]})
+
+asyncio.run(main())
+```
+
+## 4. Các operation hỗ trợ (`execute(operation, params)`)
+
+| operation | params | Mô tả |
+|-----------|--------|-------|
+| `read_range` | `range` (A1 notation, vd `"Orders!A1:H100"`) | Đọc dữ liệu thô |
+| `update_range` | `range`, `values` (list of rows) | Ghi đè một vùng |
+| `append_row` | `range`, `values` | Thêm dòng mới vào cuối bảng |
+| `batch_update` | `updates: [{range, values}, ...]` | Ghi nhiều vùng trong 1 lần gọi API |
+
+## 5. Giới hạn hiện tại
+
+- Mapping Common Schema dựa vào tên cột trong header khi **đọc**; khi **ghi**,
+  `map_from_common_schema` xuất theo thứ tự cột cố định (xem mục 3) — chưa hỗ trợ
+  mapping tùy biến theo header thực tế qua config.
+- `Order.items` lưu dưới dạng JSON string trong cột `items_json` (Sheet không có kiểu
+  dữ liệu lồng nhau).
+- Google Sheets API có giới hạn quota (mặc định 60 request ghi/phút/user) —
+  connector sẽ raise `RateLimitError` khi bị giới hạn, cần tự xử lý backoff ở tầng gọi.
+
+## 6. Chạy test
+
+```bash
+pytest connectors/google_sheet/tests
+```
diff --git a/connectors/google_sheet/__init__.py b/connectors/google_sheet/__init__.py
new file mode 100644
index 0000000..73f9ffc
--- /dev/null
+++ b/connectors/google_sheet/__init__.py
@@ -0,0 +1,5 @@
+"""Google Sheet connector for MagiC."""
+
+from .connector import GoogleSheetConnector
+
+__all__ = ["GoogleSheetConnector"]
diff --git a/connectors/google_sheet/config.example.yaml b/connectors/google_sheet/config.example.yaml
new file mode 100644
index 0000000..149223b
--- /dev/null
+++ b/connectors/google_sheet/config.example.yaml
@@ -0,0 +1,22 @@
+# Copy to config.yaml (gitignored) and fill in real values.
+#
+# 1. Tạo Service Account trên Google Cloud Console, bật Google Sheets API.
+# 2. Tải file JSON key của Service Account.
+# 3. Share Google Sheet với email của Service Account (vd:
+#    xxx@my-project.iam.gserviceaccount.com), quyền Editor.
+# 4. Copy nội dung file JSON key vào service_account_info bên dưới, hoặc
+#    dùng service_account_file trỏ tới đường dẫn file JSON.
+
+spreadsheet_id: "YOUR_SPREADSHEET_ID"  # lấy từ URL: /spreadsheets/d/<ID>/edit
+
+# Cách 1: nhúng trực tiếp nội dung JSON key
+service_account_info:
+  type: service_account
+  project_id: "your-project-id"
+  private_key_id: "..."
+  private_key: "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n"
+  client_email: "xxx@your-project.iam.gserviceaccount.com"
+  token_uri: "https://oauth2.googleapis.com/token"
+
+# Cách 2 (thay cho service_account_info): trỏ tới file JSON key
+# service_account_file: "/path/to/service-account-key.json"
diff --git a/connectors/google_sheet/connector.py b/connectors/google_sheet/connector.py
new file mode 100644
index 0000000..1c9c031
--- /dev/null
+++ b/connectors/google_sheet/connector.py
@@ -0,0 +1,269 @@
+"""GoogleSheetConnector — read/write orders & customers via Google Sheets API v4.
+
+Auth uses a Google service account (JWT bearer flow via `google-auth`) — the
+sheet must be shared with the service account's email. `google-auth` handles
+token minting/signing (real RSA crypto); this module only makes the actual
+Sheets API calls, over httpx like the other connectors.
+
+Reference: https://developers.google.com/workspace/sheets/api/reference/rest/v4/spreadsheets.values
+- Read:          GET    /v4/spreadsheets/{id}/values/{range}
+- Update:        PUT    /v4/spreadsheets/{id}/values/{range}?valueInputOption=...
+- Append a row:  POST   /v4/spreadsheets/{id}/values/{range}:append?valueInputOption=...
+- Batch update:  POST   /v4/spreadsheets/{id}/values:batchUpdate
+Error body shape is Google's standard `{"error": {"code", "message", "status"}}`.
+"""
+
+import asyncio
+import json
+from typing import Any
+from urllib.parse import quote
+
+import httpx
+from google.auth.exceptions import GoogleAuthError
+from google.auth.transport import Request as GoogleAuthRequestBase
+from google.auth.transport import Response as GoogleAuthResponseBase
+from google.oauth2 import service_account
+
+from magic_ai_sdk.connectors.base import BaseConnector
+from magic_ai_sdk.connectors.errors import AuthError, PermanentError, RateLimitError, TransientError, with_retry
+
+API_BASE = "https://sheets.googleapis.com/v4/spreadsheets"
+SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
+
+
+class _HttpxAuthResponse(GoogleAuthResponseBase):
+    """Adapts an httpx.Response to google.auth.transport.Response."""
+
+    def __init__(self, response: httpx.Response):
+        self._response = response
+
+    @property
+    def status(self) -> int:
+        return self._response.status_code
+
+    @property
+    def headers(self) -> dict[str, str]:
+        return dict(self._response.headers)
+
+    @property
+    def data(self) -> bytes:
+        return self._response.content
+
+
+class _HttpxAuthRequest(GoogleAuthRequestBase):
+    """Adapts httpx (sync client) to the google.auth.transport.Request interface,
+    so credential refresh doesn't need the separate `requests` library — the rest
+    of this connector already standardizes on httpx."""
+
+    def __call__(self, url, method="GET", body=None, headers=None, timeout=None, **kwargs) -> _HttpxAuthResponse:
+        with httpx.Client(timeout=timeout or 15.0) as client:
+            resp = client.request(method, url, content=body, headers=headers)
+        return _HttpxAuthResponse(resp)
+
+# Google's standard error `status` values that mean "back off and retry".
+_RATE_LIMIT_STATUSES = {"RESOURCE_EXHAUSTED"}
+_TRANSIENT_STATUSES = {"UNAVAILABLE", "INTERNAL", "DEADLINE_EXCEEDED", "ABORTED"}
+
+_CUSTOMER_FIELDS = ["id", "name", "phone", "email", "address"]
+_ORDER_FIELDS = ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at"]
+
+
+def _safe_float(value: Any) -> float:
+    """Malformed sheet cells shouldn't crash the whole mapping — fall back to 0.0."""
+    try:
+        return float(value) if value else 0.0
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def _as_dict(data: Any) -> dict[str, Any]:
+    """Workflows commonly pass a Pydantic Common Schema model here, not a raw dict."""
+    if isinstance(data, dict):
+        return data
+    if hasattr(data, "model_dump"):
+        return data.model_dump()
+    if hasattr(data, "dict"):
+        return data.dict()
+    raise PermanentError("common_data must be a dict or a Pydantic model")
+
+
+class GoogleSheetConnector(BaseConnector):
+    name = "google_sheet"
+
+    def __init__(self, config: dict[str, Any]):
+        super().__init__(config)
+        self.spreadsheet_id: str = config["spreadsheet_id"]
+        service_account_info = config.get("service_account_info")
+        service_account_file = config.get("service_account_file")
+        if service_account_info:
+            self._credentials = service_account.Credentials.from_service_account_info(
+                service_account_info, scopes=SCOPES
+            )
+        elif service_account_file:
+            self._credentials = service_account.Credentials.from_service_account_file(
+                service_account_file, scopes=SCOPES
+            )
+        else:
+            raise AuthError("GoogleSheetConnector requires service_account_info or service_account_file")
+        self._client: httpx.AsyncClient | None = None
+        self._token_lock = asyncio.Lock()
+
+    async def connect(self) -> bool:
+        self._client = httpx.AsyncClient(timeout=15.0)
+        try:
+            await self._ensure_token()
+        except Exception:
+            await self._client.aclose()
+            self._client = None
+            raise
+        return True
+
+    async def disconnect(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    async def _ensure_token(self) -> str:
+        async with self._token_lock:
+            if not self._credentials.valid:
+                # google-auth's refresh() is a blocking network call; run off the event loop.
+                await asyncio.to_thread(self._credentials.refresh, _HttpxAuthRequest())
+            return self._credentials.token
+
+    @with_retry(max_attempts=3, base_delay=1.0)
+    async def _request(self, method: str, url: str, **kwargs: Any) -> dict[str, Any]:
+        if self._client is None:
+            raise PermanentError("connector not connected — call connect() first")
+
+        try:
+            token = await self._ensure_token()
+            headers = {"Authorization": f"Bearer {token}"}
+            resp = await self._client.request(method, url, headers=headers, **kwargs)
+        except httpx.RequestError as e:
+            raise TransientError(f"network error calling Google Sheets API: {e}") from e
+        except GoogleAuthError as e:
+            # Covers RefreshError and friends — usually a permanently bad/revoked
+            # service account key, not something a retry will fix, so don't retry.
+            raise AuthError(f"Google auth error: {e}") from e
+
+        if resp.is_success:
+            return resp.json() if resp.content else {}
+
+        try:
+            error = resp.json().get("error", {})
+        except ValueError:
+            error = {}
+        status = error.get("status", "")
+        message = error.get("message", f"HTTP {resp.status_code}")
+
+        if resp.status_code == 401:
+            # Token expired/revoked mid-flight (credentials.valid was true when we
+            # checked, but Google invalidated it since) — clear it so the retry
+            # forces a fresh refresh instead of reusing the same bad token.
+            self._credentials.token = None
+            raise TransientError(f"Google Sheets API token expired or invalid: {message}")
+        if resp.status_code == 429 or status in _RATE_LIMIT_STATUSES:
+            raise RateLimitError(f"Google Sheets API rate limited: {message}")
+        if resp.status_code >= 500 or status in _TRANSIENT_STATUSES:
+            raise TransientError(f"Google Sheets API transient error: {message}")
+        raise PermanentError(f"Google Sheets API error ({status or resp.status_code}): {message}")
+
+    def _range_url(self, a1_range: str) -> str:
+        return f"{API_BASE}/{self.spreadsheet_id}/values/{quote(a1_range, safe='')}"
+
+    async def execute(self, operation: str, params: dict[str, Any]) -> Any:
+        if operation == "read_range":
+            return await self._request("GET", self._range_url(params["range"]))
+
+        if operation == "update_range":
+            return await self._request(
+                "PUT",
+                self._range_url(params["range"]),
+                params={"valueInputOption": params.get("value_input_option", "USER_ENTERED")},
+                json={"values": params["values"]},
+            )
+
+        if operation == "append_row":
+            return await self._request(
+                "POST",
+                self._range_url(params["range"]) + ":append",
+                params={"valueInputOption": params.get("value_input_option", "USER_ENTERED")},
+                json={"values": params["values"]},
+            )
+
+        if operation == "batch_update":
+            return await self._request(
+                "POST",
+                f"{API_BASE}/{self.spreadsheet_id}/values:batchUpdate",
+                json={
+                    "valueInputOption": params.get("value_input_option", "USER_ENTERED"),
+                    "data": [{"range": r["range"], "values": r["values"]} for r in params["updates"]],
+                },
+            )
+
+        raise PermanentError(f"unsupported operation: {operation}")
+
+    def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any]:
+        if not isinstance(raw_data, dict):
+            raise PermanentError("raw_data must be a dict with 'header' and 'row' keys")
+        header: list[str] = raw_data.get("header") or []
+        row: list[str] = raw_data.get("row") or []
+        by_col = dict(zip(header, row + [""] * (len(header) - len(row))))
+
+        if entity_type == "customer":
+            known = {f: by_col.pop(f, "") or None for f in _CUSTOMER_FIELDS}
+            return {
+                "id": known["id"] or "",
+                "source_connector": self.name,
+                "external_id": known["id"] or "",
+                "name": known["name"],
+                "phone": known["phone"],
+                "email": known["email"],
+                "address": known["address"],
+                "metadata": by_col,
+            }
+
+        if entity_type == "order":
+            items = []
+            items_raw = by_col.pop("items_json", "")
+            if items_raw:
+                try:
+                    parsed = json.loads(items_raw)
+                except ValueError:
+                    parsed = None
+                if isinstance(parsed, list):
+                    items = parsed
+            known = {f: by_col.pop(f, "") for f in _ORDER_FIELDS}
+            return {
+                "id": known["id"] or "",
+                "source_connector": self.name,
+                "external_id": known["id"] or "",
+                "customer_id": known["customer_id"] or None,
+                "status": known["status"] or "pending",
+                "items": items,
+                "total_amount": _safe_float(known["total_amount"]),
+                "shipping_fee": _safe_float(known["shipping_fee"]),
+                "created_at": known["created_at"] or None,
+                "metadata": by_col,
+            }
+
+        raise PermanentError(f"GoogleSheetConnector cannot map entity_type={entity_type!r}")
+
+    def map_from_common_schema(self, common_data: Any, entity_type: str) -> list[str]:
+        common_data = _as_dict(common_data)
+
+        if entity_type == "customer":
+            fields = _CUSTOMER_FIELDS
+        elif entity_type == "order":
+            fields = [*_ORDER_FIELDS, "items_json"]
+        else:
+            raise PermanentError(f"GoogleSheetConnector cannot map entity_type={entity_type!r}")
+
+        row = []
+        for field in fields:
+            if field == "items_json":
+                row.append(json.dumps(common_data.get("items") or []))
+            else:
+                value = common_data.get(field, "")
+                row.append("" if value is None else str(value))
+        return row
diff --git a/connectors/google_sheet/pytest.ini b/connectors/google_sheet/pytest.ini
new file mode 100644
index 0000000..2f4c80e
--- /dev/null
+++ b/connectors/google_sheet/pytest.ini
@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode = auto
diff --git a/connectors/google_sheet/requirements.txt b/connectors/google_sheet/requirements.txt
new file mode 100644
index 0000000..bb3ce27
--- /dev/null
+++ b/connectors/google_sheet/requirements.txt
@@ -0,0 +1,3 @@
+# Install the framework first: pip install -e ../../sdk/python[connectors]
+httpx>=0.27
+google-auth>=2.30
diff --git a/connectors/google_sheet/tests/__init__.py b/connectors/google_sheet/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/connectors/google_sheet/tests/test_connector.py b/connectors/google_sheet/tests/test_connector.py
new file mode 100644
index 0000000..878ef69
--- /dev/null
+++ b/connectors/google_sheet/tests/test_connector.py
@@ -0,0 +1,251 @@
+"""Unit tests for GoogleSheetConnector — HTTP mocked via respx, auth mocked (no real RSA key)."""
+
+from unittest.mock import MagicMock, patch
+
+import httpx
+import pytest
+import respx
+from google.auth.exceptions import RefreshError
+
+from google_sheet.connector import API_BASE, GoogleSheetConnector
+from magic_ai_sdk.connectors.errors import AuthError, PermanentError, RateLimitError, TransientError
+
+CONFIG = {"spreadsheet_id": "sheet-123", "service_account_info": {"type": "service_account"}}
+
+
+def _fake_credentials(token="tok-1", valid=True):
+    creds = MagicMock()
+    creds.valid = valid
+    creds.token = token
+
+    def _refresh(request):
+        creds.valid = True
+        creds.token = "refreshed-tok"
+
+    creds.refresh.side_effect = _refresh
+    return creds
+
+
+@pytest.fixture(autouse=True)
+def mock_auth():
+    with patch("google_sheet.connector.service_account.Credentials.from_service_account_info") as m:
+        m.return_value = _fake_credentials()
+        yield m
+
+
+def make_connector(**overrides):
+    return GoogleSheetConnector({**CONFIG, **overrides})
+
+
+async def test_connect_ensures_token():
+    conn = make_connector()
+    async with conn:
+        assert conn._credentials.token == "tok-1"
+
+
+@respx.mock
+async def test_read_range_success():
+    respx.get(f"{API_BASE}/sheet-123/values/Customers%21A1%3AE100").mock(
+        return_value=httpx.Response(200, json={"values": [["id", "name"], ["C1", "Anh"]]})
+    )
+    async with make_connector() as conn:
+        data = await conn.execute("read_range", {"range": "Customers!A1:E100"})
+    assert data["values"][1] == ["C1", "Anh"]
+
+
+@respx.mock
+async def test_append_row_sends_correct_body():
+    route = respx.post(f"{API_BASE}/sheet-123/values/Orders%21A1%3AH1:append").mock(
+        return_value=httpx.Response(200, json={"updates": {"updatedRows": 1}})
+    )
+    async with make_connector() as conn:
+        await conn.execute("append_row", {"range": "Orders!A1:H1", "values": [["O1", "C1", "pending"]]})
+
+    assert route.calls.last.request.url.params["valueInputOption"] == "USER_ENTERED"
+    assert route.calls.last.request.headers["Authorization"] == "Bearer tok-1"
+
+
+@respx.mock
+async def test_batch_update_sends_all_ranges():
+    route = respx.post(f"{API_BASE}/sheet-123/values:batchUpdate").mock(
+        return_value=httpx.Response(200, json={"totalUpdatedRows": 2})
+    )
+    async with make_connector() as conn:
+        await conn.execute(
+            "batch_update",
+            {"updates": [{"range": "A1:A1", "values": [["x"]]}, {"range": "B1:B1", "values": [["y"]]}]},
+        )
+    import json as jsonlib
+
+    body = jsonlib.loads(route.calls.last.request.content)
+    assert len(body["data"]) == 2
+
+
+@respx.mock
+async def test_rate_limit_error():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(429, json={"error": {"code": 429, "status": "RESOURCE_EXHAUSTED", "message": "quota"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(RateLimitError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+@respx.mock
+async def test_transient_error_on_unavailable():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(503, json={"error": {"code": 503, "status": "UNAVAILABLE", "message": "down"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(TransientError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+@respx.mock
+async def test_permanent_error_on_permission_denied():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(403, json={"error": {"code": 403, "status": "PERMISSION_DENIED", "message": "nope"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(PermanentError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+@respx.mock
+async def test_network_error_wrapped_as_transient():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(side_effect=httpx.ConnectError("refused"))
+    async with make_connector() as conn:
+        with pytest.raises(TransientError):
+            await conn.execute("read_range", {"range": "A1"})
+
+
+async def test_expired_credentials_are_refreshed(mock_auth):
+    mock_auth.return_value = _fake_credentials(valid=False)
+    conn = make_connector()
+    async with conn:
+        pass
+    assert conn._credentials.token == "refreshed-tok"
+
+
+def test_map_to_common_schema_customer():
+    conn = make_connector()
+    raw = {"header": ["id", "name", "phone", "email", "address", "note"], "row": ["C1", "Anh", "090", "a@x.com", "HN", "VIP"]}
+    result = conn.map_to_common_schema(raw, "customer")
+    assert result["id"] == "C1"
+    assert result["name"] == "Anh"
+    assert result["metadata"] == {"note": "VIP"}
+
+
+def test_map_to_common_schema_order_with_items():
+    conn = make_connector()
+    raw = {
+        "header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at", "items_json"],
+        "row": ["O1", "C1", "confirmed", "100000", "15000", "2026-01-01", '[{"name": "Ao", "quantity": 2}]'],
+    }
+    result = conn.map_to_common_schema(raw, "order")
+    assert result["total_amount"] == 100000.0
+    assert result["items"] == [{"name": "Ao", "quantity": 2}]
+
+
+def test_map_from_common_schema_order_roundtrip():
+    conn = make_connector()
+    row = conn.map_from_common_schema(
+        {"id": "O1", "customer_id": "C1", "status": "pending", "total_amount": 50.0, "items": [{"name": "x"}]},
+        "order",
+    )
+    parsed = conn.map_to_common_schema(
+        {"header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at", "items_json"], "row": row},
+        "order",
+    )
+    assert parsed["id"] == "O1"
+    assert parsed["items"] == [{"name": "x"}]
+
+
+def test_map_to_common_schema_rejects_unknown_entity():
+    conn = make_connector()
+    with pytest.raises(PermanentError):
+        conn.map_to_common_schema({"header": [], "row": []}, "case")
+
+
+def test_map_to_common_schema_rejects_non_dict_raw_data():
+    conn = make_connector()
+    with pytest.raises(PermanentError):
+        conn.map_to_common_schema(["not", "a", "dict"], "customer")
+
+
+def test_map_to_common_schema_ignores_non_list_items_json():
+    conn = make_connector()
+    raw = {
+        "header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at", "items_json"],
+        "row": ["O1", "C1", "pending", "0", "0", "", '{"not": "a list"}'],
+    }
+    result = conn.map_to_common_schema(raw, "order")
+    assert result["items"] == []
+
+
+def test_map_to_common_schema_malformed_amount_defaults_to_zero():
+    conn = make_connector()
+    raw = {
+        "header": ["id", "customer_id", "status", "total_amount", "shipping_fee", "created_at"],
+        "row": ["O1", "C1", "pending", "not-a-number", "also-bad", ""],
+    }
+    result = conn.map_to_common_schema(raw, "order")
+    assert result["total_amount"] == 0.0
+    assert result["shipping_fee"] == 0.0
+
+
+def test_map_from_common_schema_accepts_pydantic_like_model():
+    conn = make_connector()
+
+    class FakeModel:
+        def model_dump(self):
+            return {"id": "C1", "name": "Anh"}
+
+    row = conn.map_from_common_schema(FakeModel(), "customer")
+    assert row[0] == "C1"
+
+
+def test_map_from_common_schema_defaults_none_items_to_empty_list():
+    conn = make_connector()
+    row = conn.map_from_common_schema({"id": "O1", "items": None}, "order")
+    assert row[-1] == "[]"
+
+
+@respx.mock
+async def test_401_response_clears_token_and_raises_transient():
+    respx.get(f"{API_BASE}/sheet-123/values/A1").mock(
+        return_value=httpx.Response(401, json={"error": {"code": 401, "status": "UNAUTHENTICATED", "message": "bad token"}})
+    )
+    async with make_connector() as conn:
+        with pytest.raises(TransientError):
+            await conn.execute("read_range", {"range": "A1"})
+        assert conn._credentials.token is None
+
+
+async def test_connect_failure_closes_client_and_clears_it():
+    with patch("google_sheet.connector.service_account.Credentials.from_service_account_info") as m:
+        creds = _fake_credentials(valid=False)
+        creds.refresh.side_effect = RuntimeError("boom")
+        m.return_value = creds
+
+        conn = GoogleSheetConnector({**CONFIG})
+        with pytest.raises(RuntimeError):
+            await conn.connect()
+        assert conn._client is None
+
+
+async def test_request_without_connect_raises_permanent_error():
+    conn = make_connector()
+    with pytest.raises(PermanentError):
+        await conn.execute("read_range", {"range": "A1"})
+
+
+async def test_refresh_error_during_request_wrapped_as_auth_error():
+    """A revoked/invalid service account key isn't fixed by retrying — surface it
+    as AuthError so the caller stops instead of burning 3 retry attempts."""
+    conn = make_connector()
+    async with conn:
+        conn._credentials.valid = False
+        conn._credentials.refresh.side_effect = RefreshError("revoked")
+        with pytest.raises(AuthError):
+            await conn.execute("read_range", {"range": "A1"})
diff --git a/connectors/zalo/connector.py b/connectors/zalo/connector.py
index 9126ad8..b96ae2c 100644
--- a/connectors/zalo/connector.py
+++ b/connectors/zalo/connector.py
@@ -189,7 +189,7 @@ def verify_webhook_signature(self, headers: Mapping[str, str], raw_body: bytes)
             if not isinstance(payload, dict):
                 return False
             timestamp = str(payload.get("timestamp", ""))
-        except (UnicodeDecodeError, ValueError):
+        except ValueError:  # UnicodeDecodeError and JSONDecodeError both subclass ValueError
             return False
 
         mac_input = f"{self.app_id}{body_str}{timestamp}{self.oa_secret_key}"
diff --git a/connectors/zalo/webhook.py b/connectors/zalo/webhook.py
index 110579a..cd4ac7a 100644
--- a/connectors/zalo/webhook.py
+++ b/connectors/zalo/webhook.py
@@ -48,7 +48,7 @@ def do_POST(self):
 
             try:
                 payload = json.loads(raw_body)
-            except (json.JSONDecodeError, ValueError):
+            except ValueError:  # json.JSONDecodeError subclasses ValueError
                 self.send_error(400, "Invalid JSON")
                 return