diff --git a/tools/gb32960_e2e_smoke.py b/tools/gb32960_e2e_smoke.py
new file mode 100644
index 00000000..91aac0de
--- /dev/null
+++ b/tools/gb32960_e2e_smoke.py
@@ -0,0 +1,262 @@
+#!/usr/bin/env python3
+"""GB32960 TCP ingest to history/archive smoke verifier."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+import pathlib
+import socket
+import time
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any
+
+
+SHANGHAI = dt.timezone(dt.timedelta(hours=8))
+DEFAULT_SAMPLE = (
+    "modules/protocols/protocol-gb32960/src/test/resources/samples/gb32960/realtime_001.hex"
+)
+DEFAULT_FIELD_KEYS = "speed_kmh,total_mileage_km,longitude,latitude"
+
+
+@dataclass(frozen=True)
+class FrameInfo:
+    vin: str
+    command: int
+    event_time: dt.datetime | None
+
+
+def read_hex_frame(path: pathlib.Path) -> bytes:
+    text = path.read_text(encoding="utf-8")
+    return bytes.fromhex("".join(text.split()))
+
+
+def parse_frame_info(frame: bytes) -> FrameInfo:
+    if len(frame) < 25:
+        raise ValueError(f"gb32960 frame too short: {len(frame)}")
+    if frame[:2] not in (b"##", b"$$"):
+        raise ValueError("gb32960 frame must start with ## or $$")
+    data_len = int.from_bytes(frame[22:24], "big")
+    expected_len = 24 + data_len + 1
+    if len(frame) < expected_len:
+        raise ValueError(f"gb32960 frame incomplete: len={len(frame)} expected={expected_len}")
+    command = frame[2]
+    vin = frame[4:21].decode("ascii", errors="ignore").strip()
+    event_time = None
+    if command in (0x01, 0x02, 0x03, 0x04, 0x05) and data_len >= 6:
+        event_time = parse_collect_time(frame[24:30])
+    return FrameInfo(vin=vin, command=command, event_time=event_time)
+
+
+def parse_collect_time(raw: bytes) -> dt.datetime:
+    if len(raw) != 6:
+        raise ValueError("gb32960 collect time must be 6 bytes")
+    year, month, day, hour, minute, second = raw
+    return dt.datetime(2000 + year, month, day, hour, minute, second, tzinfo=SHANGHAI)
+
+
+def query_range(event_time: dt.datetime | None, minutes: int) -> tuple[str, str]:
+    center = event_time or dt.datetime.now(tz=SHANGHAI)
+    center = center.astimezone(SHANGHAI)
+    start = center - dt.timedelta(minutes=minutes)
+    end = center + dt.timedelta(minutes=minutes)
+    return start.isoformat(timespec="seconds"), end.isoformat(timespec="seconds")
+
+
+def send_frame(host: str, port: int, frame: bytes, timeout: float) -> bytes:
+    with socket.create_connection((host, port), timeout=timeout) as sock:
+        sock.settimeout(timeout)
+        sock.sendall(frame)
+        try:
+            return sock.recv(4096)
+        except socket.timeout:
+            return b""
+
+
+def valid_ack(ack: bytes) -> bool:
+    return len(ack) >= 2 and ack[:2] in (b"##", b"$$")
+
+
+def query_json(base_url: str, path: str, params: dict[str, Any], timeout: float) -> Any:
+    url = base_url.rstrip("/") + path + "?" + urllib.parse.urlencode(params)
+    request = urllib.request.Request(url, headers={"Accept": "application/json"})
+    with urllib.request.urlopen(request, timeout=timeout) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def wait_for_records(
+    history_base_url: str,
+    vin: str,
+    date_from: str,
+    date_to: str,
+    minimum: int,
+    timeout_seconds: float,
+    request_timeout: float,
+) -> list[dict[str, Any]]:
+    deadline = time.monotonic() + timeout_seconds
+    params = {
+        "protocol": "GB32960",
+        "vin": vin,
+        "dateFrom": date_from,
+        "dateTo": date_to,
+        "limit": max(minimum, 10),
+        "order": "ASC",
+    }
+    last_items: list[dict[str, Any]] = []
+    while time.monotonic() < deadline:
+        response = query_json(history_base_url, "/api/event-history/records", params, request_timeout)
+        last_items = response if isinstance(response, list) else []
+        if len(last_items) >= minimum:
+            return last_items
+        time.sleep(1)
+    return last_items
+
+
+def wait_for_fields(
+    history_base_url: str,
+    vin: str,
+    field_key: str,
+    date_from: str,
+    date_to: str,
+    timeout_seconds: float,
+    request_timeout: float,
+) -> list[dict[str, Any]]:
+    deadline = time.monotonic() + timeout_seconds
+    params = {
+        "protocol": "GB32960",
+        "vin": vin,
+        "fieldKey": field_key,
+        "dateFrom": date_from,
+        "dateTo": date_to,
+        "limit": 10,
+        "order": "ASC",
+    }
+    last_items: list[dict[str, Any]] = []
+    while time.monotonic() < deadline:
+        response = query_json(history_base_url, "/api/event-history/telemetry/fields", params, request_timeout)
+        last_items = response.get("items") if isinstance(response, dict) else []
+        if last_items:
+            return last_items
+        time.sleep(1)
+    return last_items
+
+
+def parse_field_keys(value: str) -> list[str]:
+    return [item.strip() for item in value.split(",") if item.strip()]
+
+
+def archive_path(root: pathlib.Path, uri: str) -> pathlib.Path:
+    prefix = "archive://"
+    if not uri.startswith(prefix):
+        raise ValueError(f"unsupported archive uri: {uri}")
+    return root / uri[len(prefix):]
+
+
+def check_archive(root: pathlib.Path, records: list[dict[str, Any]], limit: int) -> int:
+    checked = 0
+    seen: set[str] = set()
+    for record in records:
+        uri = record.get("rawArchiveUri") or ""
+        if not uri or uri in seen:
+            continue
+        seen.add(uri)
+        path = archive_path(root, uri)
+        if not path.exists():
+            raise RuntimeError(f"raw archive missing: {path}")
+        checked += 1
+        if checked >= limit:
+            return checked
+    return checked
+
+
+def run(args: argparse.Namespace) -> dict[str, Any]:
+    frame = read_hex_frame(pathlib.Path(args.frame_hex_file))
+    info = parse_frame_info(frame)
+    date_from, date_to = query_range(info.event_time, args.window_minutes)
+
+    ack = send_frame(args.tcp_host, args.tcp_port, frame, args.tcp_timeout)
+    if args.require_ack and not valid_ack(ack):
+        raise RuntimeError(f"gb32960 ack missing or invalid: {ack.hex()}")
+
+    records = wait_for_records(
+        args.history_base_url,
+        info.vin,
+        date_from,
+        date_to,
+        args.expect_record_count,
+        args.history_timeout,
+        args.http_timeout,
+    )
+    if len(records) < args.expect_record_count:
+        raise RuntimeError(f"gb32960 records not visible for {info.vin}: got {len(records)}")
+
+    field_counts: dict[str, int] = {}
+    for field_key in parse_field_keys(args.field_keys):
+        field_items = wait_for_fields(
+            args.history_base_url,
+            info.vin,
+            field_key,
+            date_from,
+            date_to,
+            args.history_timeout,
+            args.http_timeout,
+        )
+        field_counts[field_key] = len(field_items)
+        if not field_items:
+            raise RuntimeError(f"gb32960 telemetry field not visible for {info.vin}: {field_key}")
+
+    archive_checked = 0
+    if args.archive_root:
+        archive_checked = check_archive(pathlib.Path(args.archive_root), records, args.archive_check_limit)
+
+    return {
+        "tcpHost": args.tcp_host,
+        "tcpPort": args.tcp_port,
+        "vin": info.vin,
+        "command": f"0x{info.command:02x}",
+        "eventTime": info.event_time.isoformat() if info.event_time else None,
+        "ackBytesHex": ack.hex(),
+        "records": len(records),
+        "fieldCounts": field_counts,
+        "archiveChecked": archive_checked,
+        "firstRawUri": first_raw_uri(records),
+    }
+
+
+def first_raw_uri(records: list[dict[str, Any]]) -> str | None:
+    for record in records:
+        uri = record.get("rawArchiveUri")
+        if uri:
+            return uri
+    return None
+
+
+def parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--tcp-host", default="127.0.0.1")
+    p.add_argument("--tcp-port", type=int, default=32960)
+    p.add_argument("--history-base-url", default="http://127.0.0.1:20200")
+    p.add_argument("--frame-hex-file", default=DEFAULT_SAMPLE)
+    p.add_argument("--field-keys", default=DEFAULT_FIELD_KEYS)
+    p.add_argument("--window-minutes", type=int, default=10)
+    p.add_argument("--expect-record-count", type=int, default=1)
+    p.add_argument("--archive-root", default="")
+    p.add_argument("--archive-check-limit", type=int, default=1)
+    p.add_argument("--require-ack", action="store_true", default=True)
+    p.add_argument("--tcp-timeout", type=float, default=5)
+    p.add_argument("--http-timeout", type=float, default=5)
+    p.add_argument("--history-timeout", type=float, default=60)
+    return p
+
+
+def main() -> None:
+    args = parser().parse_args()
+    summary = run(args)
+    print(json.dumps(summary, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tools/test_gb32960_e2e_smoke.py b/tools/test_gb32960_e2e_smoke.py
new file mode 100644
index 00000000..a1e76118
--- /dev/null
+++ b/tools/test_gb32960_e2e_smoke.py
@@ -0,0 +1,81 @@
+import datetime as dt
+import pathlib
+import tempfile
+import unittest
+
+from tools import gb32960_e2e_smoke as smoke
+
+
+class Gb32960E2ESmokeTest(unittest.TestCase):
+    def test_parse_frame_info_reads_vin_and_collect_time(self):
+        frame = bytes.fromhex(
+            "232302fe4c544553543230323630363239303030310100081a061d10060f"
+            "02002a"
+        )
+
+        info = smoke.parse_frame_info(frame)
+
+        self.assertEqual(info.vin, "LTEST202606290001")
+        self.assertEqual(info.command, 0x02)
+        self.assertEqual(info.event_time, dt.datetime(2026, 6, 29, 16, 6, 15, tzinfo=smoke.SHANGHAI))
+
+    def test_query_range_surrounds_event_time_with_offset(self):
+        date_from, date_to = smoke.query_range(
+            dt.datetime(2026, 6, 29, 8, 6, 15, tzinfo=smoke.SHANGHAI),
+            minutes=5,
+        )
+
+        self.assertEqual(date_from, "2026-06-29T08:01:15+08:00")
+        self.assertEqual(date_to, "2026-06-29T08:11:15+08:00")
+
+    def test_valid_ack_requires_frame_marker(self):
+        self.assertTrue(smoke.valid_ack(b"##\x02\x01"))
+        self.assertTrue(smoke.valid_ack(b"$$\x02\x01"))
+        self.assertFalse(smoke.valid_ack(b""))
+        self.assertFalse(smoke.valid_ack(b"\x7e\x80"))
+
+    def test_archive_path_maps_archive_uri_to_root(self):
+        path = smoke.archive_path(
+            pathlib.Path("/tmp/archive"),
+            "archive://2026/06/29/GB32960/LTEST202606290001/frame.bin",
+        )
+
+        self.assertEqual(
+            path,
+            pathlib.Path("/tmp/archive/2026/06/29/GB32960/LTEST202606290001/frame.bin"),
+        )
+
+    def test_parse_field_keys_trims_blank_items(self):
+        keys = smoke.parse_field_keys("speed_kmh, total_mileage_km, ,longitude")
+
+        self.assertEqual(keys, ["speed_kmh", "total_mileage_km", "longitude"])
+
+    def test_first_raw_uri_reads_record_archive_uri(self):
+        uri = smoke.first_raw_uri([
+            {"rawArchiveUri": ""},
+            {"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST/frame.bin"},
+        ])
+
+        self.assertEqual(uri, "archive://2026/06/29/GB32960/LTEST/frame.bin")
+
+    def test_check_archive_ignores_duplicate_uris(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = pathlib.Path(temp_dir)
+            archive_file = root / "2026/06/29/GB32960/LTEST/frame.bin"
+            archive_file.parent.mkdir(parents=True)
+            archive_file.write_bytes(b"raw")
+
+            checked = smoke.check_archive(
+                root,
+                [
+                    {"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST/frame.bin"},
+                    {"rawArchiveUri": "archive://2026/06/29/GB32960/LTEST/frame.bin"},
+                ],
+                limit=3,
+            )
+
+        self.assertEqual(checked, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()