diff --git a/main.py b/main.py
new file mode 100644
index 0000000..34e3dc7
--- /dev/null
+++ b/main.py
@@ -0,0 +1,113 @@
+"""Unified command-line entry point for the Access -> SQL Server sync toolkit.
+
+Run from the project root (no ``-m`` needed)::
+
+    python main.py fullsync    [--db FILE] [--table NAME] [--clear-change-log]
+    python main.py incremental [--loop] [--poll-interval N]
+    python main.py compare     [--granularity count|ids] [--db FILE] [--table NAME] [--report PATH]
+
+Configuration is hard-coded to ``config.yaml`` next to this script -- it is not
+a command-line argument, so all three blocks always use the same config (and
+therefore the same target tables).
+
+This file lives at the repo root, outside the ``src/`` package, so it puts
+``src`` on ``sys.path`` itself to import ``sync.*`` regardless of how Python
+was launched or whether the venv already has ``src`` on its path.
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+
+# Make the src/ package importable when running this root script directly.
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))
+
+from sync.config import load_config
+from sync.logging_setup import setup_logging
+from sync.fullsync import full_sync
+from sync import service
+from sync.compare import compare, any_mismatch, format_report
+
+CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
+
+
+def _parse_args(argv):
+    p = argparse.ArgumentParser(
+        prog="python main.py",
+        description="Access -> SQL Server sync toolkit (fullsync / incremental / compare).",
+    )
+    sub = p.add_subparsers(dest="command", required=True)
+
+    pf = sub.add_parser("fullsync", help="one-shot TRUNCATE + bulk INSERT")
+    pf.add_argument("--db", help="limit to one Access file (by FileMapping.file)")
+    pf.add_argument("--table", help="limit to one table (applies to all matched files)")
+    pf.add_argument("--clear-change-log", action="store_true",
+                    help="after loading, clear TableChangeLog on the synced files")
+
+    pi = sub.add_parser("incremental", help="capture -> apply -> cleanup")
+    pi.add_argument("--loop", action="store_true",
+                    help="run continuously (service mode); default is a single pass")
+    pi.add_argument("--poll-interval", type=int, dest="poll_interval",
+                    help="override runtime.poll_interval_seconds (with --loop)")
+
+    pc = sub.add_parser("compare", help="compare Access vs SQL Server data")
+    pc.add_argument("--granularity", choices=["count", "ids"], default="count",
+                    help="count = row totals (default); ids = ID-set membership diff")
+    pc.add_argument("--db", help="limit to one Access file")
+    pc.add_argument("--table", help="limit to one table")
+    pc.add_argument("--report", help="write the report to this file as well as stdout")
+
+    return p.parse_args(argv)
+
+
+def _force_utf8_console():
+    """Render Chinese table names correctly on a Windows GBK console.
+
+    compare prints to stdout by default; without this the default console
+    codepage mojibakes non-ASCII. No-op when stdout is already UTF-8 or when it
+    does not support reconfigure (e.g. some test-capture streams).
+    """
+    for stream in (sys.stdout, sys.stderr):
+        try:
+            stream.reconfigure(encoding="utf-8", errors="replace")
+        except (AttributeError, ValueError):
+            pass
+
+
+def main(argv=None) -> int:
+    """Parse argv, load config, dispatch to the chosen block. Returns exit code."""
+    _force_utf8_console()
+    args = _parse_args(argv)
+    cfg = load_config(CONFIG_PATH)
+    setup_logging(cfg.logging)
+
+    if args.command == "fullsync":
+        full_sync(cfg, db_filter=args.db, table_filter=args.table,
+                  clear_change_log=args.clear_change_log)
+        return 0
+
+    if args.command == "incremental":
+        if args.poll_interval is not None:
+            cfg.runtime.poll_interval_seconds = args.poll_interval
+        if args.loop:
+            service.run(cfg)
+        else:
+            service.cycle(cfg)
+        return 0
+
+    if args.command == "compare":
+        results = compare(cfg, granularity=args.granularity,
+                          db_filter=args.db, table_filter=args.table)
+        report = format_report(results, args.granularity)
+        print(report)
+        if args.report:
+            with open(args.report, "w", encoding="utf-8") as f:
+                f.write(report + "\n")
+        return 1 if any_mismatch(results) else 0
+
+    return 2  # unreachable: argparse requires a subcommand
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/pyproject.toml b/pyproject.toml
index 4ce9694..2323970 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,4 +1,4 @@
 [tool.pytest.ini_options]
-pythonpath = ["src"]
+pythonpath = ["src", "."]
 testpaths = ["tests"]
 markers = ["integration: marks tests requiring real Access/SQL Server"]
diff --git a/tests/test_main.py b/tests/test_main.py
new file mode 100644
index 0000000..dd18d05
--- /dev/null
+++ b/tests/test_main.py
@@ -0,0 +1,70 @@
+"""Routing/argparse tests for the root main.py dispatcher.
+
+The backend functions (full_sync, service.cycle/run, compare) are mocked so
+these tests verify command dispatch, arg parsing, report output and exit codes
+without touching Access or SQL Server.
+"""
+from unittest.mock import patch
+
+import main
+from sync.compare import TableResult
+
+
+def test_fullsync_routes_with_filters():
+    with patch("main.load_config"), patch("main.setup_logging"), \
+         patch("main.full_sync") as fs, patch("main.service") as svc:
+        rc = main.main(["fullsync", "--db", "OEM.accdb", "--table", "表壳焊接记录",
+                        "--clear-change-log"])
+        fs.assert_called_once()
+        assert fs.call_args.kwargs["db_filter"] == "OEM.accdb"
+        assert fs.call_args.kwargs["table_filter"] == "表壳焊接记录"
+        assert fs.call_args.kwargs["clear_change_log"] is True
+        svc.cycle.assert_not_called()
+        assert rc == 0
+
+
+def test_incremental_default_runs_single_cycle():
+    with patch("main.load_config"), patch("main.setup_logging"), \
+         patch("main.full_sync"), patch("main.service") as svc:
+        rc = main.main(["incremental"])
+        svc.cycle.assert_called_once()
+        svc.run.assert_not_called()
+        assert rc == 0
+
+
+def test_incremental_loop_runs_service():
+    with patch("main.load_config"), patch("main.setup_logging"), \
+         patch("main.full_sync"), patch("main.service") as svc:
+        rc = main.main(["incremental", "--loop", "--poll-interval", "5"])
+        svc.run.assert_called_once()
+        svc.cycle.assert_not_called()
+        assert rc == 0
+
+
+def test_compare_default_granularity_is_count(capsys):
+    with patch("main.load_config"), patch("main.setup_logging"), \
+         patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
+        cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 5, 5)]
+        rc = main.main(["compare"])
+        assert cmp.call_args.kwargs["granularity"] == "count"
+        assert rc == 0
+        assert "x.accdb: T -> s.T_YEAR2026" in capsys.readouterr().out
+
+
+def test_compare_ids_granularity_and_mismatch_exit_code():
+    with patch("main.load_config"), patch("main.setup_logging"), \
+         patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
+        cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "mismatch", 5, 7)]
+        rc = main.main(["compare", "--granularity", "ids"])
+        assert cmp.call_args.kwargs["granularity"] == "ids"
+        assert rc == 1
+
+
+def test_compare_writes_report_file(tmp_path):
+    with patch("main.load_config"), patch("main.setup_logging"), \
+         patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
+        cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 1, 1)]
+        rep = tmp_path / "report.txt"
+        rc = main.main(["compare", "--report", str(rep)])
+        assert rc == 0
+        assert "x.accdb: T -> s.T_YEAR2026" in rep.read_text(encoding="utf-8")