2026-08-31 18:49:00 +08:00
30 changed files with 981 additions and 155 deletions
--- a/app/Console/Commands/BusinessMirrorStatus.php
+++ b/app/Console/Commands/BusinessMirrorStatus.php
@ -0,0 +1,40 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Throwable;
+
+class BusinessMirrorStatus extends Command
+{
+    protected $signature = 'business-mirror:status {dataset?} {--json}';
+    protected $description = '只读查看已登记镜像的启用、首次完成和最近同步状态';
+
+    public function handle(BusinessMirrorDatasetRegistry $datasets, GenericBusinessMirrorRepository $mirror): int
+    {
+        try {
+            $requested = $this->argument('dataset');
+            $selected = $requested ? [$requested => $datasets->get($requested)] : $datasets->all();
+            $rows = [];
+            foreach ($selected as $key => $dataset) {
+                $state = $mirror->state($dataset);
+                $rows[] = [
+                    'dataset' => $key, 'enabled' => $datasets->scheduled($key),
+                    'initial_completed' => ($state['initial_completed_at'] ?? null) !== null,
+                    'status' => $state['status'] ?? ($datasets->automaticInitialSyncAllowed($key) ? 'queued' : 'not_initialized'),
+                    'last_attempted_at' => $state['last_attempted_at'] ?? null,
+                    'last_succeeded_at' => $state['last_succeeded_at'] ?? null,
+                    'error_code' => $state['last_error_code'] ?? null,
+                ];
+            }
+            if ($this->option('json')) $this->line(json_encode($rows, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR));
+            else $this->table(['数据集', '启用', '首次完成', '状态', '最近尝试', '最近完成', '错误类型'], array_map(fn ($r) => array_values($r), $rows));
+            return self::SUCCESS;
+        } catch (Throwable) {
+            $this->error('无法读取镜像状态，请确认共享系统数据库和业务镜像库已接入。');
+            return self::FAILURE;
+        }
+    }
+}
--- a/app/Console/Commands/InspectUpstream.php
+++ b/app/Console/Commands/InspectUpstream.php
@ -0,0 +1,66 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\DB;
+use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
+use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
+use Modules\EcommerceFinance\Services\UpstreamSqlDialect;
+use Throwable;
+
+class InspectUpstream extends Command
+{
+    protected $signature = 'upstream:inspect {source?} {schema?} {object?} {--columns= : 明确选择样本字段；不填时只显示字段目录} {--limit=5 : 样本上限，最多20行}';
+    protected $description = '通过统一驱动查看目录或有限只读样本，不生成快照或启动同步';
+
+    public function handle(UpstreamSourceRegistry $sources, UpstreamReadOnlyQueryFactory $queries): int
+    {
+        $source = (string) $this->argument('source');
+        $schema = (string) $this->argument('schema');
+        $object = (string) $this->argument('object');
+        try {
+            if ($source === '') {
+                $rows = [];
+                foreach ($sources->all() as $key => $definition) $rows[] = [$key, $definition['label'], $sources->driver($key), $sources->enabled($key) ? '启用' : '停用'];
+                $this->table(['内部标识', '数据库名称', '类型', '状态'], $rows);
+                return self::SUCCESS;
+            }
+            $sources->definition($source);
+            if ($schema === '' || $object === '') {
+                $objects = DB::table('upstream_schema_columns')->where('source_key', $source)
+                    ->when($schema !== '', fn ($q) => $q->where('schema_name', $schema))
+                    ->select('schema_name', 'object_name')->distinct()->orderBy('schema_name')->orderBy('object_name')->get();
+                $this->table(['Schema', '表 / 视图'], $objects->map(fn ($row) => [$row->schema_name, $row->object_name])->all());
+                return self::SUCCESS;
+            }
+            $catalog = DB::table('upstream_schema_columns')->where('source_key', $source)
+                ->where('schema_name', $schema)->where('object_name', $object)->orderBy('ordinal_position')
+                ->get(['column_name', 'data_type', 'is_nullable']);
+            if ($catalog->isEmpty()) {
+                $this->error('尚无该对象的字段目录，请先核实来源并同步数据字典。');
+                return self::FAILURE;
+            }
+            if (! $this->option('columns')) {
+                $this->table(['字段', '类型', '允许空值'], $catalog->map(fn ($c) => [$c->column_name, $c->data_type, $c->is_nullable ? '是' : '否'])->all());
+                return self::SUCCESS;
+            }
+            $columns = array_values(array_unique(array_filter(array_map('trim', explode(',', (string) $this->option('columns'))))));
+            if ($columns === [] || array_diff($columns, $catalog->pluck('column_name')->all())) {
+                $this->error('样本字段必须来自该对象的数据字典。');
+                return self::FAILURE;
+            }
+            $dialect = new UpstreamSqlDialect($sources->driver($source));
+            $sql = $dialect->select(implode(', ', array_map($dialect->quote(...), $columns)),
+                'FROM ' . $dialect->quote($schema) . '.' . $dialect->quote($object),
+                min(max((int) $this->option('limit'), 1), 20));
+            $rows = $queries->forSource($source)->select($sql);
+            $this->table($columns, array_map(fn ($row) => array_values($row), $rows));
+            $this->info('样本仅用于字段核实，不能作为完整业务数据。');
+            return self::SUCCESS;
+        } catch (Throwable) {
+            $this->error('只读检查未完成，请检查共享连接、来源启用状态和访问权限。');
+            return self::FAILURE;
+        }
+    }
+}
--- a/app/Console/Commands/RegisterBusinessMirror.php
+++ b/app/Console/Commands/RegisterBusinessMirror.php
@ -0,0 +1,94 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
+use Throwable;
+
+class RegisterBusinessMirror extends Command
+{
+    protected $signature = 'business-mirror:register {source} {schema} {object}
+        {--keys= : 逗号分隔的稳定唯一键}
+        {--columns= : 逗号分隔的明确同步字段，必须包含稳定键及增量字段}
+        {--watermark= : 可靠更新时间字段；留空为仅按主键追加}
+        {--enable-sync : 启用并批准可断点首次同步及后续增量}
+        {--disable-sync : 停用此数据集的自动同步，保留镜像数据}';
+
+    protected $description = '登记声明式镜像数据集；生产启用前需获得同步范围授权';
+
+    public function handle(BusinessMirrorDatasetRegistry $datasets, UpstreamSourceRegistry $sources): int
+    {
+        if (! Schema::hasTable('business_mirror_datasets')) {
+            $this->error('尚未执行镜像数据集登记表迁移。');
+            return self::FAILURE;
+        }
+        $source = (string) $this->argument('source');
+        $schema = (string) $this->argument('schema');
+        $object = (string) $this->argument('object');
+        $key = $source . '.' . $schema . '.' . $object;
+        try {
+            $sources->definition($source);
+            if (strlen($key) > 191 || isset(config('upstream_mirrors.datasets', [])[$key])) {
+                throw new \RuntimeException('数据集键过长或由现有代码维护，不能覆盖。');
+            }
+            if ($this->option('disable-sync')) {
+                if ($this->option('enable-sync')) throw new \RuntimeException('不能同时启用和停用。');
+                if (DB::table('business_mirror_datasets')->where('dataset_key', $key)->update(['sync_enabled' => false, 'updated_at' => now()]) !== 1) {
+                    throw new \RuntimeException('未找到该登记数据集。');
+                }
+                $this->info('已停止自动同步，现有镜像保留。');
+                return self::SUCCESS;
+            }
+            $keys = $this->csv((string) $this->option('keys'));
+            $columns = $this->csv((string) $this->option('columns'));
+            $watermark = trim((string) $this->option('watermark'));
+            if ($keys === [] || $columns === [] || array_diff([...$keys, ...($watermark === '' ? [] : [$watermark])], $columns)) {
+                throw new \RuntimeException('请明确字段和稳定键；同步字段必须包含稳定键及增量字段。');
+            }
+            $catalog = DB::table('upstream_schema_columns')->where('source_key', $source)
+                ->where('schema_name', $schema)->where('object_name', $object)->pluck('column_name')->all();
+            if (array_diff($columns, $catalog)) {
+                throw new \RuntimeException('字段不在数据字典中，请先同步该来源字典并核实字段。');
+            }
+            $definition = ['source_key' => $source, 'schema' => $schema, 'object' => $object,
+                'target' => 'mirror_' . substr(hash('sha256', $key), 0, 24),
+                'keys' => $keys, 'columns' => $columns, 'watermark' => $watermark];
+            $datasets->validate($key, $definition);
+            $enable = (bool) $this->option('enable-sync');
+            DB::transaction(function () use ($key, $source, $definition, $enable): void {
+                $existing = DB::table('business_mirror_datasets')->where('dataset_key', $key)->lockForUpdate()->first();
+                if ($existing !== null && json_decode($existing->definition, true) != $definition) {
+                    throw new \RuntimeException('已有数据集定义不同，不能覆盖稳定键、字段或同步游标。请先完成变更评估。');
+                }
+                if ($existing !== null) {
+                    if ($enable) DB::table('business_mirror_datasets')->where('dataset_key', $key)->update([
+                        'sync_enabled' => true, 'initial_sync_approved' => true, 'updated_at' => now(),
+                    ]);
+                    return;
+                }
+                DB::table('business_mirror_datasets')->insert([
+                    'dataset_key' => $key, 'source_key' => $source, 'definition' => json_encode($definition, JSON_THROW_ON_ERROR),
+                    'sync_enabled' => $enable, 'initial_sync_approved' => $enable, 'created_at' => now(), 'updated_at' => now(),
+                ]);
+            });
+            $this->info($key . ($enable ? '：已启用，生产调度将自动执行首次同步并持续增量。' : '：已登记，自动同步未启用。'));
+            return self::SUCCESS;
+        } catch (\Illuminate\Database\QueryException) {
+            $this->error('登记失败，请检查系统数据库状态；原数据集保留。');
+        } catch (\RuntimeException|\InvalidArgumentException $e) {
+            $this->error($e->getMessage());
+        } catch (Throwable) {
+            $this->error('登记失败，未启用新同步。请检查系统数据库状态。');
+        }
+        return self::FAILURE;
+    }
+
+    private function csv(string $value): array
+    {
+        return array_values(array_unique(array_filter(array_map('trim', explode(',', $value)))));
+    }
+}
--- a/app/Console/Commands/SyncBusinessMirrors.php
+++ b/app/Console/Commands/SyncBusinessMirrors.php
@ -35,23 +35,27 @@ class SyncBusinessMirrors extends Command
        $requested = $this->argument('dataset');
        $keys = $requested === null
            ? $this->scheduledKeys($datasets, $mirror)
-            : [(string) $requested];
+            : [['key' => (string) $requested, 'full' => (bool) $this->option('full'), 'catchup' => false]];
        $maxDatasets = max((int) ($this->option('max-datasets')
-            ?? config('upstream_mirrors.table_sync.max_datasets_per_run', 1)), 1);
+            ?? config('upstream_pipeline.sync.max_datasets', 8)), 1);
        $processed = 0;
        $failed = false;
+        $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.cycle_seconds', 30), 1);

-        foreach ($keys as $key) {
-            if ($processed >= $maxDatasets) {
+        foreach ($keys as $task) {
+            if ($processed >= $maxDatasets || ($processed > 0 && microtime(true) >= $deadline)) {
                break;
            }
+            $key = $task['key'];

            try {
                $result = $synchronizer->sync(
                    $key,
-                    (bool) $this->option('full'),
+                    (bool) $task['full'],
                    $this->option('batch') === null ? null : (int) $this->option('batch'),
-                    $this->option('max-batches') === null ? null : (int) $this->option('max-batches'),
+                    $this->option('max-batches') === null
+                        ? ($task['catchup'] ? max((int) config('upstream_pipeline.sync.catchup_batches', 6), 1) : null)
+                        : (int) $this->option('max-batches'),
                );
            } catch (Throwable $exception) {
                $failed = true;
@ -105,7 +109,7 @@ class SyncBusinessMirrors extends Command
        return $failed ? self::FAILURE : self::SUCCESS;
    }

-    /** @return list<string> */
+    /** @return list<array{key:string,full:bool,catchup:bool,last_attempted_at:string}> */
    private function scheduledKeys(
        BusinessMirrorDatasetRegistry $datasets,
        GenericBusinessMirrorRepository $mirror,
@ -117,17 +121,20 @@ class SyncBusinessMirrors extends Command
        ), true);

        foreach ($datasets->all() as $key => $dataset) {
-            if (isset($dedicated[$key])) {
+            if (isset($dedicated[$key]) || ! $datasets->scheduled($key)) {
                continue;
            }

            $state = $mirror->state($dataset);
-            if (! is_array($state) || ($state['initial_completed_at'] ?? null) === null) {
-                continue;
-            }
+            $initial = ! is_array($state) || ($state['initial_completed_at'] ?? null) === null;
+            if ($initial && ! $datasets->automaticInitialSyncAllowed($key)) continue;
+            if (($state['status'] ?? '') === 'failed' && ! empty($state['last_attempted_at'])
+                && now()->diffInSeconds(\Illuminate\Support\Carbon::parse($state['last_attempted_at'])) < 300) continue;

            $scheduled[] = [
                'key' => $key,
+                'full' => $initial || (bool) $this->option('full'),
+                'catchup' => $initial || ($state['status'] ?? '') === 'running',
                'last_attempted_at' => (string) ($state['last_attempted_at'] ?? ''),
            ];
        }
@ -136,6 +143,15 @@ class SyncBusinessMirrors extends Command
            [$left['last_attempted_at'], $left['key']] <=> [$right['last_attempted_at'], $right['key']]
        );

-        return array_column($scheduled, 'key');
+        // Alternate unfinished windows and routine refreshes: catch-up progresses
+        // every cycle without starving small tables or monopolizing a source.
+        $catchup = array_values(array_filter($scheduled, fn ($task) => $task['catchup']));
+        $routine = array_values(array_filter($scheduled, fn ($task) => ! $task['catchup']));
+        $ordered = [];
+        while ($catchup !== [] || $routine !== []) {
+            if ($routine !== []) $ordered[] = array_shift($routine);
+            if ($catchup !== []) $ordered[] = array_shift($catchup);
+        }
+        return $ordered;
    }
 }
--- a/app/Console/Commands/UpstreamSyncSchemaCatalog.php
+++ b/app/Console/Commands/UpstreamSyncSchemaCatalog.php
@ -32,7 +32,7 @@ class UpstreamSyncSchemaCatalog extends Command
        $requested = $this->argument('source');
        $sourceKeys = $requested
            ? [(string) $requested]
-            : array_keys((array) config('upstreams.sources', []));
+            : array_keys($sources->all());
        $cache = $cacheManager->store((string) config('upstreams.snapshots.cache_store', 'upstream_shared'));
        $failed = false;

--- a/app/Http/Controllers/SystemSettings/Databases.php
+++ b/app/Http/Controllers/SystemSettings/Databases.php
@ -211,7 +211,7 @@ class Databases extends Controller

        $data = $this->validateConnection($request);
        $connection = DatabaseConnection::create($this->attributes($data, null, $request));
-        $this->resetSource($connection->source_key, $storedConfiguration);
+        $this->resetSource($connection->sourceIdentifier(), $storedConfiguration);

        flash('数据库配置已新增。')->success();

@ -234,12 +234,12 @@ class Databases extends Controller
    {
        abort_unless($storedConfiguration->editable() && $storedConfiguration->enabled(), 403);
        $connection = $this->connection($connection);
-        $oldSource = $connection->source_key;
+        $oldSource = $connection->sourceIdentifier();
        $data = $this->validateConnection($request, $connection);
        $connection->update($this->attributes($data, $connection, $request));

        $this->resetSource($oldSource, $storedConfiguration);
-        $this->resetSource($connection->source_key, $storedConfiguration);
+        $this->resetSource($connection->sourceIdentifier(), $storedConfiguration);

        flash('数据库配置已修改。')->success();

@ -250,7 +250,7 @@ class Databases extends Controller
    {
        abort_unless($storedConfiguration->editable() && $storedConfiguration->enabled(), 403);
        $connection = $this->connection($connection);
-        $source = $connection->source_key;
+        $source = $connection->sourceIdentifier();
        $connection->delete();
        $this->resetSource($source, $storedConfiguration);

@ -272,13 +272,14 @@ class Databases extends Controller
            ]);
        } catch (ServiceUnavailableHttpException $exception) {
            throw ValidationException::withMessages([
-                'connection' => $connection->driver === 'pgsql'
+                'connection' => $connection->driver !== 'sqlsrv'
                    ? $exception->getMessage()
                    : 'SQL Server 连接失败。请检查地址、端口、数据库、账号、网络白名单和 ODBC 驱动。',
            ]);
        }

-        flash($connection->name . ($connection->driver === 'pgsql' ? ' PostgreSQL SSL 连接成功，仅执行了 SELECT 1 只读测试。' : '连接成功，仅执行了 SELECT 1 只读测试。'))->success();
+        $protocol = match ($connection->driver) { 'pgsql' => ' PostgreSQL SSL ', 'mysql' => ' MySQL/MariaDB TLS ', default => '' };
+        flash($connection->name . $protocol . '连接成功，仅执行了 SELECT 1 只读测试。')->success();

        return redirect()->route('system-settings.databases.index');
    }
@ -287,16 +288,16 @@ class Databases extends Controller
    {
        $sourceKeys = array_keys($this->sourceLabels());
        $request->merge(['driver' => $request->input('driver', $connection?->driver ?? 'sqlsrv')]);
-        $postgres = $request->input('driver') === 'pgsql';
+        $postgres = in_array($request->input('driver'), ['pgsql', 'mysql'], true);

        if ($postgres && $request->input('source_key', $connection?->source_key)) {
            throw ValidationException::withMessages([
-                'source_key' => '现有业务来源使用 SQL Server，不能改绑到 PostgreSQL。请为 PostgreSQL 单独保存连接。',
+                'source_key' => '现有业务来源使用 SQL Server，不能更换其数据库类型。请单独添加新数据库。',
            ]);
        }

        return $request->validate([
-            'driver' => ['required', Rule::in(['sqlsrv', 'pgsql'])],
+            'driver' => ['required', Rule::in(['sqlsrv', 'pgsql', 'mysql'])],
            'name' => [
                'required',
                'string',
@ -342,11 +343,12 @@ class Databases extends Controller
            'port' => (int) $data['port'],
            'database_name' => trim($data['database_name']),
            'username' => trim($data['username']),
-            'odbc_driver' => $data['driver'] === 'pgsql' ? null : trim($data['odbc_driver']),
-            'tds_version' => $data['driver'] === 'pgsql' ? null : trim($data['tds_version']),
+            'odbc_driver' => $data['driver'] !== 'sqlsrv' ? null : trim($data['odbc_driver']),
+            'tds_version' => $data['driver'] !== 'sqlsrv' ? null : trim($data['tds_version']),
            'connect_timeout' => (int) $data['connect_timeout'],
            'query_timeout' => (int) $data['query_timeout'],
-            'allowed_scopes' => $this->parseScopes((string) ($data['allowed_scopes'] ?? '')),
+            'allowed_scopes' => array_key_exists('allowed_scopes', $data)
+                ? $this->parseScopes((string) $data['allowed_scopes']) : ($connection?->allowed_scopes ?? []),
            'charset' => 'UTF-8',
            'enabled' => $request->boolean('enabled'),
        ];
@ -382,7 +384,7 @@ class Databases extends Controller

    private function sourceLabels(): array
    {
-        return collect((array) config('upstreams.sources', []))
+        return collect(app(UpstreamSourceRegistry::class)->all())
            ->mapWithKeys(fn (array $source, string $key) => [$key => $source['label'] ?? $key])
            ->all();
    }
@ -393,7 +395,7 @@ class Databases extends Controller
            'name' => '',
            'driver' => 'sqlsrv',
            'source_key' => '',
-            'enabled' => false,
+            'enabled' => true,
            'host' => '',
            'port' => 1433,
            'database_name' => '',
@ -460,6 +462,6 @@ class Databases extends Controller
        }

        $storedConfiguration->forget($source);
-        UpstreamOdbcConnection::reset($source);
+        \Modules\EcommerceFinance\Services\UpstreamConnection::reset($source);
    }
 }
--- a/app/Models/Settings/DatabaseConnection.php
+++ b/app/Models/Settings/DatabaseConnection.php
@ -65,4 +65,10 @@ class DatabaseConnection extends Model
    {
        return $this->password;
    }
+
+    public function sourceIdentifier(): string
+    {
+        // Legacy names stay stable; new connections need no manually assigned binding.
+        return $this->source_key ?: 'database_' . $this->id;
+    }
 }
--- a/app/Utilities/SystemDatabaseConfiguration.php
+++ b/app/Utilities/SystemDatabaseConfiguration.php
@ -178,6 +178,12 @@ class SystemDatabaseConfiguration
        $updated = $this->replaceEnvironmentValues($current, [
            'DB_CONNECTION' => 'mysql',
            'DB_HOST' => (string) $settings['host'],
+            // Laravel prioritizes split read/write hosts, URL and socket over
+            // DB_HOST. They must point to the same newly tested connection.
+            'DB_HOST_READ' => (string) $settings['host'],
+            'DB_HOST_WRITE' => (string) $settings['host'],
+            'DATABASE_URL' => '',
+            'DB_SOCKET' => '',
            'DB_PORT' => (string) $settings['port'],
            'DB_DATABASE' => (string) $settings['database'],
            'DB_USERNAME' => (string) $settings['username'],
@ -555,10 +561,9 @@ class SystemDatabaseConfiguration
            $found = false;

            foreach ($lines as $index => $line) {
-                if (preg_match('/^' . preg_quote($key, '/') . '=/', $line) === 1) {
+                if (preg_match('/^\s*(?:export\s+)?' . preg_quote($key, '/') . '\s*=/', $line) === 1) {
                    $lines[$index] = $replacement;
                    $found = true;
-                    break;
                }
            }

--- a/config/upstream_pipeline.php
+++ b/config/upstream_pipeline.php
@ -0,0 +1,12 @@
+<?php
+
+return [
+    // Connection records supply addresses and credentials, never these defaults.
+    'defaults' => ['enabled' => false, 'connection' => ['connect_timeout' => 5, 'query_timeout' => 15]],
+    'sync' => [
+        'max_datasets' => (int) env('BUSINESS_MIRROR_MAX_DATASETS', 8),
+        'cycle_seconds' => (int) env('BUSINESS_MIRROR_CYCLE_SECONDS', 30),
+        'dataset_seconds' => (int) env('BUSINESS_MIRROR_DATASET_SECONDS', 8),
+        'catchup_batches' => (int) env('BUSINESS_MIRROR_CATCHUP_BATCHES', 6),
+    ],
+];
--- a/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php
+++ b/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php
@ -0,0 +1,24 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration {
+    public function up(): void
+    {
+        Schema::create('business_mirror_datasets', function (Blueprint $table): void {
+            $table->string('dataset_key', 191)->primary();
+            $table->string('source_key', 64)->index();
+            $table->json('definition');
+            $table->boolean('sync_enabled')->default(false);
+            $table->boolean('initial_sync_approved')->default(false);
+            $table->timestamps();
+        });
+    }
+
+    public function down(): void
+    {
+        Schema::dropIfExists('business_mirror_datasets');
+    }
+};
--- a/docs/开发与部署SOP.md
+++ b/docs/开发与部署SOP.md
@ -163,6 +163,8 @@ git pull --ff-only

 ## 8. 多上游财务数据库同步

+新数据库和新数据集优先使用 [通用上游数据接入](通用上游数据接入.md)：来源由连接记录自动登记，已批准的声明式数据集由调度自动首次同步和增量同步。下面的固定来源、逐项 `--full` 和旧快照说明适用于既有兼容路径，不应当作为新增普通开发任务的重复审批要求。
+
 ### 8.1 统一接入边界

 本系统按 3 个以上上游数据库设计。生产系统数据库的启动连接只存在于服务器私有 `runtime/.env`；本地新人只填写系统数据库密码，即可自动解锁两套上游的固定主机、端口、库名、账号和密码，不需要复制生产 `APP_KEY`、填写上游环境变量或等待负责人开放范围。
--- a/docs/通用上游数据接入.md
+++ b/docs/通用上游数据接入.md
@ -0,0 +1,49 @@
+# 通用上游数据接入
+
+## 使用边界
+
+财务主库仍为团队 MariaDB/MySQL。SQL Server、PostgreSQL、MySQL/MariaDB 上游共用来源发现、字典、有限样本查询和表级镜像入口。添加同类型的第二个数据库不需要修改来源枚举、工厂或刷新器。
+
+数据库连接记录未设置历史 `source_key` 时，系统使用稳定标识 `database_<记录ID>`。已有 `jiansun`、`operations_replica`、`js_data_middle_platform` 等映射保留。普通连接表单不再要求手工绑定来源或填写旧快照 scope；编辑旧连接仍保留这些历史字段。
+
+原生 PostgreSQL/MySQL 连接统一要求 TLS、只读会话、非持久连接及最长 15 秒查询。SQL Server 继续复用已验证的 ODBC/DBLIB 入口。应用限制不替代上游只读账号；不提供接受用户原始 SQL 的网页接口。
+
+## 开发流程（由 Codex 执行）
+
+1. 新人通过标准初始化启动本地环境，首次输入一次团队系统数据库密码。Docker healthy 只证明当前程序可用，`system-database:verify` 成功才表示共享接入通过。普通页面、计算逻辑和自动测试可先使用明确的隔离 fixture；不得把模拟数据当作真实财务数据。
+2. 在数据库管理页面添加并启用连接；“测试”与实际读取共用同一驱动。生产及现有字典刷新任务会发现新来源，不再要求逐库添加专用刷新器。
+3. `php artisan upstream:inspect` 列出来源标识，不输出凭据。`upstream:inspect <source>` 列出已有字典对象；`upstream:inspect <source> <schema> <object>` 查看字段。
+4. 缺少目录时运行 `upstream:schema:sync <source>`。此命令只读上游元数据、写入系统库目录，失败保留原目录。
+5. 需要样本时明确指定 `--columns=id,updated_at,label --limit=5`，最多返回20行。只选本任务有权读取的非敏感字段；真实样本不得进入 Git、PR 或公开日志。样本不能证明全量完整性。
+6. 确认业务键、需要的字段、增量依据和业务含义后，登记数据集。以下仅为示例，表名必须来自实际目录：
+
+```sh
+php artisan business-mirror:register database_17 public example_orders \
+  --keys=id --columns=id,updated_at,label --watermark=updated_at
+```
+
+登记默认不启动同步。生产首次同步涉及系统库登记和业务镜像表创建/写入，应先说明精确字段、批次、影响及备份，再取得负责人授权。授权后对同一声明追加 `--enable-sync`，生产调度即可自动完成或续传首次全量，完成后转为增量；不再为每次部署、分页、筛选重新审批或生成快照。`--disable-sync` 停止后续自动执行，保留已有镜像。
+
+声明式登记不能覆盖代码维护的数据集，也不能悄悄修改已有声明的稳定键、字段和增量口径；此类变化需评估历史数据/游标兼容性。来源表没有可靠更新时间时只按主键追加，不能承诺自动发现历史修改或删除，需另行声明校准策略。
+
+## 页面读取与同步状态
+
+新页面在原有公司、角色和数据权限校验之后调用 `BusinessMirrorDataReader::paginate($datasetKey, $filters, $page, $perPage)`。它只读业务镜像，返回完整总数、稳定分页、实际同步状态和最近完成时间。来源暂时关闭或不可达时，已完成镜像仍可读取；无首次镜像时明确报“尚未完成首次同步”，不登记逐页快照、不从页面直接访问上游。
+
+通用筛选只接受已声明字段的等值条件；复杂业务联表、金额口径、公司行级过滤继续由固定 Repository 实现。不能把此内部读取器当作绕过业务权限的公开接口。
+
+`php artisan business-mirror:status [dataset] --json` 只读输出启用、首次完成、最近尝试、最近完成及错误类型。`running` 表示一个同步窗口尚未完成，不保证此刻正有进程运行。最后尝试时间不是最后数据完成时间。
+
+通用调度默认每轮最多8个数据集、30秒周期预算；未完成窗口与常规数据集交替处理，单数据集默认8秒软预算，积压任务默认最多6批。预算在批次之间检查，单条查询仍受驱动超时限制；不提高上游并发。失败任务至少冷却5分钟。专用商品/价格/钉钉调度保持既有路径。
+
+只对明确批准的登记数据集自动首次同步；既有代码数据集仍保留原首次授权流程。未完成的旧页面切换不等于已经上线，必须逐页验收并与正在开发的 PR 协调。
+
+## 类型及验收
+
+- PostgreSQL 带时区时间戳在镜像中保存为 UTC；业务展示按自己的明确时区换算。无时区日期时间保留原值。`bytea` 使用十六进制文本表示；不自动解释 JSON、数组、枚举中的业务语义。
+- 验收包含：没有修改来源枚举也能发现新库；TLS/只读保护；字段目录；只复制明确字段；首次同步、第二页、末页、总数和筛选；增量插入与更新；源不可达时读旧镜像；失败和未授权状态。
+- SQL Server 既有 Repository 与专用同步器需回归。不得把“新连接测试成功”当作全项目页面已迁移或所有业务表已同步。
+
+## 系统连接保存后回滚的排查
+
+截图中的“新配置未通过应用级验证”表示已尝试保存并复核，不等于未输入密码。保存系统连接必须同时统一 `DB_HOST`、读写地址，清除旧 `DATABASE_URL` 和 socket 覆盖，并处理重复赋值；验证失败时恢复完整旧配置。不要只改 `SYSTEM_DATABASE_MODE` 标志，也不要关闭失败回滚来制造“连接成功”。
--- a/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php
+++ b/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php
@ -0,0 +1,46 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use InvalidArgumentException;
+use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
+use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
+
+/** Backend primitive: callers must retain their existing company/permission checks. */
+class BusinessMirrorDataReader
+{
+    public function __construct(
+        private readonly BusinessMirrorDatasetRegistry $datasets,
+        private readonly BusinessMirrorSchemaManager $schemas,
+        private readonly GenericBusinessMirrorRepository $mirror,
+    ) {}
+
+    public function paginate(string $key, array $filters = [], int $page = 1, int $perPage = 25): array
+    {
+        $dataset = $this->datasets->get($key);
+        if (! $this->mirror->ready($dataset)) {
+            throw new ServiceUnavailableHttpException(30, '该数据集尚未完成首次同步，请查看同步状态；页面不会临时访问上游。');
+        }
+        $columns = array_column($this->schemas->columns($dataset), 'column_name');
+        $query = $this->mirror->connection()->table($dataset->targetTable());
+        foreach ($filters as $column => $value) {
+            if (! in_array($column, $columns, true) || (! is_scalar($value) && $value !== null)) {
+                throw new InvalidArgumentException('镜像筛选字段或值无效。');
+            }
+            $query->where($column, $value);
+        }
+        $total = (clone $query)->count();
+        foreach ($dataset->keyColumns() as $column) $query->orderBy($column);
+        $page = max($page, 1);
+        $perPage = min(max($perPage, 1), 100);
+        $state = $this->mirror->state($dataset);
+        return [
+            'rows' => $query->select($columns)->offset(($page - 1) * $perPage)->limit($perPage)->get()
+                ->map(fn ($row) => (array) $row)->all(),
+            'total' => $total, 'page' => $page, 'per_page' => $perPage,
+            'last_page' => max((int) ceil($total / $perPage), 1),
+            'sync_status' => $state['status'] ?? 'unknown',
+            'last_succeeded_at' => $state['last_succeeded_at'] ?? null,
+        ];
+    }
+}
--- a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php
+++ b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php
@ -3,6 +3,8 @@
 namespace Modules\EcommerceFinance\Services;

 use InvalidArgumentException;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
 use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;

 class BusinessMirrorDatasetRegistry
@ -12,7 +14,7 @@ class BusinessMirrorDatasetRegistry
    {
        $datasets = [];

-        foreach ((array) config('upstream_mirrors.datasets', []) as $key => $definition) {
+        foreach ($this->definitions() as $key => $definition) {
            $datasets[(string) $key] = $this->make((string) $key, (array) $definition);
        }

@ -21,7 +23,7 @@ class BusinessMirrorDatasetRegistry

    public function get(string $key): BusinessMirrorDataset
    {
-        $definitions = (array) config('upstream_mirrors.datasets', []);
+        $definitions = $this->definitions();
        $definition = (array) ($definitions[$key] ?? []);

        if ($definition === []) {
@ -31,6 +33,43 @@ class BusinessMirrorDatasetRegistry
        return $this->make($key, $definition);
    }

+    private function definitions(): array
+    {
+        $definitions = (array) config('upstream_mirrors.datasets', []);
+        if (Schema::hasTable('business_mirror_datasets')) {
+            foreach (DB::table('business_mirror_datasets')->get() as $row) {
+                // Existing code-owned datasets cannot be redirected by a new registration.
+                $definitions[$row->dataset_key] ??= json_decode($row->definition, true, 32, JSON_THROW_ON_ERROR);
+            }
+        }
+        return $definitions;
+    }
+
+    public function automaticInitialSyncAllowed(string $key): bool
+    {
+        return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets')
+            ->where('dataset_key', $key)->where('sync_enabled', true)->where('initial_sync_approved', true)->exists();
+    }
+
+    public function scheduled(string $key): bool
+    {
+        $definition = $this->definitions()[$key] ?? null;
+        if (! $definition || ! app(UpstreamSourceRegistry::class)->enabled($definition['source_key'])) return false;
+        if (array_key_exists($key, (array) config('upstream_mirrors.datasets', []))) return true;
+        return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets')
+            ->where('dataset_key', $key)->where('sync_enabled', true)->exists();
+    }
+
+    public function validate(string $key, array $definition): BusinessMirrorDataset
+    {
+        return $this->make($key, $definition);
+    }
+
+    public function columnSelection(string $key): ?array
+    {
+        return $this->definitions()[$key]['columns'] ?? null;
+    }
+
    private function make(string $key, array $definition): BusinessMirrorDataset
    {
        $source = trim((string) ($definition['source_key'] ?? ''));
@ -45,17 +84,23 @@ class BusinessMirrorDatasetRegistry
        ));
        $indexColumns = array_merge(...($indexes ?: [[]]));

-        foreach ([$source, $schema, $object, $target, ...$keys, ...$indexColumns] as $identifier) {
+        foreach ([$source, $target] as $identifier) {
            if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) {
                throw new InvalidArgumentException('业务镜像数据集包含不安全的标识符：' . $key);
            }
        }

+        foreach ([$schema, $object, ...$keys, ...$indexColumns] as $identifier) {
+            if ($identifier === '' || mb_strlen($identifier) > 128 || str_contains($identifier, "\0")) {
+                throw new InvalidArgumentException('业务镜像数据集包含无效的标识符：' . $key);
+            }
+        }
+
        if ($keys === [] || in_array([], $indexes, true) || strlen($target) > 64) {
            throw new InvalidArgumentException('业务镜像数据集缺少稳定键或目标表名过长：' . $key);
        }

-        if ($watermark !== '' && preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $watermark) !== 1) {
+        if ($watermark !== '' && (mb_strlen($watermark) > 128 || str_contains($watermark, "\0"))) {
            throw new InvalidArgumentException('业务镜像数据集的增量字段不安全：' . $key);
        }

--- a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php
+++ b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php
@ -6,6 +6,7 @@ use Illuminate\Database\Schema\Blueprint;
 use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\Schema;
 use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
 use RuntimeException;

 class BusinessMirrorSchemaManager
@ -87,6 +88,13 @@ class BusinessMirrorSchemaManager
            ->map(fn (object $row): array => (array) $row)
            ->all();

+        $selected = app(BusinessMirrorDatasetRegistry::class)->columnSelection($dataset->key());
+        if ($selected !== null) {
+            $selected = array_map('strtolower', $selected);
+            $rows = array_values(array_filter($rows, fn (array $row): bool => in_array(strtolower($row['column_name']), $selected, true)));
+            if (count($rows) !== count($selected)) throw new RuntimeException('business_mirror_selected_column_missing:' . $dataset->key());
+        }
+
        if ($rows === []) {
            throw new RuntimeException('business_mirror_schema_catalog_missing:' . $dataset->key());
        }
@ -121,24 +129,24 @@ class BusinessMirrorSchemaManager
        $scale = min(max((int) ($column['numeric_scale'] ?? 0), 0), min($precision, 30));

        $definition = match ($type) {
-            'bit' => $table->boolean($name),
+            'bit', 'boolean' => $table->boolean($name),
            'tinyint' => $table->unsignedTinyInteger($name),
            'smallint' => $table->smallInteger($name),
-            'int' => $table->integer($name),
+            'int', 'integer' => $table->integer($name),
            'bigint' => $table->bigInteger($name),
            'decimal', 'numeric', 'money', 'smallmoney' => $table->decimal($name, $precision, $scale),
-            'float', 'real' => $table->double($name),
+            'float', 'real', 'double', 'double precision' => $table->double($name),
            'date' => $table->date($name),
-            'datetime', 'datetime2', 'smalldatetime' => $table->dateTime($name, 6),
+            'datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone' => $table->dateTime($name, 6),
            'time' => $table->time($name, 6),
-            'uniqueidentifier' => $table->char($name, 36),
+            'uniqueidentifier', 'uuid' => $table->char($name, 36),
            'binary', 'varbinary', 'image', 'timestamp', 'rowversion' => $table->binary($name),
            // MariaDB counts every inline varchar column towards the 65,535-byte
            // row limit. Wide legacy SQL Server tables can contain dozens of
            // nvarchar columns even though each individual column looks small.
            // Keep only genuinely short values inline; TEXT is stored off-page
            // and lets the mirror preserve the complete source schema.
-            'char', 'nchar', 'varchar', 'nvarchar' => $length > 0 && $length <= 64
+            'char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying' => $length > 0 && $length <= 64
                ? $table->string($name, $length)
                : $table->longText($name),
            default => $table->longText($name),
@ -157,8 +165,8 @@ class BusinessMirrorSchemaManager
            $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 <= 64));
+            return ! in_array($type, ['text', 'ntext', 'xml', 'image', 'binary', 'varbinary', 'bytea', 'json', 'jsonb', 'array', 'user-defined'], true)
+                && (! in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying'], true) || ($length > 0 && $length <= 64));
        }

        return false;
--- a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php
+++ b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php
@ -11,6 +11,7 @@ use Throwable;

 class BusinessMirrorSynchronizer
 {
+    private string $driver = 'sqlsrv';
    public function __construct(
        private readonly BusinessMirrorDatasetRegistry $datasets,
        private readonly BusinessMirrorSchemaManager $schemas,
@ -23,10 +24,11 @@ class BusinessMirrorSynchronizer
    public function sync(string $datasetKey, bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array
    {
        $dataset = $this->datasets->get($datasetKey);
+        $this->driver = app(UpstreamSourceRegistry::class)->driver($dataset->sourceKey());
        // A long-running scheduler may have retained a timed-out PDO instance
        // from the previous dataset. Each dataset starts with a fresh upstream
        // connection so one source failure cannot poison subsequent mirrors.
-        UpstreamOdbcConnection::reset($dataset->sourceKey());
+        UpstreamConnection::reset($dataset->sourceKey());
        $state = $this->mirror->state($dataset);

        if (! $full && ($state['initial_completed_at'] ?? null) === null) {
@ -60,6 +62,7 @@ class BusinessMirrorSynchronizer
        $batchSize = min(max($batchSize ?? (int) config('upstream_mirrors.table_sync.batch_size', 500), 1), 5000);
        $maxBatches = max($maxBatches ?? (int) config('upstream_mirrors.table_sync.max_batches_per_run', 2), 0);
        $batches = 0;
+        $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.dataset_seconds', 8), 1);
        $rows = (int) ($state['last_upserted_rows'] ?? 0);

        try {
@ -92,7 +95,7 @@ class BusinessMirrorSynchronizer
                    return $this->result($dataset, 'succeeded', $mode, $batches, $rows);
                }

-                if ($maxBatches > 0 && $batches >= $maxBatches) {
+                if ($maxBatches > 0 && ($batches >= $maxBatches || microtime(true) >= $deadline)) {
                    return $this->result($dataset, 'running', $mode, $batches, $rows);
                }
            }
@ -120,8 +123,11 @@ class BusinessMirrorSynchronizer
        if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) {
            $watermark = $this->quote($dataset->watermarkColumn());
            $where[] = $watermark . ' >= ? AND ' . $watermark . ' < ?';
-            $bindings[] = $this->sqlServerDateTime($windowStart);
-            $bindings[] = $this->sqlServerDateTime($windowEnd);
+            $timezoneAware = $this->driver === 'pgsql' && collect($columns)->contains(fn ($c) =>
+                $c['column_name'] === $dataset->watermarkColumn() && $c['data_type'] === 'timestamp with time zone'
+            );
+            $bindings[] = $timezoneAware ? $windowStart?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowStart);
+            $bindings[] = $timezoneAware ? $windowEnd?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowEnd);
        } elseif ($mode === 'incremental' && $cursor === null && $highWatermark !== []) {
            [$predicate, $values] = $this->keyset($dataset->keyColumns(), $highWatermark);
            $where[] = $predicate;
@ -134,11 +140,12 @@ class BusinessMirrorSynchronizer
            array_push($bindings, ...$values);
        }

-        $sql = 'SELECT TOP (' . $limit . ') '
-            . implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns))
-            . ' FROM ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject())
+        $tail = '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));
+        $sql = (new UpstreamSqlDialect($this->driver))->select(
+            implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)), $tail, $limit,
+        );

        return $this->queries->forSource($dataset->sourceKey())->select($sql, $bindings);
    }
@ -149,6 +156,17 @@ class BusinessMirrorSynchronizer
        $name = (string) $column['column_name'];
        $quoted = $this->quote($name);

+        if ($this->driver !== 'sqlsrv') {
+            if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'timestamp with time zone') {
+                // Store timezone-aware PostgreSQL instants as UTC in the shared mirror.
+                return "to_char({$quoted} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}";
+            }
+            if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'bytea') {
+                return "encode({$quoted}, 'hex') AS {$quoted}";
+            }
+            return $quoted;
+        }
+
        // FreeTDS may return SQL Server date values using the connection locale
        // (for example "Jan 1 2026"). Convert temporal columns at the source so
        // every mirror receives deterministic ISO values that MariaDB accepts.
@ -244,7 +262,7 @@ class BusinessMirrorSynchronizer
                $type = $types[strtolower((string) $name)] ?? '';
                if ($type === 'date') {
                    $row[$name] = substr($value, 0, 10);
-                } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime'], true)) {
+                } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone'], true)) {
                    $row[$name] = preg_replace('/(\.\d{6})\d+/', '$1', str_replace('T', ' ', $value));
                } elseif (in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'text', 'ntext', 'xml'], true)) {
                    $row[$name] = $this->normalizeText($value);
@ -271,11 +289,7 @@ class BusinessMirrorSynchronizer

    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 . ']';
+        return (new UpstreamSqlDialect($this->driver))->quote($identifier);
    }

    private function cursorValue(array $cursor, string $column): mixed
--- a/modules/EcommerceFinance/Services/UpstreamConnection.php
+++ b/modules/EcommerceFinance/Services/UpstreamConnection.php
@ -0,0 +1,73 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use PDO;
+use PDOException;
+use SensitiveParameter;
+use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
+
+class UpstreamConnection
+{
+    /** @var array<string,PDO> */
+    private static array $native = [];
+
+    public static function get(string $source): PDO
+    {
+        $registry = app(UpstreamSourceRegistry::class);
+        if ($registry->driver($source) === 'sqlsrv') {
+            return UpstreamOdbcConnection::get($source);
+        }
+        if (! $registry->enabled($source)) {
+            throw new ServiceUnavailableHttpException(30, '该数据源尚未启用。');
+        }
+        return self::$native[$source] ??= self::openNative($registry->driver($source), $registry->value($source, 'connection', []));
+    }
+
+    public static function reset(?string $source = null): void
+    {
+        if ($source === null) self::$native = [];
+        else unset(self::$native[$source]);
+        UpstreamOdbcConnection::reset($source);
+    }
+
+    public static function openNative(string $driver, #[SensitiveParameter] array $config): PDO
+    {
+        if (! in_array($driver, ['pgsql', 'mysql'], true) || ! in_array($driver, PDO::getAvailableDrivers(), true)) {
+            throw new ServiceUnavailableHttpException(30, '应用缺少该数据库类型的驱动，请更新应用镜像。');
+        }
+        $timeout = min(max((int) ($config['connect_timeout'] ?? 5), 1), 30);
+        $seconds = min(max((int) ($config['query_timeout'] ?? 15), 1), 15);
+        $options = [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_PERSISTENT => false, PDO::ATTR_TIMEOUT => $timeout];
+        $host = (string) $config['host'];
+        $database = (string) $config['database'];
+        if (preg_match('/[;\x00-\x1F]/', $host . $database)) {
+            throw new ServiceUnavailableHttpException(30, '主机或数据库名称包含不支持的字符。');
+        }
+        $port = min(max((int) $config['port'], 1), 65535);
+        if ($driver === 'pgsql') {
+            $quote = static fn (string $s): string => "'" . str_replace(['\\', "'"], ['\\\\', "\\'"], $s) . "'";
+            $dsn = 'pgsql:host=' . $quote($host) . ';port=' . $port . ';dbname=' . $quote($database)
+                . ';sslmode=require;connect_timeout=' . $timeout
+                . ";options='-c default_transaction_read_only=on -c statement_timeout=" . ($seconds * 1000) . "'";
+        } else {
+            $dsn = "mysql:host={$host};port={$port};dbname={$database};charset=utf8mb4";
+            $options[PDO::MYSQL_ATTR_SSL_CIPHER] = 'DEFAULT';
+            $options[PDO::MYSQL_ATTR_SSL_VERIFY_SERVER_CERT] = false;
+        }
+        try {
+            $pdo = new PDO($dsn, (string) $config['username'], (string) $config['password'], $options);
+            if ($driver === 'mysql') {
+                $tls = $pdo->query("SHOW SESSION STATUS LIKE 'Ssl_cipher'")->fetch(PDO::FETCH_NUM);
+                if (empty($tls[1])) throw new ServiceUnavailableHttpException(30, '上游数据库未建立 TLS 加密连接。');
+                $pdo->exec('SET SESSION TRANSACTION READ ONLY');
+                $pdo->exec(str_contains($pdo->getAttribute(PDO::ATTR_SERVER_VERSION), 'MariaDB')
+                    ? 'SET SESSION max_statement_time=' . $seconds
+                    : 'SET SESSION MAX_EXECUTION_TIME=' . ($seconds * 1000));
+            }
+            return $pdo;
+        } catch (PDOException) {
+            throw new ServiceUnavailableHttpException(30, '上游只读连接失败，请检查账号、权限、网络和 TLS 配置。');
+        }
+    }
+}
--- a/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php
+++ b/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php
@ -7,62 +7,32 @@ use PDOException;
 use SensitiveParameter;
 use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;

-/** Connection tests only: never used to route business queries or synchronize tables. */
+/** The test button and business readers share one driver implementation. */
 class UpstreamDatabaseProbe
 {
    public function probe(string $driver, #[SensitiveParameter] array $connection, string $label): void
    {
        if ($driver === 'sqlsrv') {
            UpstreamOdbcConnection::probeConfiguration($connection, $label);
-
            return;
        }
-
-        if ($driver !== 'pgsql') {
-            throw new ServiceUnavailableHttpException(30, '不支持此数据库类型，请选择 SQL Server 或 PostgreSQL。');
+        if (! in_array($driver, ['pgsql', 'mysql'], true)) {
+            throw new ServiceUnavailableHttpException(30, '不支持此数据库类型。');
        }
-
-        $timeout = min(max((int) ($connection['connect_timeout'] ?? 5), 1), 30);
-        $queryTimeout = min(max((int) ($connection['query_timeout'] ?? 15), 1), 15) * 1000;
-        $dsn = 'pgsql:host=' . $this->quote((string) $connection['host'])
-            . ';port=' . min(max((int) $connection['port'], 1), 65535)
-            . ';dbname=' . $this->quote((string) $connection['database'])
-            . ';sslmode=require;connect_timeout=' . $timeout
-            . ";options='-c default_transaction_read_only=on -c statement_timeout=" . $queryTimeout . "'";
-
        try {
-            // One short-lived connection, TLS and read-only enforced at session startup.
-            $pdo = $this->connect($dsn, (string) $connection['username'], (string) $connection['password']);
+            $pdo = $this->connect($driver, $connection);
            $statement = $pdo->query('SELECT 1');
            if ($statement === false || (int) $statement->fetchColumn() !== 1) {
-                throw new ServiceUnavailableHttpException(30, 'PostgreSQL 只读测试未返回预期结果。');
+                throw new ServiceUnavailableHttpException(30, '只读测试未返回预期结果。');
            }
            $statement->closeCursor();
        } catch (PDOException) {
-            // Do not expose raw driver errors, DSNs or credentials in logs/session flashes.
-            throw new ServiceUnavailableHttpException(30, 'PostgreSQL 连接失败。请检查账号、密码、数据库访问权限，以及服务器网络和 SSL 配置。');
+            throw new ServiceUnavailableHttpException(30, '上游只读连接失败，请检查账号、权限、网络和 TLS 配置。');
        }
    }

-    protected function connect(string $dsn, string $username, #[SensitiveParameter] string $password): PDO
+    protected function connect(string $driver, #[SensitiveParameter] array $connection): PDO
    {
-        if (! in_array('pgsql', PDO::getAvailableDrivers(), true)) {
-            throw new ServiceUnavailableHttpException(30, '应用缺少 PostgreSQL 驱动，请部署包含 pdo_pgsql 的新版应用镜像。');
-        }
-
-        return new PDO($dsn, $username, $password, [
-            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
-            PDO::ATTR_PERSISTENT => false,
-        ]);
-    }
-
-    private function quote(string $value): string
-    {
-        // PDO replaces all DSN semicolons with spaces; reject rather than alter database names.
-        if (preg_match('/[;\x00-\x1F]/', $value)) {
-            throw new ServiceUnavailableHttpException(30, '主机或数据库名称包含不支持的字符。');
-        }
-
-        return "'" . str_replace(['\\', "'"], ['\\\\', "\\'"], $value) . "'";
+        return UpstreamConnection::openNative($driver, $connection);
    }
 }
--- a/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php
+++ b/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php
@ -27,7 +27,7 @@ class UpstreamReadOnlyQuery
    {
        self::assertReadOnlySql($sql);

-        $statement = UpstreamOdbcConnection::get($this->source)->prepare($sql);
+        $statement = UpstreamConnection::get($this->source)->prepare($sql);
        $statement->execute($bindings);

        return $statement->fetchAll(PDO::FETCH_ASSOC);
@ -45,7 +45,7 @@ class UpstreamReadOnlyQuery
    {
        self::assertReadOnlySql($sql);

-        $statement = UpstreamOdbcConnection::get($this->source)->prepare($sql);
+        $statement = UpstreamConnection::get($this->source)->prepare($sql);
        $statement->execute($bindings);

        return $statement->fetchColumn();
@ -68,6 +68,7 @@ class UpstreamReadOnlyQuery
        $inspect = preg_replace("/'(?:''|[^'])*'/s", "''", $sql) ?? $sql;
        $inspect = preg_replace('/\[(?:\]\]|[^\]])*\]/s', '[]', $inspect) ?? $inspect;
        $inspect = preg_replace('/"(?:""|[^"])*"/s', '""', $inspect) ?? $inspect;
+        $inspect = preg_replace('/`(?:``|[^`])*`/s', '``', $inspect) ?? $inspect;
        $inspect = strtoupper(ltrim($inspect));

        if (preg_match('/^(SELECT|WITH)\b/', $inspect) !== 1) {
@ -76,7 +77,7 @@ class UpstreamReadOnlyQuery

        $forbidden = '/\b(?:INSERT|UPDATE|DELETE|MERGE|DROP|ALTER|CREATE|TRUNCATE|EXEC|EXECUTE|'
            . 'GRANT|REVOKE|DENY|BACKUP|RESTORE|DBCC|BULK|OPENROWSET|OPENDATASOURCE|INTO|'
-            . 'WAITFOR|KILL|SHUTDOWN)\b|\bNEXT\s+VALUE\s+FOR\b/i';
+            . 'WAITFOR|KILL|SHUTDOWN|SET_CONFIG|NEXTVAL|SETVAL|SLEEP|BENCHMARK|GET_LOCK|RELEASE_LOCK|LOAD_FILE|PG_SLEEP)\b|\bNEXT\s+VALUE\s+FOR\b/i';

        if (preg_match($forbidden, $inspect) === 1) {
            throw new InvalidArgumentException('上游查询包含禁止的写入或高风险语法。');
--- a/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php
+++ b/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php
@ -7,7 +7,7 @@ use Illuminate\Support\Facades\Schema;
 use RuntimeException;

 /**
- * 将上游 SQL Server 的表、视图和字段结构同步到共享系统数据库。
+ * 将已支持类型的上游表、视图和字段结构同步到共享系统数据库。
 *
 * 这里只读取 INFORMATION_SCHEMA 元数据，不读取业务行，也不保存连接资料。
 * 获取完整结果成功后才替换单个来源的旧目录，避免网络故障清空现有字典。
@ -28,25 +28,15 @@ class UpstreamSchemaCatalogSynchronizer
            throw new RuntimeException('该上游数据库尚未启用。');
        }

-        $rows = $this->queries->forSource($source)->select(
-            "SELECT
-                c.TABLE_SCHEMA AS schema_name,
-                c.TABLE_NAME AS object_name,
-                CASE WHEN t.TABLE_TYPE = 'VIEW' THEN 'view' ELSE 'table' END AS object_type,
-                c.COLUMN_NAME AS column_name,
-                c.ORDINAL_POSITION AS ordinal_position,
-                c.DATA_TYPE AS data_type,
-                c.CHARACTER_MAXIMUM_LENGTH AS max_length,
-                c.NUMERIC_PRECISION AS numeric_precision,
-                c.NUMERIC_SCALE AS numeric_scale,
-                c.IS_NULLABLE AS is_nullable
-            FROM INFORMATION_SCHEMA.COLUMNS c
-            INNER JOIN INFORMATION_SCHEMA.TABLES t
-                ON t.TABLE_SCHEMA = c.TABLE_SCHEMA
-                AND t.TABLE_NAME = c.TABLE_NAME
-            WHERE c.TABLE_SCHEMA NOT IN ('sys', 'INFORMATION_SCHEMA')
-            ORDER BY c.TABLE_SCHEMA, c.TABLE_NAME, c.ORDINAL_POSITION"
-        );
+        $driver = $this->sources->driver($source);
+        $rows = $this->queries->forSource($source)->select((new UpstreamSqlDialect($driver))->catalogSql());
+        if ($driver === 'mysql') {
+            foreach ($rows as &$row) {
+                // SQL Server timestamp is binary; MySQL timestamp is temporal.
+                if (strtolower($row['data_type'] ?? '') === 'timestamp') $row['data_type'] = 'datetime';
+            }
+            unset($row);
+        }

        if ($rows === []) {
            throw new RuntimeException('上游数据库没有返回可用的表字段目录。');
--- a/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php
+++ b/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php
@ -28,7 +28,11 @@ class UpstreamSourceRegistry
        $definition = config('upstreams.sources.' . $source);

        if (! is_array($definition)) {
-            throw new InvalidArgumentException('未登记的上游数据源：' . $source);
+            $connection = $this->storedConfiguration->find($source);
+            if ($connection === null) {
+                throw new InvalidArgumentException('未登记的上游数据源：' . $source);
+            }
+            $definition = ['label' => $connection->name, 'config' => 'upstream_pipeline.defaults'];
        }

        $configKey = trim((string) ($definition['config'] ?? ''));
@ -77,6 +81,25 @@ class UpstreamSourceRegistry
        return (bool) $this->value($source, 'enabled', false);
    }

+    public function driver(string $source): string
+    {
+        $definition = $this->definition($source);
+        $default = config($definition['config'] . '.connection.driver', 'sqlsrv');
+        return (string) (($this->usesLocalConfiguration()
+            ? $default : $this->storedConfiguration->value($source, 'connection.driver', $default)) ?: 'sqlsrv');
+    }
+
+    /** @return array<string,array> */
+    public function all(): array
+    {
+        $keys = array_unique(array_merge(array_keys((array) config('upstreams.sources', [])), array_keys($this->storedConfiguration->all())));
+        $definitions = [];
+        foreach ($keys as $key) {
+            $definitions[$key] = $this->definition($key);
+        }
+        return $definitions;
+    }
+
    private function usesLocalConfiguration(): bool
    {
        return (bool) config('upstreams.local_configuration.enabled', false);
@ -85,6 +108,7 @@ class UpstreamSourceRegistry
    private function connection(string $source, array $configured): array
    {
        $fields = array_values(array_unique(array_merge(array_keys($configured), [
+            'driver',
            'host',
            'port',
            'database',
--- a/modules/EcommerceFinance/Services/UpstreamSqlDialect.php
+++ b/modules/EcommerceFinance/Services/UpstreamSqlDialect.php
@ -0,0 +1,55 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use InvalidArgumentException;
+
+class UpstreamSqlDialect
+{
+    public function __construct(public readonly string $driver)
+    {
+        if (! in_array($driver, ['sqlsrv', 'pgsql', 'mysql'], true)) {
+            throw new InvalidArgumentException('不支持的上游数据库类型。');
+        }
+    }
+
+    public function quote(string $identifier): string
+    {
+        if ($identifier === '' || str_contains($identifier, "\0")) throw new InvalidArgumentException('无效的数据库标识符。');
+        return match ($this->driver) {
+            'pgsql' => '"' . str_replace('"', '""', $identifier) . '"',
+            'mysql' => '`' . str_replace('`', '``', $identifier) . '`',
+            default => '[' . str_replace(']', ']]', $identifier) . ']',
+        };
+    }
+
+    public function select(string $columns, string $tail, int $limit): string
+    {
+        $limit = min(max($limit, 1), 5000);
+        return $this->driver === 'sqlsrv'
+            ? 'SELECT TOP (' . $limit . ') ' . $columns . ' ' . $tail
+            : 'SELECT ' . $columns . ' ' . $tail . ' LIMIT ' . $limit;
+    }
+
+    public function catalogSql(): string
+    {
+        // Same normalized catalog shape for all engines; MySQL never enumerates other databases.
+        $schema = $this->driver === 'sqlsrv' ? 'INFORMATION_SCHEMA' : 'information_schema';
+        $sql = "SELECT c.table_schema AS schema_name, c.table_name AS object_name,
+            CASE WHEN t.table_type = 'VIEW' THEN 'view' ELSE 'table' END AS object_type,
+            c.column_name AS column_name, c.ordinal_position AS ordinal_position,
+            c.data_type AS data_type, c.character_maximum_length AS max_length,
+            c.numeric_precision AS numeric_precision, c.numeric_scale AS numeric_scale,
+            c.is_nullable AS is_nullable
+            FROM {$schema}." . ($this->driver === 'sqlsrv' ? 'COLUMNS' : 'columns') . " c
+            INNER JOIN {$schema}." . ($this->driver === 'sqlsrv' ? 'TABLES' : 'tables') . " t
+              ON t.table_schema = c.table_schema AND t.table_name = c.table_name
+            WHERE c.table_schema NOT IN ('sys', 'INFORMATION_SCHEMA', 'information_schema', 'pg_catalog', 'mysql', 'performance_schema')";
+        if ($this->driver === 'mysql') $sql .= ' AND c.table_schema = DATABASE()';
+        $sql .= ' ORDER BY c.table_schema, c.table_name, c.ordinal_position';
+        if ($this->driver === 'sqlsrv') {
+            $sql = preg_replace_callback('/\b([ct])\.([a-z_]+)\b/', fn ($m) => $m[1] . '.' . strtoupper($m[2]), $sql);
+        }
+        return $sql;
+    }
+}
--- a/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php
+++ b/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php
@ -42,10 +42,13 @@ class UpstreamStoredConfiguration
            return $this->connections[$source] = null;
        }

-        $this->connections[$source] = DatabaseConnection::query()
-            ->where('company_id', self::SYSTEM_COMPANY_ID)
-            ->where('source_key', $source)
-            ->first();
+        $query = DatabaseConnection::query()->where('company_id', self::SYSTEM_COMPANY_ID);
+        if (preg_match('/^database_([1-9][0-9]*)$/', $source, $matches)) {
+            $query->where('id', $matches[1])->whereNull('source_key');
+        } else {
+            $query->where('source_key', $source);
+        }
+        $this->connections[$source] = $query->first();

        return $this->connections[$source];
    }
@ -55,6 +58,16 @@ class UpstreamStoredConfiguration
        unset($this->connections[$source]);
    }

+    /** @return array<string,DatabaseConnection> */
+    public function all(): array
+    {
+        if (! $this->enabled() || ! Schema::hasTable('system_database_connections')) {
+            return [];
+        }
+        return DatabaseConnection::query()->where('company_id', self::SYSTEM_COMPANY_ID)
+            ->get()->keyBy(fn (DatabaseConnection $row) => $row->sourceIdentifier())->all();
+    }
+
    public function value(string $source, string $path, mixed $default = null): mixed
    {
        $connection = $this->find($source);
@ -68,6 +81,7 @@ class UpstreamStoredConfiguration
            'allowed_scopes' => $connection->allowed_scopes ?? [],
            'connection' => $this->connection($connection, $default),
            'connection.host' => $connection->host,
+            'connection.driver' => $connection->driver,
            'connection.port' => $connection->port,
            'connection.database' => $connection->database_name,
            'connection.username' => $connection->username,
@ -85,6 +99,7 @@ class UpstreamStoredConfiguration
        $configured = is_array($default) ? $default : [];

        return array_replace($configured, [
+            'driver' => $connection->driver,
            'host' => $connection->host,
            'port' => $connection->port,
            'database' => $connection->database_name,
--- a/resources/views/system_settings/databases/form.blade.php
+++ b/resources/views/system_settings/databases/form.blade.php
@ -5,7 +5,8 @@
        <div class="my-6">
            @include('system_settings.partials.navigation')

-            @php($postgres = old('driver', $values['driver']) === 'pgsql')
+            @php($selectedDriver = old('driver', $values['driver']))
+            @php($postgres = $selectedDriver !== 'sqlsrv')
            <form id="cloud-database-form" method="POST" action="{{ $connection ? route('system-settings.databases.update', $connection->id) : route('system-settings.databases.store') }}" class="max-w-4xl rounded-xl border border-gray-200 bg-white p-6">
                @csrf
                @if ($connection) @method('PATCH') @endif
@ -20,11 +21,12 @@
                    <label class="block sm:col-span-2">
                        <span class="text-sm font-semibold text-gray-700">数据库类型 *</span>
                        <select name="driver" required class="mt-1 w-full rounded-lg border border-[#D8D2C7] px-3 py-2 text-sm focus:border-[#856632] focus:ring-[#856632]">
-                            <option value="sqlsrv" @selected(! $postgres)>SQL Server</option>
-                            <option value="pgsql" @selected($postgres)>PostgreSQL</option>
+                            <option value="sqlsrv" @selected($selectedDriver === 'sqlsrv')>SQL Server</option>
+                            <option value="pgsql" @selected($selectedDriver === 'pgsql')>PostgreSQL</option>
+                            <option value="mysql" @selected($selectedDriver === 'mysql')>MySQL / MariaDB</option>
                        </select>
                    </label>
-                    <p data-postgres-help class="text-sm text-gray-600 sm:col-span-2" style="{{ $postgres ? '' : 'display:none' }}" role="status">通过 SSL 加密连接（sslmode=require）；仅执行只读测试，查询最长 15 秒。保存连接不会自动同步业务表。</p>
+                    <p data-postgres-help class="text-sm text-gray-600 sm:col-span-2" style="{{ $postgres ? '' : 'display:none' }}" role="status">通过 TLS 加密连接（PostgreSQL 使用 sslmode=require）；只读查询最长 15 秒。启用后自动发现字段目录，业务表按已批准的数据集同步。</p>
                    <label class="block sm:col-span-2">
                        <span class="text-sm font-semibold text-gray-700">配置名称 *</span>
                        <input name="name" value="{{ old('name', $values['name']) }}" required maxlength="191" placeholder="例如：JianSun 测试库" class="mt-1 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm placeholder-gray-500 focus:border-[#856632] focus:ring-[#856632]">
@ -50,16 +52,6 @@
                        <input type="password" name="password" @required(! $connection) maxlength="1000" autocomplete="new-password" class="mt-1 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-[#856632] focus:ring-[#856632]">
                        <span class="mt-1 block text-xs text-gray-500">{{ $connection ? '留空表示保留原密码。' : '保存后不会回显。' }}</span>
                    </label>
-                    <label data-sqlserver-only class="block sm:col-span-2" style="{{ $postgres ? 'display:none' : '' }}">
-                        <span class="text-sm font-semibold text-gray-700">绑定业务来源（开发配置）</span>
-                        <select name="source_key" class="mt-1 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-[#856632] focus:ring-[#856632]">
-                            <option value="">不绑定，仅保存和测试</option>
-                            @foreach ($sourceLabels as $key => $label)
-                                <option value="{{ $key }}" @selected(old('source_key', $values['source_key']) === $key)>{{ $label }}</option>
-                            @endforeach
-                        </select>
-                        <span class="mt-1 block text-xs text-gray-500">用于关联现有业务查询；连接测试不需要选择。不要将新数据库绑定到其他业务库的来源。</span>
-                    </label>
                </div>

                <details class="mt-6 rounded-lg border border-gray-200 px-4 py-3">
@ -69,7 +61,6 @@
                        <label data-sqlserver-only class="block" style="{{ $postgres ? 'display:none' : '' }}"><span class="text-sm font-semibold text-gray-700">TDS 版本</span><input name="tds_version" value="{{ old('tds_version', $values['tds_version'] ?: '7.4') }}" required @disabled($postgres) maxlength="16" class="mt-1 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-[#856632] focus:ring-[#856632]"></label>
                        <label class="block"><span class="text-sm font-semibold text-gray-700">连接超时（秒）</span><input type="number" name="connect_timeout" value="{{ old('connect_timeout', $values['connect_timeout']) }}" required min="1" max="30" class="mt-1 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-[#856632] focus:ring-[#856632]"></label>
                        <label class="block"><span class="text-sm font-semibold text-gray-700">查询超时（秒）</span><input type="number" name="query_timeout" value="{{ old('query_timeout', $values['query_timeout']) }}" required min="1" max="120" class="mt-1 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm focus:border-[#856632] focus:ring-[#856632]"></label>
-                        <label class="block sm:col-span-2"><span class="text-sm font-semibold text-gray-700">允许的数据范围</span><textarea name="allowed_scopes" rows="2" maxlength="4000" placeholder="绑定业务来源时填写，每行或逗号分隔" class="mt-1 w-full rounded-lg border border-gray-300 px-3 py-2 text-sm placeholder-gray-500 focus:border-[#856632] focus:ring-[#856632]">{{ old('allowed_scopes', $values['allowed_scopes']) }}</textarea></label>
                    </div>
                </details>

@ -90,11 +81,12 @@
                (() => {
                    const form = document.getElementById('cloud-database-form');
                    const driver = form.elements.driver;
+                    const ports = {sqlsrv: '1433', pgsql: '5432', mysql: '3306'};
                    let previous = driver.value;
                    const update = (changePort) => {
-                        const postgres = driver.value === 'pgsql';
-                        if (changePort && form.elements.port.value === (previous === 'pgsql' ? '5432' : '1433')) {
-                            form.elements.port.value = postgres ? '5432' : '1433';
+                        const postgres = driver.value !== 'sqlsrv';
+                        if (changePort && form.elements.port.value === ports[previous]) {
+                            form.elements.port.value = ports[driver.value];
                        }
                        form.querySelectorAll('[data-sqlserver-only]').forEach((group) => {
                            group.style.display = postgres ? 'none' : '';
--- a/resources/views/system_settings/databases/index.blade.php
+++ b/resources/views/system_settings/databases/index.blade.php
@ -88,7 +88,6 @@
                                <th class="px-5 py-3 font-semibold">名称</th>
                                <th class="px-5 py-3 font-semibold">地址</th>
                                <th class="px-5 py-3 font-semibold">数据库</th>
-                                <th class="px-5 py-3 font-semibold">业务绑定</th>
                                <th class="px-5 py-3 font-semibold">状态</th>
                                <th class="px-5 py-3 text-right font-semibold">操作</th>
                            </tr>
@ -96,10 +95,9 @@
                        <tbody class="divide-y divide-gray-100">
                            @forelse ($connections as $connection)
                                <tr>
-                                    <td class="px-5 py-4 font-semibold text-gray-900">{{ $connection->name }}<span class="mt-1 block text-xs font-normal text-gray-600">{{ $connection->driver === 'pgsql' ? 'PostgreSQL · SSL' : 'SQL Server' }}</span></td>
+                                    <td class="px-5 py-4 font-semibold text-gray-900">{{ $connection->name }}<span class="mt-1 block text-xs font-normal text-gray-600">{{ match ($connection->driver) { 'pgsql' => 'PostgreSQL · SSL', 'mysql' => 'MySQL / MariaDB · TLS', default => 'SQL Server' } }}</span></td>
                                    <td class="px-5 py-4 text-gray-600">{{ $connection->host }}:{{ $connection->port }}</td>
                                    <td class="px-5 py-4 text-gray-600">{{ $connection->database_name }}</td>
-                                    <td class="px-5 py-4 text-gray-600">{{ $sourceLabels[$connection->source_key] ?? '未绑定' }}</td>
                                    <td class="px-5 py-4">
                                        <span @class([
                                            'inline-flex rounded-full px-2.5 py-1 text-xs font-semibold',
@ -122,7 +120,7 @@
                                </tr>
                            @empty
                                <tr>
-                                    <td colspan="6" class="px-5 py-12 text-center text-gray-500">
+                                    <td colspan="5" class="px-5 py-12 text-center text-gray-500">
                                        {{ $systemDatabase['mode'] === 'bootstrap'
                                            ? '连接团队共享系统数据库后，两套云数据库配置会自动显示在这里。'
                                            : '共享系统数据库中还没有云数据库配置。' }}
--- a/tests/Feature/Console/BusinessMirrorScheduleTest.php
+++ b/tests/Feature/Console/BusinessMirrorScheduleTest.php
@ -12,6 +12,49 @@ use Tests\TestCase;

 class BusinessMirrorScheduleTest extends TestCase
 {
+    public function test_unapproved_first_sync_is_skipped(): void
+    {
+        config(['upstream_mirrors.refresh_enabled' => true]);
+        $dataset = Mockery::mock(BusinessMirrorDataset::class);
+        $registry = Mockery::mock(BusinessMirrorDatasetRegistry::class);
+        $registry->shouldReceive('all')->once()->andReturn(['pending' => $dataset]);
+        $registry->shouldReceive('scheduled')->with('pending')->andReturn(true);
+        $registry->shouldReceive('automaticInitialSyncAllowed')->with('pending')->andReturn(false);
+        $mirror = Mockery::mock(GenericBusinessMirrorRepository::class);
+        $mirror->shouldReceive('state')->with($dataset)->andReturn(null);
+        $sync = Mockery::mock(BusinessMirrorSynchronizer::class);
+        $sync->shouldNotReceive('sync');
+        $this->app->instance(BusinessMirrorDatasetRegistry::class, $registry);
+        $this->app->instance(GenericBusinessMirrorRepository::class, $mirror);
+        $this->app->instance(BusinessMirrorSynchronizer::class, $sync);
+        $this->artisan('business-mirror:sync-tables')->assertSuccessful();
+    }
+
+    public function test_catchup_and_routine_datasets_both_receive_a_turn(): void
+    {
+        config(['upstream_mirrors.refresh_enabled' => true, 'upstream_pipeline.sync.max_datasets' => 2]);
+        $registry = Mockery::mock(BusinessMirrorDatasetRegistry::class);
+        $mirror = Mockery::mock(GenericBusinessMirrorRepository::class);
+        $objects = [];
+        foreach (['routine-a', 'routine-b', 'catchup-z'] as $key) {
+            $objects[$key] = Mockery::mock(BusinessMirrorDataset::class);
+            $registry->shouldReceive('scheduled')->with($key)->andReturn(true);
+            $mirror->shouldReceive('state')->with($objects[$key])->andReturn([
+                'initial_completed_at' => '2026-08-01 00:00:00', 'last_attempted_at' => '2026-08-01 00:00:00',
+                'status' => $key === 'catchup-z' ? 'running' : 'succeeded',
+            ]);
+        }
+        $registry->shouldReceive('all')->andReturn($objects);
+        $sync = Mockery::mock(BusinessMirrorSynchronizer::class);
+        $result = ['status' => 'succeeded', 'batches' => 1, 'rows' => 1];
+        $sync->shouldReceive('sync')->once()->with('routine-a', false, null, null)->ordered()->andReturn($result);
+        $sync->shouldReceive('sync')->once()->with('catchup-z', false, null, 6)->ordered()->andReturn($result);
+        $this->app->instance(BusinessMirrorDatasetRegistry::class, $registry);
+        $this->app->instance(GenericBusinessMirrorRepository::class, $mirror);
+        $this->app->instance(BusinessMirrorSynchronizer::class, $sync);
+        $this->artisan('business-mirror:sync-tables')->assertSuccessful();
+    }
+
    public function test_generic_rotation_excludes_datasets_with_a_dedicated_scheduler(): void
    {
        config([
@ -28,6 +71,7 @@ class BusinessMirrorScheduleTest extends TestCase
            ProductCatalogMirrorRepository::DATASET_KEY => $catalog,
            'test.dbo.other' => $other,
        ]);
+        $datasets->shouldReceive('scheduled')->once()->with('test.dbo.other')->andReturn(true);
        $mirror = Mockery::mock(GenericBusinessMirrorRepository::class);
        $mirror->shouldReceive('state')->once()->with($other)->andReturn([
            'initial_completed_at' => '2026-08-01 02:00:00',
--- a/tests/Feature/SystemSettings/UpstreamDatabasesTest.php
+++ b/tests/Feature/SystemSettings/UpstreamDatabasesTest.php
@ -578,6 +578,30 @@ class UpstreamDatabasesTest extends FeatureTestCase
        ], $overrides);
    }

+    public function test_mysql_connection_appears_in_dictionary_sources_without_a_binding(): void
+    {
+        $payload = $this->payload(['driver' => 'mysql', 'port' => 3306]);
+        unset($payload['source_key'], $payload['allowed_scopes'], $payload['odbc_driver'], $payload['tds_version']);
+        $this->loginAs()->post(route('system-settings.databases.store'), $payload)->assertSessionHasNoErrors();
+        $row = DatabaseConnection::where('driver', 'mysql')->firstOrFail();
+        $this->assertArrayHasKey($row->sourceIdentifier(), app(UpstreamSourceRegistry::class)->all());
+        $this->get(route('system-settings.databases.create'))->assertOk()
+            ->assertSee('MySQL / MariaDB')->assertDontSee('name="source_key"', false)->assertDontSee('name="allowed_scopes"', false);
+    }
+
+    public function test_editing_simplified_form_preserves_legacy_binding_and_scopes(): void
+    {
+        $row = $this->connection('keep-secret');
+        $row->update(['source_key' => 'jiansun']);
+        $payload = $this->payload(['name' => $row->name, 'password' => '']);
+        unset($payload['source_key'], $payload['allowed_scopes']);
+        $this->loginAs()->patch(route('system-settings.databases.update', $row->id), $payload)->assertSessionHasNoErrors();
+        $row->refresh();
+        $this->assertSame('jiansun', $row->source_key);
+        $this->assertSame(['supplier.options'], $row->allowed_scopes);
+        $this->assertSame('keep-secret', $row->password);
+    }
+
    private function connection(string $password): DatabaseConnection
    {
        return DatabaseConnection::create([
--- a/tests/Unit/EcommerceFinance/GenericUpstreamPipelineTest.php
+++ b/tests/Unit/EcommerceFinance/GenericUpstreamPipelineTest.php
@ -0,0 +1,184 @@
+<?php
+
+namespace Tests\Unit\EcommerceFinance;
+
+use App\Models\Settings\DatabaseConnection;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Modules\EcommerceFinance\Services\BusinessMirrorDataReader;
+use Modules\EcommerceFinance\Services\UpstreamConnection;
+use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
+use Modules\EcommerceFinance\Services\UpstreamSchemaCatalogSynchronizer;
+use Modules\EcommerceFinance\Services\UpstreamSqlDialect;
+use PDO;
+use PDOException;
+use Tests\TestCase;
+
+class GenericUpstreamPipelineTest extends TestCase
+{
+    protected function setUp(): void
+    {
+        parent::setUp();
+        config([
+            'upstreams.stored_configuration.enabled' => true,
+            'upstreams.local_configuration.enabled' => false,
+            'database.connections.sqlite.password' => 'fixture-cipher-key',
+            'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
+            'upstream_mirrors.connection' => 'business_mirror',
+            'upstream_mirrors.datasets' => [],
+            'upstream_mirrors.refresh_enabled' => true,
+        ]);
+        DB::purge('business_mirror');
+        Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $t): void {
+            $t->string('dataset_key')->primary();
+            foreach (['source_key', 'source_object', 'target_table', 'status', 'run_mode', 'generation', 'last_error_code'] as $c) $t->string($c)->nullable();
+            foreach (['cursor', 'high_watermark'] as $c) $t->json($c)->nullable();
+            foreach (['window_started_at', 'window_ends_at', 'initial_completed_at', 'last_full_succeeded_at', 'last_attempted_at', 'last_succeeded_at'] as $c) $t->dateTime($c)->nullable();
+            $t->unsignedBigInteger('last_upserted_rows')->default(0);
+            $t->unsignedBigInteger('mirror_row_count')->default(0);
+            $t->timestamps();
+        });
+    }
+
+    protected function tearDown(): void
+    {
+        UpstreamConnection::reset();
+        parent::tearDown();
+    }
+
+    private function connection(string $driver = 'pgsql'): DatabaseConnection
+    {
+        return DatabaseConnection::create([
+            'company_id' => 0, 'name' => 'Isolated fixture', 'driver' => $driver, 'host' => 'fixture.invalid',
+            'port' => $driver === 'pgsql' ? 5432 : 3306, 'database_name' => 'fixture',
+            'username' => 'fixture', 'password' => 'fixture-password', 'enabled' => true,
+            'connect_timeout' => 5, 'query_timeout' => 15,
+        ]);
+    }
+
+    public function test_new_connection_is_immediately_a_source_without_changing_config(): void
+    {
+        $row = $this->connection();
+        $key = $row->sourceIdentifier();
+        $sources = app(UpstreamSourceRegistry::class);
+        $this->assertNull($row->source_key);
+        $this->assertNull(config('upstreams.sources.' . $key));
+        $this->assertSame('pgsql', $sources->driver($key));
+        $this->assertTrue($sources->enabled($key));
+        $this->assertSame('fixture-password', $sources->value($key, 'connection.password'));
+        $this->assertArrayHasKey($key, $sources->all());
+        $this->assertSame('jiansun', $sources->definition('jiansun')['key']);
+    }
+
+    public function test_dialects_use_engine_specific_limits_and_quotes(): void
+    {
+        $this->assertSame('SELECT TOP (5) [id] FROM [dbo].[records]', (new UpstreamSqlDialect('sqlsrv'))->select('[id]', 'FROM [dbo].[records]', 5));
+        $this->assertSame('SELECT "id" FROM "public"."records" LIMIT 5', (new UpstreamSqlDialect('pgsql'))->select('"id"', 'FROM "public"."records"', 5));
+        $this->assertSame('SELECT `id` FROM `records` LIMIT 5', (new UpstreamSqlDialect('mysql'))->select('`id`', 'FROM `records`', 5));
+        $this->assertStringContainsString('c.table_schema = DATABASE()', (new UpstreamSqlDialect('mysql'))->catalogSql());
+    }
+
+    public function test_source_discovery_does_not_decrypt_credentials(): void
+    {
+        $row = $this->connection();
+        DB::table('system_database_connections')->where('id', $row->id)->update(['shared_password' => 'invalid-fixture-envelope']);
+        $sources = app(UpstreamSourceRegistry::class);
+        $this->assertArrayHasKey($row->sourceIdentifier(), $sources->all());
+        $this->assertSame('pgsql', $sources->driver($row->sourceIdentifier()));
+    }
+
+    public function test_declarations_require_fields_and_explicit_initial_sync_approval(): void
+    {
+        $row = $this->connection();
+        $source = $row->sourceIdentifier();
+        DB::table('upstream_schema_columns')->insert([
+            'source_key' => $source, 'schema_name' => 'public', 'object_name' => 'records',
+            'object_type' => 'table', 'column_name' => 'id', 'ordinal_position' => 1, 'data_type' => 'bigint',
+            'is_nullable' => false, 'synchronized_at' => now(), 'created_at' => now(), 'updated_at' => now(),
+        ]);
+        $args = ['source' => $source, 'schema' => 'public', 'object' => 'records', '--keys' => 'id', '--columns' => 'id'];
+        $this->artisan('business-mirror:register', $args)->assertSuccessful();
+        $key = $source . '.public.records';
+        $registry = app(BusinessMirrorDatasetRegistry::class);
+        $this->assertFalse($registry->automaticInitialSyncAllowed($key));
+        $this->artisan('business-mirror:sync-tables')->assertSuccessful();
+        $this->assertSame(0, DB::connection('business_mirror')->table('business_mirror_table_states')->count());
+        $this->artisan('business-mirror:register', $args + ['--enable-sync' => true])->assertSuccessful();
+        $this->assertTrue($registry->automaticInitialSyncAllowed($key));
+        $this->assertSame(['id'], $registry->columnSelection($key));
+        $this->artisan('business-mirror:register', ['source' => $source, 'schema' => 'public', 'object' => 'records', '--disable-sync' => true])->assertSuccessful();
+        $this->assertFalse($registry->scheduled($key));
+    }
+
+    /** @dataProvider nativeDrivers */
+    public function test_native_tls_catalog_first_sync_pagination_and_incremental(string $driver, bool $timezoneAware = false, ?string $fixtureHost = null): void
+    {
+        if (getenv('CAIWU_PIPELINE_INTEGRATION') !== '1') $this->markTestSkipped('Requires isolated pipeline fixture containers.');
+        $host = $fixtureHost ?? ($driver === 'pgsql' ? 'caiwu-pipeline-pg' : 'caiwu-pipeline-mysql');
+        $database = $driver === 'pgsql' ? 'postgres' : 'pipeline_fixture';
+        $user = $driver === 'pgsql' ? 'postgres' : 'root';
+        $schema = $driver === 'pgsql' ? 'public' : $database;
+        $dsn = $driver === 'pgsql' ? "pgsql:host={$host};dbname={$database};sslmode=require" : "mysql:host={$host};dbname={$database}";
+        $options = [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION];
+        if ($driver === 'mysql') {
+            $options[PDO::MYSQL_ATTR_SSL_CIPHER] = 'DEFAULT';
+            $options[PDO::MYSQL_ATTR_SSL_VERIFY_SERVER_CERT] = false;
+        }
+        $admin = new PDO($dsn, $user, 'fixture-password', $options);
+        $admin->exec('DROP TABLE IF EXISTS pipeline_records');
+        $type = $driver === 'pgsql' ? ($timezoneAware ? 'timestamp with time zone' : 'timestamp without time zone') : 'datetime(6)';
+        $admin->exec('CREATE TABLE pipeline_records (id bigint primary key, modified_at ' . $type . ', amount decimal(18,2), label varchar(50), private_note text)');
+        $date = now()->subMinute()->format($timezoneAware ? 'Y-m-d H:i:sP' : 'Y-m-d H:i:s');
+        $insert = $admin->prepare('INSERT INTO pipeline_records (id, modified_at, amount, label) VALUES (?, ?, ?, ?)');
+        $insert->execute([1, $date, '12.30', 'first']);
+        $insert->execute([2, $date, '45.60', 'second']);
+        $row = $this->connection($driver);
+        $row->update(['host' => $host, 'database_name' => $database, 'username' => $user]);
+        $source = $row->sourceIdentifier();
+        $result = app(UpstreamSchemaCatalogSynchronizer::class)->sync($source);
+        $this->assertGreaterThanOrEqual(5, $result['columns']);
+        $this->artisan('business-mirror:register', [
+            'source' => $source, 'schema' => $schema, 'object' => 'pipeline_records', '--keys' => 'id',
+            '--columns' => 'id,modified_at,amount,label', '--watermark' => 'modified_at', '--enable-sync' => true,
+        ])->assertSuccessful();
+        $this->artisan('business-mirror:sync-tables')->assertSuccessful();
+        $dataset = app(BusinessMirrorDatasetRegistry::class)->get($source . '.' . $schema . '.pipeline_records');
+        $mirror = DB::connection('business_mirror')->table($dataset->targetTable());
+        $this->assertSame(2, $mirror->count());
+        $this->assertSame(2, (int) $mirror->orderBy('id')->offset(1)->value('id'));
+        $this->assertFalse(Schema::connection('business_mirror')->hasColumn($dataset->targetTable(), 'private_note'));
+        $admin->exec("UPDATE pipeline_records SET amount=99.50 WHERE id=1");
+        $insert->execute([3, $date, '6.70', 'third']);
+        $insert->execute([4, now()->addHour()->format($timezoneAware ? 'Y-m-d H:i:sP' : 'Y-m-d H:i:s'), '1.00', 'future']);
+        $this->artisan('business-mirror:sync-tables')->assertSuccessful();
+        $this->assertSame(3, DB::connection('business_mirror')->table($dataset->targetTable())->count());
+        $this->assertEquals(99.5, DB::connection('business_mirror')->table($dataset->targetTable())->where('id', 1)->value('amount'));
+        $pdo = UpstreamConnection::get($source);
+        $row->update(['enabled' => false]);
+        $page = app(BusinessMirrorDataReader::class)->paginate($dataset->key(), [], 2, 2);
+        $this->assertSame(3, $page['total']);
+        $this->assertSame(3, (int) $page['rows'][0]['id']);
+        $this->assertArrayNotHasKey('_mirror_key', $page['rows'][0]);
+        $filtered = app(BusinessMirrorDataReader::class)->paginate($dataset->key(), ['label' => 'first']);
+        $this->assertSame(1, $filtered['total']);
+        if ($driver === 'pgsql') {
+            $this->assertSame('on', $pdo->query('SHOW default_transaction_read_only')->fetchColumn());
+            $this->assertTrue((bool) $pdo->query('SELECT ssl FROM pg_stat_ssl WHERE pid=pg_backend_pid()')->fetchColumn());
+        } else {
+            $this->assertNotEmpty($pdo->query("SHOW SESSION STATUS LIKE 'Ssl_cipher'")->fetch(PDO::FETCH_NUM)[1]);
+        }
+        try {
+            $pdo->exec("INSERT INTO pipeline_records (id) VALUES (99)");
+            $this->fail('The native upstream connection must reject writes.');
+        } catch (PDOException $e) {
+            $this->assertStringContainsString('read', strtolower($e->getMessage()));
+        }
+    }
+
+    public static function nativeDrivers(): array
+    {
+        return [['pgsql'], ['mysql'], ['pgsql', true], ['mysql', false, 'caiwu-pipeline-mariadb']];
+    }
+}
--- a/tests/Unit/EcommerceFinance/UpstreamDatabaseProbeTest.php
+++ b/tests/Unit/EcommerceFinance/UpstreamDatabaseProbeTest.php
@ -21,7 +21,7 @@ class UpstreamDatabaseProbeTest extends TestCase
            'username' => 'reader', 'password' => 'fixture-secret', 'connect_timeout' => 5, 'query_timeout' => 120];
    }

-    public function test_postgres_uses_tls_read_only_timeout_and_one_select(): void
+    public function test_postgres_probe_uses_the_shared_connector_and_one_select(): void
    {
        $statement = Mockery::mock(PDOStatement::class);
        $statement->shouldReceive('fetchColumn')->once()->andReturn(1);
@ -29,16 +29,7 @@ class UpstreamDatabaseProbeTest extends TestCase
        $pdo = Mockery::mock(PDO::class);
        $pdo->shouldReceive('query')->once()->with('SELECT 1')->andReturn($statement);
        $probe = Mockery::mock(UpstreamDatabaseProbe::class)->makePartial()->shouldAllowMockingProtectedMethods();
-        $probe->shouldReceive('connect')->once()->withArgs(function ($dsn, $user, $password) {
-            $this->assertStringContainsString('sslmode=require', $dsn);
-            $this->assertStringContainsString('default_transaction_read_only=on', $dsn);
-            $this->assertStringContainsString('statement_timeout=15000', $dsn);
-            $this->assertStringContainsString("dbname='test\\' database'", $dsn);
-            $this->assertStringContainsString('connect_timeout=5', $dsn);
-            $this->assertStringNotContainsString('fixture-secret', $dsn);
-
-            return $user === 'reader' && $password === 'fixture-secret';
-        })->andReturn($pdo);
+        $probe->shouldReceive('connect')->once()->with('pgsql', $this->connection())->andReturn($pdo);
        $probe->probe('pgsql', $this->connection(), '测试库');
    }

@ -50,7 +41,7 @@ class UpstreamDatabaseProbeTest extends TestCase
            $probe->probe('pgsql', $this->connection(), '测试库');
            $this->fail('Expected sanitized connection error');
        } catch (ServiceUnavailableHttpException $exception) {
-            $this->assertStringContainsString('PostgreSQL 连接失败', $exception->getMessage());
+            $this->assertStringContainsString('上游只读连接失败', $exception->getMessage());
            $this->assertStringNotContainsString('fixture-secret', $exception->getMessage());
            $this->assertNull($exception->getPrevious());
        }
--- a/tests/Unit/Utilities/SystemDatabasePersistenceTest.php
+++ b/tests/Unit/Utilities/SystemDatabasePersistenceTest.php
@ -9,6 +9,42 @@ use Tests\TestCase;

 class SystemDatabasePersistenceTest extends TestCase
 {
+    public function test_switch_replaces_all_legacy_connection_routes_and_duplicate_keys(): void
+    {
+        $environmentPath = storage_path('framework/testing-legacy-routes.env');
+        $original = "DB_HOST=mysql\nDB_HOST_READ=old-read\nDB_HOST_WRITE=old-write\n"
+            . "DATABASE_URL=mysql://fixture:fixture@old-url/old_database\nDB_SOCKET=/tmp/old.sock\n"
+            . "DB_HOST=old-duplicate\nSYSTEM_DATABASE_MODE=bootstrap\nUNRELATED_OPTION=keep\n";
+        file_put_contents($environmentPath, $original);
+        $configuration = new class($environmentPath) extends SystemDatabaseConfiguration {
+            public array $verified = [];
+            protected function runFreshVerification(string $command): void
+            {
+                $this->verified = $this->freshVerificationEnvironment();
+            }
+            protected function clearConfigurationCache(): void {}
+        };
+        try {
+            $backup = $configuration->persist([
+                'host' => 'shared.example.test', 'port' => 8443, 'database' => 'shared_system',
+                'username' => 'fixture_user', 'password' => 'fixture_password', 'prefix' => '',
+                'tls_required' => true, 'tls_verify_server_cert' => false,
+            ]);
+            $saved = \Dotenv\Dotenv::parse(file_get_contents($environmentPath));
+            foreach (['DB_HOST', 'DB_HOST_READ', 'DB_HOST_WRITE'] as $key) {
+                $this->assertSame('shared.example.test', $saved[$key]);
+                $this->assertSame('shared.example.test', $configuration->verified[$key]);
+            }
+            $this->assertSame('', $saved['DATABASE_URL']);
+            $this->assertSame('', $saved['DB_SOCKET']);
+            $this->assertSame('keep', $saved['UNRELATED_OPTION']);
+            $this->assertSame($original, file_get_contents($backup));
+        } finally {
+            @unlink($environmentPath);
+            if (isset($backup)) @unlink($backup);
+        }
+    }
+
    public function test_failed_fresh_process_verification_restores_original_environment(): void
    {
        $environmentPath = storage_path('framework/testing-system-database.env');