diff --git a/app/Console/Commands/SyncBusinessMirrors.php b/app/Console/Commands/SyncBusinessMirrors.php
new file mode 100644
index 00000000..5abb3fa5
--- /dev/null
+++ b/app/Console/Commands/SyncBusinessMirrors.php
@@ -0,0 +1,107 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\Log;
+use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Modules\EcommerceFinance\Services\BusinessMirrorSynchronizer;
+use Throwable;
+
+class SyncBusinessMirrors extends Command
+{
+    protected $signature = 'business-mirror:sync-tables
+        {dataset? : 只同步指定数据集}
+        {--full : 执行或继续首次全量/校准同步}
+        {--batch= : 每批最多读取行数}
+        {--max-batches= : 每个数据集本轮最多批次数，0 表示直到完成}
+        {--max-datasets= : 本轮最多处理的数据集数量}';
+
+    protected $description = '将项目实际使用的上游表按稳定游标同步到共享业务镜像库';
+
+    public function handle(
+        BusinessMirrorDatasetRegistry $datasets,
+        BusinessMirrorSynchronizer $synchronizer,
+        GenericBusinessMirrorRepository $mirror,
+    ): int
+    {
+        if (! (bool) config('upstream_mirrors.refresh_enabled', false) && ! $this->option('full')) {
+            return self::SUCCESS;
+        }
+
+        $requested = $this->argument('dataset');
+        $keys = $requested === null
+            ? $this->scheduledKeys($datasets, $mirror)
+            : [(string) $requested];
+        $maxDatasets = max((int) ($this->option('max-datasets')
+            ?? config('upstream_mirrors.table_sync.max_datasets_per_run', 1)), 1);
+        $processed = 0;
+        $failed = false;
+
+        foreach ($keys as $key) {
+            if ($processed >= $maxDatasets) {
+                break;
+            }
+
+            try {
+                $result = $synchronizer->sync(
+                    $key,
+                    (bool) $this->option('full'),
+                    $this->option('batch') === null ? null : (int) $this->option('batch'),
+                    $this->option('max-batches') === null ? null : (int) $this->option('max-batches'),
+                );
+            } catch (Throwable $exception) {
+                $failed = true;
+                Log::error('共享业务镜像表同步失败。', ['dataset' => $key, 'exception' => $exception::class]);
+                $this->error($key . '：同步失败，已保留上一版共享数据。');
+                $processed++;
+                continue;
+            }
+
+            if ($result['status'] === 'requires_full_sync') {
+                if ($requested !== null) {
+                    $this->warn($key . '：尚未完成首次全量同步，请显式使用 --full。');
+                }
+                continue;
+            }
+
+            $processed++;
+            $this->info(sprintf(
+                '%s：%s，本轮 %d 批，累计处理 %d 行。',
+                $key,
+                $result['status'] === 'succeeded' ? '同步完成' : '同步进行中',
+                $result['batches'],
+                $result['rows'],
+            ));
+        }
+
+        return $failed ? self::FAILURE : self::SUCCESS;
+    }
+
+    /** @return list<string> */
+    private function scheduledKeys(
+        BusinessMirrorDatasetRegistry $datasets,
+        GenericBusinessMirrorRepository $mirror,
+    ): array {
+        $scheduled = [];
+
+        foreach ($datasets->all() as $key => $dataset) {
+            $state = $mirror->state($dataset);
+            if (! is_array($state) || ($state['initial_completed_at'] ?? null) === null) {
+                continue;
+            }
+
+            $scheduled[] = [
+                'key' => $key,
+                'last_attempted_at' => (string) ($state['last_attempted_at'] ?? ''),
+            ];
+        }
+
+        usort($scheduled, fn (array $left, array $right): int =>
+            [$left['last_attempted_at'], $left['key']] <=> [$right['last_attempted_at'], $right['key']]
+        );
+
+        return array_column($scheduled, 'key');
+    }
+}
diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php
index 68573f37..2df70817 100644
--- a/config/upstream_mirrors.php
+++ b/config/upstream_mirrors.php
@@ -18,4 +18,68 @@ return [
         // 上游完成记录通常不再变化；仍回看 48 小时，覆盖迟到或修正数据。
         'overlap_hours' => (int) env('PRODUCT_PRICE_MIRROR_OVERLAP_HOURS', 48),
     ],
+
+    /*
+    |--------------------------------------------------------------------------
+    | 上游原表镜像目录
+    |--------------------------------------------------------------------------
+    |
+    | 页面不能再按“页码 + 筛选条件”临时生成一份快照。这里登记项目代码实际
+    | 使用的上游表/视图，由后台按表同步到 caiwu_business_mirror。首次全量
+    | 必须显式执行；日常任务只续传已完成首次全量的数据集。
+    |
+    | watermark_column 为空时按稳定业务键追加，并依靠周期全量校准修正历史
+    | 更新和删除；存在可靠更新时间时使用重叠窗口增量同步。
+    */
+    'table_sync' => [
+        'batch_size' => (int) env('BUSINESS_MIRROR_TABLE_BATCH_SIZE', 500),
+        'max_batches_per_run' => (int) env('BUSINESS_MIRROR_TABLE_MAX_BATCHES', 2),
+        'max_datasets_per_run' => (int) env('BUSINESS_MIRROR_MAX_DATASETS', 1),
+        'overlap_hours' => (int) env('BUSINESS_MIRROR_OVERLAP_HOURS', 48),
+        'full_reconcile_hours' => (int) env('BUSINESS_MIRROR_FULL_RECONCILE_HOURS', 168),
+    ],
+
+    'datasets' => [
+        // 经营分析副本库：组织资料、订单、退款和聚水潭基础数据。
+        'operations_replica.rpt.fys_companyInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'fys_companyInfo', 'target' => 'ops_fys_company_info', 'keys' => ['Id'], 'watermark' => 'updateTIme'],
+        'operations_replica.rpt.fys_shopInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'fys_shopInfo', 'target' => 'ops_fys_shop_info', 'keys' => ['Id'], 'watermark' => 'updateTime'],
+        'operations_replica.rpt.fys_TrademarkInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'fys_TrademarkInfo', 'target' => 'ops_fys_trademark_info', 'keys' => ['Id'], 'watermark' => 'updateTime'],
+        'operations_replica.rpt.jst_Orders' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_Orders', 'target' => 'ops_jst_orders', 'keys' => ['Id'], 'watermark' => 'updateTime'],
+        'operations_replica.rpt.jst_OrderItem' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_OrderItem', 'target' => 'ops_jst_order_item', 'keys' => ['Id']],
+        'operations_replica.rpt.jst_sku' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_sku', 'target' => 'ops_jst_sku', 'keys' => ['autoid']],
+        'operations_replica.rpt.jst_warehouseInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_warehouseInfo', 'target' => 'ops_jst_warehouse_info', 'keys' => ['Id'], 'watermark' => 'updateTime'],
+        'operations_replica.rpt.jst_skumap' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_skumap', 'target' => 'ops_jst_sku_map', 'keys' => ['shop_id', 'shop_i_id', 'shop_sku_id'], 'watermark' => 'modifiedTime'],
+        'operations_replica.rpt.ads_itemid_sales_cost_ratio_v2_di' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'ads_itemid_sales_cost_ratio_v2_di', 'target' => 'ops_item_sales_cost_ratio', 'keys' => ['stat_date', 'item_id'], 'watermark' => 'item_id_update'],
+        'operations_replica.rpt.jst_returnAndRefundInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_returnAndRefundInfo', 'target' => 'ops_jst_return_refund_info', 'keys' => ['Id'], 'watermark' => 'updateTime'],
+        'operations_replica.rpt.jst_returnAndRefundItem' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_returnAndRefundItem', 'target' => 'ops_jst_return_refund_item', 'keys' => ['Id']],
+        'operations_replica.rpt.jst_shopInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_shopInfo', 'target' => 'ops_jst_shop_info', 'keys' => ['Id'], 'watermark' => 'updateTime'],
+
+        // JianSun 旧业务库：营销账单、供应商应付、板材、封边条和库存。
+        'jiansun.dbo.tmall_spread_bills' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'tmall_spread_bills', 'target' => 'js_tmall_spread_bills', 'keys' => ['id']],
+        'jiansun.dbo.pdd_spread_bills' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pdd_spread_bills', 'target' => 'js_pdd_spread_bills', 'keys' => ['id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.alibaba_bill' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'alibaba_bill', 'target' => 'js_alibaba_bill', 'keys' => ['id']],
+        'jiansun.dbo.jd_spreadBills' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'jd_spreadBills', 'target' => 'js_jd_spread_bills', 'keys' => ['id'], 'watermark' => 'updateTime'],
+        'jiansun.dbo.jd_costBills' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'jd_costBills', 'target' => 'js_jd_cost_bills', 'keys' => ['id'], 'watermark' => 'creatTime'],
+        'jiansun.dbo.finance_purchaseReturn' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_purchaseReturn', 'target' => 'js_finance_purchase_return', 'keys' => ['id']],
+        'jiansun.dbo.finance_invoiceDeduction' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_invoiceDeduction', 'target' => 'js_finance_invoice_deduction', 'keys' => ['id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.fys_supplierPaymentSummary' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'fys_supplierPaymentSummary', 'target' => 'js_supplier_payment_summary', 'keys' => ['id'], 'watermark' => 'modifiedTime'],
+        'jiansun.dbo.finance_advanceLoans' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_advanceLoans', 'target' => 'js_finance_advance_loans', 'keys' => ['id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.pm_unOrderDetails' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_unOrderDetails', 'target' => 'js_pm_unorder_details', 'keys' => ['Id']],
+        'jiansun.dbo.finance_sampleDeduction' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_sampleDeduction', 'target' => 'js_finance_sample_deduction', 'keys' => ['id'], 'watermark' => 'finishTime'],
+        'jiansun.dbo.fys_express_courier_settlement' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'fys_express_courier_settlement', 'target' => 'js_express_courier_settlement', 'keys' => ['Id'], 'watermark' => 'modify_time'],
+        'jiansun.dbo.finance_boardDeduction' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_boardDeduction', 'target' => 'js_finance_board_deduction', 'keys' => ['id']],
+        'jiansun.dbo.pm_plainPlateData' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_plainPlateData', 'target' => 'js_pm_plain_plate_data', 'keys' => ['Id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.pm_boardPurchasePrice' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_boardPurchasePrice', 'target' => 'js_pm_board_purchase_price', 'keys' => ['id'], 'watermark' => 'updateTime'],
+        'jiansun.dbo.pm_boardPriceChangeImport' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_boardPriceChangeImport', 'target' => 'js_pm_board_price_change', 'keys' => ['id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.pm_boardApplyDetail' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_boardApplyDetail', 'target' => 'js_pm_board_apply_detail', 'keys' => ['Id']],
+        'jiansun.dbo.pm_edgeBandData' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_edgeBandData', 'target' => 'js_pm_edge_band_data', 'keys' => ['id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.pm_edgeBandApplyDetail' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_edgeBandApplyDetail', 'target' => 'js_pm_edge_band_apply_detail', 'keys' => ['Id']],
+        'jiansun.dbo.pm_edgeBandProfitInfo' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_edgeBandProfitInfo', 'target' => 'js_pm_edge_band_profit_info', 'keys' => ['id']],
+        'jiansun.dbo.finance_edgeBandDeductio' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_edgeBandDeductio', 'target' => 'js_finance_edge_band_deduction', 'keys' => ['id']],
+        'jiansun.dbo.pm_averageUnitPrice' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_averageUnitPrice', 'target' => 'js_pm_average_unit_price', 'keys' => ['id']],
+        'jiansun.dbo.pm_boardInventoryDetail_snapshot' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_boardInventoryDetail_snapshot', 'target' => 'js_pm_board_inventory_snapshot', 'keys' => ['id']],
+        'jiansun.dbo.fys_supplierGoods_sttlement_purchase' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'fys_supplierGoods_sttlement_purchase', 'target' => 'js_supplier_goods_purchase', 'keys' => ['Id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.fys_supplierGoods_sttlement_sale' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'fys_supplierGoods_sttlement_sale', 'target' => 'js_supplier_goods_sale', 'keys' => ['Id'], 'watermark' => 'createTime'],
+        'jiansun.dbo.pm_purchaseIn' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'pm_purchaseIn', 'target' => 'js_pm_purchase_in', 'keys' => ['id'], 'watermark' => 'updateTime'],
+    ],
 ];
diff --git a/database/migrations/2026_08_14_230000_create_generic_business_mirror_tables.php b/database/migrations/2026_08_14_230000_create_generic_business_mirror_tables.php
new file mode 100644
index 00000000..53f18ce9
--- /dev/null
+++ b/database/migrations/2026_08_14_230000_create_generic_business_mirror_tables.php
@@ -0,0 +1,62 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration
+{
+    private function connection(): string
+    {
+        return (string) config('upstream_mirrors.connection', 'business_mirror');
+    }
+
+    public function up(): void
+    {
+        if (app()->environment('testing')
+            || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') {
+            return;
+        }
+
+        $schema = Schema::connection($this->connection());
+
+        if ($schema->hasTable('business_mirror_table_states')) {
+            return;
+        }
+
+        $schema->create('business_mirror_table_states', function (Blueprint $table): void {
+            $table->string('dataset_key', 191)->primary();
+            $table->string('source_key', 64);
+            $table->string('source_object', 191);
+            $table->string('target_table', 64);
+            $table->string('status', 24)->default('pending');
+            $table->string('run_mode', 24)->nullable();
+            $table->uuid('generation')->nullable();
+            $table->json('cursor')->nullable();
+            $table->json('high_watermark')->nullable();
+            $table->dateTime('window_started_at', 6)->nullable();
+            $table->dateTime('window_ends_at', 6)->nullable();
+            $table->dateTime('initial_completed_at')->nullable();
+            $table->dateTime('last_full_succeeded_at')->nullable();
+            $table->dateTime('last_attempted_at')->nullable();
+            $table->dateTime('last_succeeded_at')->nullable();
+            $table->unsignedBigInteger('last_upserted_rows')->default(0);
+            $table->unsignedBigInteger('mirror_row_count')->default(0);
+            $table->string('last_error_code', 120)->nullable();
+            $table->timestamps();
+
+            $table->index(['source_key', 'status'], 'mirror_table_states_source_status');
+            $table->index('last_succeeded_at', 'mirror_table_states_last_success');
+        });
+    }
+
+    public function down(): void
+    {
+        if (app()->environment('testing')
+            || (string) config('system_database.mode', 'bootstrap') === 'bootstrap') {
+            return;
+        }
+
+        Schema::connection($this->connection())->dropIfExists('business_mirror_table_states');
+    }
+};
diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh
index 9bc68558..2e7f6097 100644
--- a/deploy/activate-incremental-release.sh
+++ b/deploy/activate-incremental-release.sh
@@ -128,7 +128,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha"
 # 只运行已登记的上游后台读取，不启用整个 Laravel 调度器，避免意外启动
 # 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。
 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF'
-* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet' >/dev/null 2>&1
+* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1
 EOF
 chmod 644 /etc/cron.d/caiwu-jiansun-snapshots
 
diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh
index 31559d6f..45a4b552 100644
--- a/deploy/activate-release.sh
+++ b/deploy/activate-release.sh
@@ -132,7 +132,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha"
 # 只运行已登记的上游后台读取，不启用整个 Laravel 调度器，避免意外启动
 # 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。
 cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF'
-* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet' >/dev/null 2>&1
+* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1
 EOF
 chmod 644 /etc/cron.d/caiwu-jiansun-snapshots
 
diff --git a/docs/上游业务镜像清单.md b/docs/上游业务镜像清单.md
new file mode 100644
index 00000000..0cc43440
--- /dev/null
+++ b/docs/上游业务镜像清单.md
@@ -0,0 +1,60 @@
+# 上游业务镜像清单
+
+## 结论
+
+项目不是只使用 `rpt.Bulk_Price_Limit` 一张表。当前主线代码直接读取两套上游中的约 40 个表或视图，覆盖组织资料、订单、退款、营销账户、供应商应付、板材、封边条和库存。
+
+`caiwu_business_mirror` 的目标是把项目实际使用的上游原表数据保存为共享业务镜像。网页只读镜像，后台刷新节点在低压、可断点的条件下完成首次全量和后续增量；不得再按页码或筛选条件临时生成不完整的数据副本。
+
+## 当前迁移状态
+
+| 状态 | 说明 |
+| --- | --- |
+| 已完成 | `operations_replica/rpt.Bulk_Price_Limit` 已进入类型化表 `product_price_mirror_rows`，商品价格页面已读取镜像并支持完整分页。 |
+| 本次建立 | `config/upstream_mirrors.php` 中登记全部当前代码引用的原表；通用表级同步框架负责建表、稳定游标、断点续传、增量重叠和全量校准。 |
+| 待逐域切换 | 其他业务 Repository 仍需从上游 SQL Server 查询改为读取对应镜像表。切换前保留旧快照作为回滚，不得宣称已经完成。 |
+
+## 经营分析副本库（operations_replica）
+
+| 业务范围 | 上游对象 | 镜像目标 |
+| --- | --- | --- |
+| 公司资料 | `rpt.fys_companyInfo` | `ops_fys_company_info` |
+| 店铺资料 | `rpt.fys_shopInfo` | `ops_fys_shop_info` |
+| 商标资料 | `rpt.fys_TrademarkInfo` | `ops_fys_trademark_info` |
+| 订单 | `rpt.jst_Orders`、`rpt.jst_OrderItem` | `ops_jst_orders`、`ops_jst_order_item` |
+| 商品与仓库 | `rpt.jst_sku`、`rpt.jst_warehouseInfo`、`rpt.jst_skumap` | `ops_jst_*` |
+| 推广费用 | `rpt.ads_itemid_sales_cost_ratio_v2_di` | `ops_item_sales_cost_ratio` |
+| 退款退货 | `rpt.jst_returnAndRefundInfo`、`rpt.jst_returnAndRefundItem` | `ops_jst_return_refund_*` |
+| 聚水潭店铺 | `rpt.jst_shopInfo` | `ops_jst_shop_info` |
+
+商品价格 `rpt.Bulk_Price_Limit` 继续使用已经上线的类型化同步器，不在通用原表任务中重复同步。
+
+## JianSun 旧业务库（jiansun）
+
+| 业务范围 | 上游对象 |
+| --- | --- |
+| 营销账户 | `dbo.tmall_spread_bills`、`dbo.pdd_spread_bills`、`dbo.alibaba_bill`、`dbo.jd_spreadBills`、`dbo.jd_costBills` |
+| 供应商应付 | `dbo.finance_purchaseReturn`、`dbo.finance_invoiceDeduction`、`dbo.fys_supplierPaymentSummary`、`dbo.finance_advanceLoans`、`dbo.pm_unOrderDetails`、`dbo.finance_sampleDeduction`、`dbo.fys_express_courier_settlement` |
+| 板材 | `dbo.finance_boardDeduction`、`dbo.pm_plainPlateData`、`dbo.pm_boardPurchasePrice`、`dbo.pm_boardPriceChangeImport`、`dbo.pm_boardApplyDetail` |
+| 封边条 | `dbo.pm_edgeBandData`、`dbo.pm_edgeBandApplyDetail`、`dbo.pm_edgeBandProfitInfo`、`dbo.finance_edgeBandDeductio`、`dbo.pm_averageUnitPrice` |
+| 库存与货款对账 | `dbo.pm_boardInventoryDetail_snapshot`、`dbo.fys_supplierGoods_sttlement_purchase`、`dbo.fys_supplierGoods_sttlement_sale`、`dbo.pm_purchaseIn` |
+
+其中 `fys_supplierGoods_sttlement_sale` 当前约 280 万行，是首次全量同步的最大对象。必须在低峰期小批量、可断点执行，禁止把全量同步塞进发布流程。
+
+## 同步规则
+
+1. 首次全量只能通过 `business-mirror:sync-tables <dataset> --full` 显式启动；失败或中断后使用同一命令继续。
+2. 日常任务只处理已经完成首次全量的数据集。存在可靠更新时间的表使用 48 小时重叠窗口；没有可靠更新时间的表按稳定主键追加。
+3. 全量校准使用新的同步代次，只有全部完成后才删除镜像中本轮未出现的旧行，因此中途失败不会清空当前可用数据。
+4. 每轮默认只处理 1 个数据集、每个数据集 2 批，每批 500 行，避免同时压迫两个旧系统。
+5. 页面切换必须以“总记录数、第一页、第二页、末页、筛选、汇总、空数据和更新时间”作为验收，不能只看首页有数据。
+
+## 后续切换顺序
+
+1. 公司、店铺、商标等小型维表。
+2. 营销账户五张单表账单。
+3. 供应商应付与货款对账。
+4. 订单、退款和推广费用。
+5. 板材、封边条和库存的跨表查询。
+
+每个阶段都必须先完成首次全量和数量核对，再将对应 Repository 切到镜像；任一阶段失败时保留上一版镜像和旧只读链路用于回滚。
diff --git a/docs/开发与部署SOP.md b/docs/开发与部署SOP.md
index bb011ae4..146c0a6d 100644
--- a/docs/开发与部署SOP.md
+++ b/docs/开发与部署SOP.md
@@ -234,6 +234,15 @@ php artisan business-mirror:sync-product-prices --full
 php artisan business-mirror:sync-product-prices
 ```
 
+其余已在代码中使用的上游表由统一的数据集目录管理。首次全量必须逐个数据集显式执行，普通定时任务只续传已完成首次全量的数据集：
+
+```bash
+php artisan business-mirror:sync-tables <dataset-key> --full
+php artisan business-mirror:sync-tables
+```
+
+完整数据集、来源对象、镜像表和分阶段切换状态见 `docs/上游业务镜像清单.md`。不得把“已登记同步目录”误报成“页面已经切换完成”；每个业务域仍需在首份全量核对通过后，把对应 Repository 改为只读镜像。
+
 首次 `--full` 属于生产数据写入，必须先备份 `caiwu_system_config` 和 `caiwu_business_mirror`，并由负责人明确授权。后续无参数命令只执行或继续受控增量。
 
 ### 8.3 开发者需要尚未接入的上游数据时
diff --git a/modules/EcommerceFinance/Contracts/BusinessMirrorDataset.php b/modules/EcommerceFinance/Contracts/BusinessMirrorDataset.php
new file mode 100644
index 00000000..015d6b41
--- /dev/null
+++ b/modules/EcommerceFinance/Contracts/BusinessMirrorDataset.php
@@ -0,0 +1,21 @@
+<?php
+
+namespace Modules\EcommerceFinance\Contracts;
+
+interface BusinessMirrorDataset
+{
+    public function key(): string;
+
+    public function sourceKey(): string;
+
+    public function sourceSchema(): string;
+
+    public function sourceObject(): string;
+
+    public function targetTable(): string;
+
+    /** @return list<string> */
+    public function keyColumns(): array;
+
+    public function watermarkColumn(): ?string;
+}
diff --git a/modules/EcommerceFinance/Repositories/GenericBusinessMirrorRepository.php b/modules/EcommerceFinance/Repositories/GenericBusinessMirrorRepository.php
new file mode 100644
index 00000000..4dcee6ad
--- /dev/null
+++ b/modules/EcommerceFinance/Repositories/GenericBusinessMirrorRepository.php
@@ -0,0 +1,256 @@
+<?php
+
+namespace Modules\EcommerceFinance\Repositories;
+
+use Illuminate\Database\ConnectionInterface;
+use Illuminate\Database\DatabaseManager;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use RuntimeException;
+
+class GenericBusinessMirrorRepository
+{
+    private readonly ConnectionInterface $connection;
+
+    public function __construct(DatabaseManager $database)
+    {
+        $this->connection = $database->connection(
+            (string) config('upstream_mirrors.connection', 'business_mirror')
+        );
+    }
+
+    public function connection(): ConnectionInterface
+    {
+        return $this->connection;
+    }
+
+    public function ready(BusinessMirrorDataset $dataset): bool
+    {
+        $schema = Schema::connection($this->connection->getName());
+
+        if (! $schema->hasTable('business_mirror_table_states')
+            || ! $schema->hasTable($dataset->targetTable())) {
+            return false;
+        }
+
+        $state = $this->state($dataset);
+
+        return is_array($state) && ($state['initial_completed_at'] ?? null) !== null;
+    }
+
+    /** @return array<string,mixed>|null */
+    public function state(BusinessMirrorDataset $dataset): ?array
+    {
+        if (! Schema::connection($this->connection->getName())->hasTable('business_mirror_table_states')) {
+            return null;
+        }
+
+        $row = $this->connection->table('business_mirror_table_states')
+            ->where('dataset_key', $dataset->key())
+            ->first();
+
+        if ($row === null) {
+            return null;
+        }
+
+        $state = (array) $row;
+        $state['cursor'] = $this->decode($state['cursor'] ?? null);
+        $state['high_watermark'] = $this->decode($state['high_watermark'] ?? null);
+
+        return $state;
+    }
+
+    public function startRun(
+        BusinessMirrorDataset $dataset,
+        string $mode,
+        string $generation,
+        ?string $windowStart,
+        ?string $windowEnd,
+        bool $resume,
+    ): array {
+        $now = now();
+        $values = [
+            'source_key' => $dataset->sourceKey(),
+            'source_object' => $dataset->sourceSchema() . '.' . $dataset->sourceObject(),
+            'target_table' => $dataset->targetTable(),
+            'status' => 'running',
+            'run_mode' => $mode,
+            'generation' => $generation,
+            'last_attempted_at' => $now,
+            'last_error_code' => null,
+            'updated_at' => $now,
+        ];
+
+        if (! $resume) {
+            $values += [
+                'cursor' => null,
+                'window_started_at' => $windowStart,
+                'window_ends_at' => $windowEnd,
+                'last_upserted_rows' => 0,
+            ];
+        }
+
+        $query = $this->connection->table('business_mirror_table_states')
+            ->where('dataset_key', $dataset->key());
+
+        if ($query->exists()) {
+            $query->update($values);
+        } else {
+            $this->connection->table('business_mirror_table_states')->insert(
+                ['dataset_key' => $dataset->key(), 'created_at' => $now] + $values,
+            );
+        }
+
+        return $this->state($dataset) ?? [];
+    }
+
+    public function storeBatch(
+        BusinessMirrorDataset $dataset,
+        array $columns,
+        array $rows,
+        array $cursor,
+        string $generation,
+        int $cumulativeRows,
+    ): void {
+        $now = now();
+        $columnNames = array_map(fn (array $column): string => (string) $column['column_name'], $columns);
+        $mapped = [];
+
+        foreach ($rows as $row) {
+            $values = [];
+            foreach ($columnNames as $columnName) {
+                $values[$columnName] = $this->value($row, $columnName);
+            }
+
+            $identity = array_map(
+                fn (string $column): string => $this->identityValue($this->value($row, $column)),
+                $dataset->keyColumns(),
+            );
+
+            $mapped[] = $values + [
+                '_mirror_key' => hash('sha256', implode("\0", $identity)),
+                '_mirror_generation' => $generation,
+                '_mirror_synced_at' => $now,
+            ];
+        }
+
+        $this->connection->transaction(function () use (
+            $dataset, $columnNames, $mapped, $cursor, $cumulativeRows, $now
+        ): void {
+            $updates = [...$columnNames, '_mirror_generation', '_mirror_synced_at'];
+            $chunkSize = max(1, intdiv(50000, max(count($updates) + 1, 1)));
+
+            foreach (array_chunk($mapped, $chunkSize) as $chunk) {
+                $this->connection->table($dataset->targetTable())->upsert($chunk, ['_mirror_key'], $updates);
+            }
+
+            $this->connection->table('business_mirror_table_states')
+                ->where('dataset_key', $dataset->key())
+                ->update([
+                    'cursor' => json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION),
+                    'last_upserted_rows' => $cumulativeRows,
+                    'updated_at' => $now,
+                ]);
+        });
+    }
+
+    public function complete(
+        BusinessMirrorDataset $dataset,
+        string $mode,
+        string $generation,
+        array $highWatermark,
+        int $rows,
+    ): void {
+        $now = now();
+
+        $this->connection->transaction(function () use (
+            $dataset, $mode, $generation, $highWatermark, $rows, $now
+        ): void {
+            if ($mode === 'full') {
+                $existingRows = $this->connection->table($dataset->targetTable())->count();
+
+                if ($rows === 0 && $existingRows > 0) {
+                    throw new RuntimeException('business_mirror_unexpected_empty_source');
+                }
+
+                $this->connection->table($dataset->targetTable())
+                    ->where(function ($query) use ($generation): void {
+                        $query->whereNull('_mirror_generation')
+                            ->orWhere('_mirror_generation', '!=', $generation);
+                    })
+                    ->delete();
+            }
+
+            $rowCount = $this->connection->table($dataset->targetTable())->count();
+            $values = [
+                'status' => 'succeeded',
+                'cursor' => null,
+                'high_watermark' => json_encode($highWatermark, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION),
+                'last_succeeded_at' => $now,
+                'last_upserted_rows' => $rows,
+                'mirror_row_count' => $rowCount,
+                'last_error_code' => null,
+                'updated_at' => $now,
+            ];
+
+            if ($mode === 'full') {
+                $values['initial_completed_at'] = $now;
+                $values['last_full_succeeded_at'] = $now;
+            }
+
+            $this->connection->table('business_mirror_table_states')
+                ->where('dataset_key', $dataset->key())
+                ->update($values);
+        });
+    }
+
+    public function fail(BusinessMirrorDataset $dataset, string $errorCode): void
+    {
+        $this->connection->table('business_mirror_table_states')
+            ->where('dataset_key', $dataset->key())
+            ->update([
+                'status' => 'failed',
+                'last_error_code' => mb_substr($errorCode, 0, 120),
+                'updated_at' => now(),
+            ]);
+    }
+
+    private function decode(mixed $value): ?array
+    {
+        if (is_array($value)) {
+            return $value;
+        }
+
+        if (! is_string($value) || $value === '') {
+            return null;
+        }
+
+        $decoded = json_decode($value, true);
+
+        return is_array($decoded) ? $decoded : null;
+    }
+
+    private function value(array $row, string $name): mixed
+    {
+        foreach ($row as $key => $value) {
+            if (strcasecmp((string) $key, $name) === 0) {
+                return $value;
+            }
+        }
+
+        return null;
+    }
+
+    private function identityValue(mixed $value): string
+    {
+        if ($value === null) {
+            return '<NULL>';
+        }
+
+        if (is_bool($value)) {
+            return $value ? '1' : '0';
+        }
+
+        return (string) $value;
+    }
+}
diff --git a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php
new file mode 100644
index 00000000..5df609ee
--- /dev/null
+++ b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php
@@ -0,0 +1,78 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use InvalidArgumentException;
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+
+class BusinessMirrorDatasetRegistry
+{
+    /** @return array<string,BusinessMirrorDataset> */
+    public function all(): array
+    {
+        $datasets = [];
+
+        foreach ((array) config('upstream_mirrors.datasets', []) as $key => $definition) {
+            $datasets[(string) $key] = $this->make((string) $key, (array) $definition);
+        }
+
+        return $datasets;
+    }
+
+    public function get(string $key): BusinessMirrorDataset
+    {
+        $definitions = (array) config('upstream_mirrors.datasets', []);
+        $definition = (array) ($definitions[$key] ?? []);
+
+        if ($definition === []) {
+            throw new InvalidArgumentException('未知的业务镜像数据集：' . $key);
+        }
+
+        return $this->make($key, $definition);
+    }
+
+    private function make(string $key, array $definition): BusinessMirrorDataset
+    {
+        $source = trim((string) ($definition['source_key'] ?? ''));
+        $schema = trim((string) ($definition['schema'] ?? ''));
+        $object = trim((string) ($definition['object'] ?? ''));
+        $target = trim((string) ($definition['target'] ?? ''));
+        $keys = array_values(array_filter(array_map('strval', (array) ($definition['keys'] ?? []))));
+        $watermark = trim((string) ($definition['watermark'] ?? ''));
+
+        foreach ([$source, $schema, $object, $target, ...$keys] as $identifier) {
+            if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) {
+                throw new InvalidArgumentException('业务镜像数据集包含不安全的标识符：' . $key);
+            }
+        }
+
+        if ($keys === [] || strlen($target) > 64) {
+            throw new InvalidArgumentException('业务镜像数据集缺少稳定键或目标表名过长：' . $key);
+        }
+
+        if ($watermark !== '' && preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $watermark) !== 1) {
+            throw new InvalidArgumentException('业务镜像数据集的增量字段不安全：' . $key);
+        }
+
+        return new readonly class($key, $source, $schema, $object, $target, $keys, $watermark ?: null) implements BusinessMirrorDataset {
+            public function __construct(
+                private string $key,
+                private string $source,
+                private string $schema,
+                private string $object,
+                private string $target,
+                private array $keys,
+                private ?string $watermark,
+            ) {
+            }
+
+            public function key(): string { return $this->key; }
+            public function sourceKey(): string { return $this->source; }
+            public function sourceSchema(): string { return $this->schema; }
+            public function sourceObject(): string { return $this->object; }
+            public function targetTable(): string { return $this->target; }
+            public function keyColumns(): array { return $this->keys; }
+            public function watermarkColumn(): ?string { return $this->watermark; }
+        };
+    }
+}
diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php
new file mode 100644
index 00000000..5608d1fd
--- /dev/null
+++ b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php
@@ -0,0 +1,145 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use RuntimeException;
+
+class BusinessMirrorSchemaManager
+{
+    /** @return list<array<string,mixed>> */
+    public function ensure(BusinessMirrorDataset $dataset): array
+    {
+        $columns = $this->columns($dataset);
+        $connection = (string) config('upstream_mirrors.connection', 'business_mirror');
+        $schema = Schema::connection($connection);
+
+        if (! $schema->hasTable($dataset->targetTable())) {
+            $schema->create($dataset->targetTable(), function (Blueprint $table) use ($columns, $dataset): void {
+                $table->char('_mirror_key', 64)->primary();
+
+                foreach ($columns as $column) {
+                    $this->addColumn($table, $column);
+                }
+
+                $table->uuid('_mirror_generation')->nullable()->index();
+                $table->dateTime('_mirror_synced_at', 6)->index();
+
+                $watermark = $dataset->watermarkColumn();
+                if ($watermark !== null && $this->isIndexable($columns, $watermark)) {
+                    $table->index($watermark, $this->indexName($dataset->targetTable(), 'watermark'));
+                }
+            });
+
+            return $columns;
+        }
+
+        $existing = array_map('strtolower', $schema->getColumnListing($dataset->targetTable()));
+        $missing = array_values(array_filter(
+            $columns,
+            fn (array $column): bool => ! in_array(strtolower((string) $column['column_name']), $existing, true),
+        ));
+
+        if ($missing !== []) {
+            $schema->table($dataset->targetTable(), function (Blueprint $table) use ($missing): void {
+                foreach ($missing as $column) {
+                    $this->addColumn($table, $column);
+                }
+            });
+        }
+
+        return $columns;
+    }
+
+    /** @return list<array<string,mixed>> */
+    public function columns(BusinessMirrorDataset $dataset): array
+    {
+        $rows = DB::table('upstream_schema_columns')
+            ->where('source_key', $dataset->sourceKey())
+            ->whereRaw('LOWER(schema_name) = ?', [strtolower($dataset->sourceSchema())])
+            ->whereRaw('LOWER(object_name) = ?', [strtolower($dataset->sourceObject())])
+            ->orderBy('ordinal_position')
+            ->get()
+            ->map(fn (object $row): array => (array) $row)
+            ->all();
+
+        if ($rows === []) {
+            throw new RuntimeException('business_mirror_schema_catalog_missing:' . $dataset->key());
+        }
+
+        $names = array_map(fn (array $row): string => strtolower((string) $row['column_name']), $rows);
+        if (count($names) !== count(array_unique($names))) {
+            throw new RuntimeException('business_mirror_duplicate_column:' . $dataset->key());
+        }
+
+        foreach ($rows as $row) {
+            $name = (string) $row['column_name'];
+            if (strlen($name) > 64 || str_starts_with(strtolower($name), '_mirror_')) {
+                throw new RuntimeException('business_mirror_unsupported_column:' . $dataset->key());
+            }
+        }
+
+        foreach ([...$dataset->keyColumns(), $dataset->watermarkColumn()] as $required) {
+            if ($required !== null && ! in_array(strtolower($required), $names, true)) {
+                throw new RuntimeException('business_mirror_required_column_missing:' . $dataset->key());
+            }
+        }
+
+        return $rows;
+    }
+
+    private function addColumn(Blueprint $table, array $column): void
+    {
+        $name = (string) $column['column_name'];
+        $type = strtolower((string) $column['data_type']);
+        $length = (int) ($column['max_length'] ?? 0);
+        $precision = min(max((int) ($column['numeric_precision'] ?? 20), 1), 65);
+        $scale = min(max((int) ($column['numeric_scale'] ?? 0), 0), min($precision, 30));
+
+        $definition = match ($type) {
+            'bit' => $table->boolean($name),
+            'tinyint' => $table->unsignedTinyInteger($name),
+            'smallint' => $table->smallInteger($name),
+            'int' => $table->integer($name),
+            'bigint' => $table->bigInteger($name),
+            'decimal', 'numeric', 'money', 'smallmoney' => $table->decimal($name, $precision, $scale),
+            'float', 'real' => $table->double($name),
+            'date' => $table->date($name),
+            'datetime', 'datetime2', 'smalldatetime' => $table->dateTime($name, 6),
+            'time' => $table->time($name, 6),
+            'uniqueidentifier' => $table->char($name, 36),
+            'binary', 'varbinary', 'image', 'timestamp', 'rowversion' => $table->binary($name),
+            'char', 'nchar', 'varchar', 'nvarchar' => $length > 0 && $length <= 1024
+                ? $table->string($name, $length)
+                : $table->longText($name),
+            default => $table->longText($name),
+        };
+
+        $definition->nullable();
+    }
+
+    private function isIndexable(array $columns, string $name): bool
+    {
+        foreach ($columns as $column) {
+            if (strcasecmp((string) $column['column_name'], $name) !== 0) {
+                continue;
+            }
+
+            $type = strtolower((string) $column['data_type']);
+            $length = (int) ($column['max_length'] ?? 0);
+
+            return ! in_array($type, ['text', 'ntext', 'xml', 'image', 'binary', 'varbinary'], true)
+                && (! in_array($type, ['char', 'nchar', 'varchar', 'nvarchar'], true) || ($length > 0 && $length <= 191));
+        }
+
+        return false;
+    }
+
+    private function indexName(string $table, string $suffix): string
+    {
+        return substr($table . '_' . $suffix, 0, 64);
+    }
+}
diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php
new file mode 100644
index 00000000..8ef379b4
--- /dev/null
+++ b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php
@@ -0,0 +1,263 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use Illuminate\Support\Carbon;
+use Illuminate\Support\Str;
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
+use RuntimeException;
+use Throwable;
+
+class BusinessMirrorSynchronizer
+{
+    public function __construct(
+        private readonly BusinessMirrorDatasetRegistry $datasets,
+        private readonly BusinessMirrorSchemaManager $schemas,
+        private readonly GenericBusinessMirrorRepository $mirror,
+        private readonly UpstreamReadOnlyQueryFactory $queries,
+    ) {
+    }
+
+    /** @return array{status:string,mode:string,batches:int,rows:int,dataset:string} */
+    public function sync(string $datasetKey, bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array
+    {
+        $dataset = $this->datasets->get($datasetKey);
+        $state = $this->mirror->state($dataset);
+
+        if (! $full && ($state['initial_completed_at'] ?? null) === null) {
+            return $this->result($dataset, 'requires_full_sync', 'incremental', 0, 0);
+        }
+
+        $columns = $this->schemas->ensure($dataset);
+
+        $mode = $full ? 'full' : 'incremental';
+        $resume = is_array($state)
+            && ($state['run_mode'] ?? null) === $mode
+            && in_array($state['status'] ?? null, ['running', 'failed'], true)
+            && is_array($state['cursor'] ?? null);
+        $generation = $resume && is_string($state['generation'] ?? null)
+            ? (string) $state['generation']
+            : (string) Str::uuid();
+        [$windowStart, $windowEnd] = $this->window($dataset, $state, $mode, $resume);
+        $state = $this->mirror->startRun(
+            $dataset,
+            $mode,
+            $generation,
+            $windowStart?->format('Y-m-d H:i:s.u'),
+            $windowEnd?->format('Y-m-d H:i:s.u'),
+            $resume,
+        );
+        $cursor = is_array($state['cursor'] ?? null) ? $state['cursor'] : null;
+        $highWatermark = is_array($state['high_watermark'] ?? null) ? $state['high_watermark'] : [];
+        $batchSize = min(max($batchSize ?? (int) config('upstream_mirrors.table_sync.batch_size', 500), 1), 2000);
+        $maxBatches = max($maxBatches ?? (int) config('upstream_mirrors.table_sync.max_batches_per_run', 2), 0);
+        $batches = 0;
+        $rows = (int) ($state['last_upserted_rows'] ?? 0);
+
+        try {
+            while (true) {
+                $batch = $this->readBatch(
+                    $dataset,
+                    $columns,
+                    $mode,
+                    $cursor,
+                    $highWatermark,
+                    $windowStart,
+                    $windowEnd,
+                    $batchSize,
+                );
+
+                if ($batch === []) {
+                    $this->mirror->complete($dataset, $mode, $generation, $cursor ?? $highWatermark, $rows);
+
+                    return $this->result($dataset, 'succeeded', $mode, $batches, $rows);
+                }
+
+                $cursor = $this->cursor($dataset, $batch[array_key_last($batch)], $mode);
+                $rows += count($batch);
+                $batches++;
+                $this->mirror->storeBatch($dataset, $columns, $this->normalize($columns, $batch), $cursor, $generation, $rows);
+
+                if (count($batch) < $batchSize) {
+                    $this->mirror->complete($dataset, $mode, $generation, $cursor, $rows);
+
+                    return $this->result($dataset, 'succeeded', $mode, $batches, $rows);
+                }
+
+                if ($maxBatches > 0 && $batches >= $maxBatches) {
+                    return $this->result($dataset, 'running', $mode, $batches, $rows);
+                }
+            }
+        } catch (Throwable $exception) {
+            $this->mirror->fail($dataset, $exception::class);
+            throw $exception;
+        }
+    }
+
+    /** @return list<array<string,mixed>> */
+    private function readBatch(
+        BusinessMirrorDataset $dataset,
+        array $columns,
+        string $mode,
+        ?array $cursor,
+        array $highWatermark,
+        ?Carbon $windowStart,
+        ?Carbon $windowEnd,
+        int $limit,
+    ): array {
+        $names = array_map(fn (array $column): string => (string) $column['column_name'], $columns);
+        $order = $this->orderColumns($dataset, $mode);
+        $where = [];
+        $bindings = [];
+
+        if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) {
+            $watermark = $this->quote($dataset->watermarkColumn());
+            $where[] = $watermark . ' >= ? AND ' . $watermark . ' < ?';
+            $bindings[] = $windowStart?->format('Y-m-d H:i:s.u');
+            $bindings[] = $windowEnd?->format('Y-m-d H:i:s.u');
+        } elseif ($mode === 'incremental' && $cursor === null && $highWatermark !== []) {
+            [$predicate, $values] = $this->keyset($dataset->keyColumns(), $highWatermark);
+            $where[] = $predicate;
+            array_push($bindings, ...$values);
+        }
+
+        if ($cursor !== null) {
+            [$predicate, $values] = $this->keyset($order, $cursor);
+            $where[] = $predicate;
+            array_push($bindings, ...$values);
+        }
+
+        $sql = 'SELECT TOP (' . $limit . ') '
+            . implode(', ', array_map(fn (string $name): string => $this->quote($name), $names))
+            . ' FROM ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject())
+            . ($where === [] ? '' : ' WHERE (' . implode(') AND (', $where) . ')')
+            . ' ORDER BY ' . implode(', ', array_map(fn (string $name): string => $this->quote($name), $order));
+
+        return $this->queries->forSource($dataset->sourceKey())->select($sql, $bindings);
+    }
+
+    /** @return array{0:string,1:list<mixed>} */
+    private function keyset(array $columns, array $cursor): array
+    {
+        $parts = [];
+        $bindings = [];
+
+        foreach (array_values($columns) as $index => $column) {
+            $and = [];
+            for ($previous = 0; $previous < $index; $previous++) {
+                $name = $columns[$previous];
+                $and[] = $this->quote($name) . ' = ?';
+                $bindings[] = $this->cursorValue($cursor, $name);
+            }
+
+            $and[] = $this->quote($column) . ' > ?';
+            $bindings[] = $this->cursorValue($cursor, $column);
+            $parts[] = '(' . implode(' AND ', $and) . ')';
+        }
+
+        return ['(' . implode(' OR ', $parts) . ')', $bindings];
+    }
+
+    /** @return list<string> */
+    private function orderColumns(BusinessMirrorDataset $dataset, string $mode): array
+    {
+        $columns = $dataset->keyColumns();
+
+        if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) {
+            array_unshift($columns, $dataset->watermarkColumn());
+        }
+
+        return array_values(array_unique($columns));
+    }
+
+    /** @return array<string,mixed> */
+    private function cursor(BusinessMirrorDataset $dataset, array $row, string $mode): array
+    {
+        $cursor = [];
+        foreach ($this->orderColumns($dataset, $mode) as $column) {
+            $cursor[$column] = $this->rowValue($row, $column);
+        }
+
+        return $cursor;
+    }
+
+    /** @return array{0:?Carbon,1:?Carbon} */
+    private function window(BusinessMirrorDataset $dataset, ?array $state, string $mode, bool $resume): array
+    {
+        if ($mode === 'full' || $dataset->watermarkColumn() === null) {
+            return [null, null];
+        }
+
+        if ($resume && ($state['window_started_at'] ?? null) !== null && ($state['window_ends_at'] ?? null) !== null) {
+            return [Carbon::parse((string) $state['window_started_at']), Carbon::parse((string) $state['window_ends_at'])];
+        }
+
+        $overlap = max((int) config('upstream_mirrors.table_sync.overlap_hours', 48), 1);
+        $last = Carbon::parse((string) ($state['last_succeeded_at'] ?? now()->subHours($overlap)));
+
+        return [$last->subHours($overlap), now()->addSecond()];
+    }
+
+    /** @return list<array<string,mixed>> */
+    private function normalize(array $columns, array $rows): array
+    {
+        $types = [];
+        foreach ($columns as $column) {
+            $types[strtolower((string) $column['column_name'])] = strtolower((string) $column['data_type']);
+        }
+
+        return array_map(function (array $row) use ($types): array {
+            foreach ($row as $name => $value) {
+                if (! is_string($value)) {
+                    continue;
+                }
+
+                $type = $types[strtolower((string) $name)] ?? '';
+                if ($type === 'date') {
+                    $row[$name] = substr($value, 0, 10);
+                } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime'], true)) {
+                    $row[$name] = preg_replace('/(\.\d{6})\d+/', '$1', str_replace('T', ' ', $value));
+                }
+            }
+
+            return $row;
+        }, $rows);
+    }
+
+    private function quote(string $identifier): string
+    {
+        if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) {
+            throw new RuntimeException('business_mirror_invalid_identifier');
+        }
+
+        return '[' . $identifier . ']';
+    }
+
+    private function cursorValue(array $cursor, string $column): mixed
+    {
+        foreach ($cursor as $key => $value) {
+            if (strcasecmp((string) $key, $column) === 0) {
+                return $value;
+            }
+        }
+
+        throw new RuntimeException('business_mirror_cursor_incomplete');
+    }
+
+    private function rowValue(array $row, string $column): mixed
+    {
+        foreach ($row as $key => $value) {
+            if (strcasecmp((string) $key, $column) === 0) {
+                return $value;
+            }
+        }
+
+        throw new RuntimeException('business_mirror_row_key_missing');
+    }
+
+    private function result(BusinessMirrorDataset $dataset, string $status, string $mode, int $batches, int $rows): array
+    {
+        return ['status' => $status, 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset->key()];
+    }
+}
diff --git a/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php
new file mode 100644
index 00000000..be3455b2
--- /dev/null
+++ b/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php
@@ -0,0 +1,223 @@
+<?php
+
+namespace Tests\Unit\EcommerceFinance;
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Modules\EcommerceFinance\Services\BusinessMirrorSchemaManager;
+use Modules\EcommerceFinance\Services\BusinessMirrorSynchronizer;
+use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
+use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
+use Tests\TestCase;
+
+class BusinessMirrorSynchronizerTest extends TestCase
+{
+    private FakeBusinessMirrorQuery $query;
+    private BusinessMirrorSynchronizer $synchronizer;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+
+        config([
+            'database.connections.business_mirror' => [
+                'driver' => 'sqlite',
+                'database' => ':memory:',
+                'prefix' => '',
+            ],
+            'upstream_mirrors.connection' => 'business_mirror',
+            'upstream_mirrors.datasets' => [
+                'test.dbo.records' => [
+                    'source_key' => 'test',
+                    'schema' => 'dbo',
+                    'object' => 'records',
+                    'target' => 'test_records',
+                    'keys' => ['Id'],
+                    'watermark' => 'updateTime',
+                ],
+            ],
+            'upstream_mirrors.table_sync.overlap_hours' => 24,
+        ]);
+        DB::purge('business_mirror');
+
+        Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $table): void {
+            $table->string('dataset_key')->primary();
+            $table->string('source_key');
+            $table->string('source_object');
+            $table->string('target_table');
+            $table->string('status')->default('pending');
+            $table->string('run_mode')->nullable();
+            $table->string('generation')->nullable();
+            $table->json('cursor')->nullable();
+            $table->json('high_watermark')->nullable();
+            $table->dateTime('window_started_at')->nullable();
+            $table->dateTime('window_ends_at')->nullable();
+            $table->dateTime('initial_completed_at')->nullable();
+            $table->dateTime('last_full_succeeded_at')->nullable();
+            $table->dateTime('last_attempted_at')->nullable();
+            $table->dateTime('last_succeeded_at')->nullable();
+            $table->unsignedBigInteger('last_upserted_rows')->default(0);
+            $table->unsignedBigInteger('mirror_row_count')->default(0);
+            $table->string('last_error_code')->nullable();
+            $table->timestamps();
+        });
+
+        DB::table('upstream_schema_columns')->insert([
+            $this->column('Id', 1, 'int', null),
+            $this->column('updateTime', 2, 'datetime2', null),
+            $this->column('displayName', 3, 'nvarchar', 120),
+        ]);
+
+        $this->query = new FakeBusinessMirrorQuery();
+        $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
+            public function __construct(private readonly UpstreamReadOnlyQuery $query) {}
+            public function forSource(string $source): UpstreamReadOnlyQuery { return $this->query; }
+        };
+        $registry = new BusinessMirrorDatasetRegistry();
+        $this->synchronizer = new BusinessMirrorSynchronizer(
+            $registry,
+            new BusinessMirrorSchemaManager(),
+            new GenericBusinessMirrorRepository(app('db')),
+            $factory,
+        );
+    }
+
+    public function test_full_sync_materializes_complete_table_and_marks_state_ready(): void
+    {
+        $this->query->batches = [[
+            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00.1234567', 'displayName' => '一号'],
+            ['Id' => 2, 'updateTime' => '2026-08-14 11:00:00.0000000', 'displayName' => '二号'],
+        ]];
+
+        $result = $this->synchronizer->sync('test.dbo.records', true, 10, 1);
+
+        $this->assertSame('succeeded', $result['status']);
+        $this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
+        $this->assertSame(
+            '2026-08-14 10:00:00.123456',
+            DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('updateTime'),
+        );
+        $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
+        $this->assertSame('succeeded', $state->status);
+        $this->assertNotNull($state->initial_completed_at);
+        $this->assertSame(2, (int) $state->mirror_row_count);
+        $this->assertStringContainsString('SELECT TOP (10)', $this->query->calls[0]['sql']);
+        $this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[0]['sql']);
+    }
+
+    public function test_incremental_sync_uses_overlap_window_and_updates_existing_key(): void
+    {
+        $this->query->batches = [[
+            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '旧值'],
+        ]];
+        $this->synchronizer->sync('test.dbo.records', true, 10, 1);
+
+        $this->query->calls = [];
+        $this->query->batches = [[
+            ['Id' => 1, 'updateTime' => '2026-08-14 12:00:00', 'displayName' => '新值'],
+            ['Id' => 2, 'updateTime' => '2026-08-14 12:30:00', 'displayName' => '新增'],
+        ]];
+
+        $result = $this->synchronizer->sync('test.dbo.records', false, 10, 1);
+
+        $this->assertSame('succeeded', $result['status']);
+        $this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
+        $this->assertSame('新值', DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('displayName'));
+        $this->assertStringContainsString('[updateTime] >= ? AND [updateTime] < ?', $this->query->calls[0]['sql']);
+        $this->assertStringContainsString('ORDER BY [updateTime], [Id]', $this->query->calls[0]['sql']);
+        $this->assertCount(2, $this->query->calls[0]['bindings']);
+    }
+
+    public function test_incremental_sync_refuses_to_guess_before_first_full_sync(): void
+    {
+        $result = $this->synchronizer->sync('test.dbo.records', false, 10, 1);
+
+        $this->assertSame('requires_full_sync', $result['status']);
+        $this->assertSame([], $this->query->calls);
+    }
+
+    public function test_empty_full_read_does_not_delete_previous_mirror_rows(): void
+    {
+        $this->query->batches = [[
+            ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留'],
+        ]];
+        $this->synchronizer->sync('test.dbo.records', true, 10, 1);
+
+        $this->query->batches = [[]];
+
+        try {
+            $this->synchronizer->sync('test.dbo.records', true, 10, 1);
+            $this->fail('空的上游响应不应覆盖已有镜像。');
+        } catch (\RuntimeException $exception) {
+            $this->assertSame('business_mirror_unexpected_empty_source', $exception->getMessage());
+        }
+
+        $this->assertSame(1, DB::connection('business_mirror')->table('test_records')->count());
+        $this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName'));
+        $this->assertSame(
+            'failed',
+            DB::connection('business_mirror')->table('business_mirror_table_states')->value('status'),
+        );
+    }
+
+    public function test_project_dataset_registry_has_unique_safe_targets(): void
+    {
+        $configured = require base_path('config/upstream_mirrors.php');
+        config(['upstream_mirrors.datasets' => $configured['datasets']]);
+        $datasets = (new BusinessMirrorDatasetRegistry())->all();
+        $targets = array_map(fn ($dataset): string => $dataset->targetTable(), $datasets);
+
+        $this->assertCount(38, $datasets);
+        $this->assertCount(count($targets), array_unique($targets));
+        $this->assertContains('operations_replica', array_map(fn ($dataset): string => $dataset->sourceKey(), $datasets));
+        $this->assertContains('jiansun', array_map(fn ($dataset): string => $dataset->sourceKey(), $datasets));
+        $this->assertNotContains('Bulk_Price_Limit', array_map(fn ($dataset): string => $dataset->sourceObject(), $datasets));
+    }
+
+    /** @return array<string,mixed> */
+    private function column(string $name, int $position, string $type, ?int $length): array
+    {
+        $now = now();
+
+        return [
+            'source_key' => 'test',
+            'schema_name' => 'dbo',
+            'object_name' => 'records',
+            'object_type' => 'table',
+            'column_name' => $name,
+            'ordinal_position' => $position,
+            'data_type' => $type,
+            'max_length' => $length,
+            'numeric_precision' => null,
+            'numeric_scale' => null,
+            'is_nullable' => true,
+            'synchronized_at' => $now,
+            'created_at' => $now,
+            'updated_at' => $now,
+        ];
+    }
+}
+
+class FakeBusinessMirrorQuery extends UpstreamReadOnlyQuery
+{
+    /** @var list<list<array<string,mixed>>> */
+    public array $batches = [];
+
+    /** @var list<array{sql:string,bindings:array}> */
+    public array $calls = [];
+
+    public function __construct()
+    {
+        parent::__construct('test');
+    }
+
+    public function select(string $sql, array $bindings = []): array
+    {
+        $this->calls[] = compact('sql', 'bindings');
+
+        return array_shift($this->batches) ?? [];
+    }
+}