### ZYT /Users/long/Work/zyt through eec320440
diff --git a/docs/ej-pharmacy-incremental-medicine-sync.md b/docs/ej-pharmacy-incremental-medicine-sync.md
new file mode 100644
index 000000000..89fbabafa
--- /dev/null
+++ b/docs/ej-pharmacy-incremental-medicine-sync.md
@@ -0,0 +1,49 @@
+# EJ 药材非破坏性增量同步
+
+`ej-pharmacy:push-medicines` 将 ZYT 中启用且未删除的药材，通过现有 HMAC OpenAPI 增量推送到 EJ。
+
+## 不变式
+
+- EJ 只执行 `POST /api/openapi/v1/medicine-imports` 的新增/幂等确认，不删除或清空 EJ 药材。
+- EJ 中由其他来源或人工录入的药材保持不变。
+- ZYT 已同步且在 EJ 中仍正常启用的药材保持名称、价格、库存和编码不变。
+- ZYT 已同步但在 EJ 中被软删除的药材恢复原记录和原 `medicine_code`；被停用的药材重新启用。
+- ZYT 后续新增、且从未同步过的药材追加到 EJ，并初始化零库存。
+- ZYT 只增量写入或更新 EJ 返回的目录投影，不清空目录和映射。
+- ZYT 中已停用或软删除的人工映射不重新启用。
+- 命令默认 dry-run；只有同时提供 `--apply` 和确认令牌才执行远端导入。
+
+## 先预检
+
+```bash
+cd /www/wwwroot/zyt/server
+php think ej-pharmacy:push-medicines
+```
+
+输出示例：
+
+```text
+dry-run source=654 candidates=654 batches=7 mapped=654 unmapped=0 preserved_inactive=0 remote_delete=0 local_delete=0
+```
+
+## 执行增量同步
+
+```bash
+cd /www/wwwroot/zyt/server
+php think ej-pharmacy:push-medicines \
+  --apply \
+  --confirm=INCREMENTAL_NO_DELETE \
+  --batch-size=100
+```
+
+可用 `--run-id=<ID>` 固定本次批次的幂等标识；ID 最长 25 位。若省略，命令使用稳定默认值 `zyt-incremental`，相同内容重试时复用同一 import ID。若旧批次已经完成、但需要修复 EJ 中后来被单独移除的药材，应提供新的维修 run ID，例如 `--run-id=repair-20260910`。
+
+执行结果会分别报告 EJ 新增、EJ 已存在、本地目录新增/更新、本地映射新增/更新/不变，以及两侧删除数量；删除数量固定为零。
+
+## 方向说明
+
+- `ej-pharmacy:push-medicines`：ZYT → EJ，非破坏性增量新增或恢复远端缺项。
+- `ej-pharmacy:sync-catalog`：EJ → ZYT，拉取 EJ 目录变化。
+- `ej-pharmacy:bootstrap-medicines`：一次性初始化并替换本地投影，不用于已有业务数据的生产环境增量同步。
+
+同步范围以 `source_system=zyt` 和 `source_medicine_id` 标识来源。恢复和重新启用只作用于原来由 ZYT 同步过去的药材，因此不会修改 EJ 自己新增的药材。
diff --git a/server/app/command/EjPharmacyPushMedicines.php b/server/app/command/EjPharmacyPushMedicines.php
new file mode 100644
index 000000000..cb97a8ff2
--- /dev/null
+++ b/server/app/command/EjPharmacyPushMedicines.php
@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\command;
+
+use app\common\service\pharmacy\EjMedicineIncrementalPushService;
+use think\console\Command;
+use think\console\Input;
+use think\console\input\Option;
+use think\console\Output;
+
+final class EjPharmacyPushMedicines extends Command
+{
+    protected function configure()
+    {
+        $this->setName('ej-pharmacy:push-medicines')
+            ->setDescription('非破坏性增量推送 ZYT 药材到恩济药房；保留双方已有药材')
+            ->addOption('apply', null, Option::VALUE_NONE, '执行远端增量导入；缺省仅做 dry-run')
+            ->addOption('confirm', null, Option::VALUE_OPTIONAL, '增量写入确认令牌：INCREMENTAL_NO_DELETE', '')
+            ->addOption('batch-size', null, Option::VALUE_OPTIONAL, '每批药材数量（1-500）', 100)
+            ->addOption('run-id', null, Option::VALUE_OPTIONAL, '幂等运行标识（缺省 zyt-incremental，最长 25 位）', '');
+    }
+
+    protected function execute(Input $input, Output $output)
+    {
+        try {
+            $apply = (bool) $input->getOption('apply');
+            $confirm = (string) $input->getOption('confirm');
+            $batchSize = (int) $input->getOption('batch-size');
+            EjMedicineIncrementalPushService::assertCommandGate($apply, $confirm, $batchSize);
+
+            if (!$apply) {
+                $plan = EjMedicineIncrementalPushService::plan($batchSize);
+                $output->writeln(sprintf(
+                    'dry-run source=%d candidates=%d batches=%d mapped=%d unmapped=%d '
+                    . 'preserved_inactive=%d remote_delete=0 local_delete=0',
+                    $plan['source_count'],
+                    $plan['candidate_count'],
+                    $plan['batch_count'],
+                    $plan['mapped_count'],
+                    $plan['unmapped_count'],
+                    $plan['preserved_inactive']
+                ));
+                return 0;
+            }
+
+            $result = EjMedicineIncrementalPushService::execute(
+                $batchSize,
+                (string) $input->getOption('run-id')
+            );
+            $output->writeln(sprintf(
+                'incremental 完成 run_id=%s source=%d candidates=%d batches=%d '
+                . 'remote_created=%d remote_existing=%d preserved_inactive=%d '
+                . 'catalog_created=%d catalog_updated=%d mapping_created=%d mapping_updated=%d '
+                . 'mapping_unchanged=%d remote_delete=0 local_delete=0',
+                $result['run_id'],
+                $result['source_count'],
+                $result['candidate_count'],
+                $result['batch_count'],
+                $result['remote_created'],
+                $result['remote_existing'],
+                $result['preserved_inactive'],
+                $result['catalog_created'],
+                $result['catalog_updated'],
+                $result['mapping_created'],
+                $result['mapping_updated'],
+                $result['mapping_unchanged']
+            ));
+            return 0;
+        } catch (\Throwable $exception) {
+            $output->error($exception->getMessage());
+            return 1;
+        }
+    }
+}
diff --git a/server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php b/server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php
new file mode 100644
index 000000000..836722d81
--- /dev/null
+++ b/server/app/common/service/pharmacy/EjMedicineIncrementalPushService.php
@@ -0,0 +1,590 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\common\service\pharmacy;
+
+use DomainException;
+use InvalidArgumentException;
+use RuntimeException;
+use think\facade\Db;
+
+final class EjMedicineIncrementalPushService
+{
+    private const SOURCE_SYSTEM = 'zyt';
+    private const APPLY_CONFIRMATION = 'INCREMENTAL_NO_DELETE';
+    private const DEFAULT_RUN_ID = 'zyt-incremental';
+    private const STATE_ID = 1;
+    private const LOCK_TTL = 120;
+
+    public static function assertCommandGate(bool $apply, string $confirm, int $batchSize): void
+    {
+        self::assertBatchSize($batchSize);
+        if ($apply && !hash_equals(self::APPLY_CONFIRMATION, $confirm)) {
+            throw new InvalidArgumentException(
+                '执行增量写入必须提供 --confirm=' . self::APPLY_CONFIRMATION
+            );
+        }
+    }
+
+    /**
+     * @param null|callable():array<int,array<string,mixed>> $sourceLoader
+     * @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
+     * @return array{source_count:int,candidate_count:int,batch_count:int,mapped_count:int,unmapped_count:int,preserved_inactive:int,remote_delete_count:int,local_delete_count:int}
+     */
+    public static function plan(
+        int $batchSize = 100,
+        ?callable $sourceLoader = null,
+        ?callable $mappingLoader = null
+    ): array {
+        self::assertBatchSize($batchSize);
+        $sourceRows = $sourceLoader === null ? self::loadSourceRows() : $sourceLoader();
+        $items = EjMedicineBootstrapItem::fromRows($sourceRows);
+        $localIds = array_map('intval', array_column($items, 'source_medicine_id'));
+        $mappingRows = $mappingLoader === null
+            ? self::loadMappingRows($localIds)
+            : $mappingLoader($localIds);
+        $selection = self::selectCandidates($items, $mappingRows);
+
+        return [
+            'source_count' => count($items),
+            'candidate_count' => count($selection['candidates']),
+            'batch_count' => (int) ceil(count($selection['candidates']) / $batchSize),
+            'mapped_count' => $selection['mapped_count'],
+            'unmapped_count' => $selection['unmapped_count'],
+            'preserved_inactive' => $selection['preserved_inactive'],
+            'remote_delete_count' => 0,
+            'local_delete_count' => 0,
+        ];
+    }
+
+    /**
+     * Incrementally imports every active local medicine through EJ's idempotent
+     * source identity and upserts only the returned ZYT projection rows.
+     * Existing EJ-only medicines and unrelated local projections are untouched.
+     *
+     * @param null|callable():array<int,array<string,mixed>> $sourceLoader
+     * @param null|callable(array<string,mixed>):array<string,mixed> $importer
+     * @param null|callable(array<int,array<string,mixed>>):array<string,int> $projectionUpserter
+     * @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
+     * @param null|callable(string):bool $lockAcquirer
+     * @param null|callable(string):bool $lockRenewer
+     * @param null|callable(string):void $lockReleaser
+     * @return array<string,int|string>
+     */
+    public static function execute(
+        int $batchSize = 100,
+        string $runId = '',
+        ?callable $sourceLoader = null,
+        ?callable $importer = null,
+        ?callable $projectionUpserter = null,
+        ?callable $mappingLoader = null,
+        ?callable $lockAcquirer = null,
+        ?callable $lockRenewer = null,
+        ?callable $lockReleaser = null
+    ): array {
+        self::assertBatchSize($batchSize);
+
+        $customLockCallbacks = count(array_filter(
+            [$lockAcquirer, $lockRenewer, $lockReleaser],
+            static fn (?callable $callback): bool => $callback !== null
+        ));
+        if ($customLockCallbacks !== 0 && $customLockCallbacks !== 3) {
+            throw new InvalidArgumentException('增量同步锁回调必须同时提供 acquire、renew 和 release');
+        }
+        $lockAcquirer ??= static fn (string $token): bool => self::acquireLock($token);
+        $lockRenewer ??= static fn (string $token): bool => self::renewLock($token);
+        $lockReleaser ??= static function (string $token): void {
+            self::releaseLock($token);
+        };
+
+        $lockToken = bin2hex(random_bytes(16));
+        if (!$lockAcquirer($lockToken)) {
+            throw new DomainException('恩济药材同步正在执行，请稍后重试');
+        }
+
+        try {
+            return self::executeLocked(
+                $batchSize,
+                $runId,
+                $sourceLoader,
+                $importer,
+                $projectionUpserter,
+                $mappingLoader,
+                $lockRenewer,
+                $lockToken
+            );
+        } finally {
+            $lockReleaser($lockToken);
+        }
+    }
+
+    /**
+     * @param null|callable():array<int,array<string,mixed>> $sourceLoader
+     * @param null|callable(array<string,mixed>):array<string,mixed> $importer
+     * @param null|callable(array<int,array<string,mixed>>):array<string,int> $projectionUpserter
+     * @param null|callable(array<int,int>):array<int,array<string,mixed>> $mappingLoader
+     * @param callable(string):bool $lockRenewer
+     * @return array<string,int|string>
+     */
+    private static function executeLocked(
+        int $batchSize,
+        string $runId,
+        ?callable $sourceLoader,
+        ?callable $importer,
+        ?callable $projectionUpserter,
+        ?callable $mappingLoader,
+        callable $lockRenewer,
+        string $lockToken
+    ): array {
+        $sourceRows = $sourceLoader === null ? self::loadSourceRows() : $sourceLoader();
+        $allItems = EjMedicineBootstrapItem::fromRows($sourceRows);
+        $localIds = array_map('intval', array_column($allItems, 'source_medicine_id'));
+        $mappingRows = $mappingLoader === null
+            ? self::loadMappingRows($localIds)
+            : $mappingLoader($localIds);
+        $selection = self::selectCandidates($allItems, $mappingRows);
+        $items = $selection['candidates'];
+        $runId = self::normalizeRunId($runId);
+        $batches = self::buildBatches($items, $batchSize, $runId);
+
+        if ($items !== [] && $importer === null) {
+            if (!EjPharmacyClient::isConfigured()) {
+                throw new RuntimeException('恩济药房接口未启用或配置不完整');
+            }
+            $client = new EjPharmacyClient();
+            $importer = static fn (array $payload): array => $client->importMedicines($payload);
+        }
+
+        $sourceById = [];
+        foreach ($items as $item) {
+            $sourceById[(string) $item['source_medicine_id']] = $item;
+        }
+        $seenCodes = [];
+        $seenVersions = [];
+        $projectionRows = [];
+        $remoteCreated = 0;
+        $remoteExisting = 0;
+        foreach ($batches as $payload) {
+            self::assertLockLease($lockRenewer, $lockToken);
+            $response = $importer($payload);
+            self::assertLockLease($lockRenewer, $lockToken);
+            $responseItems = self::validateImportResponse(
+                $response,
+                $payload,
+                $seenCodes,
+                $seenVersions
+            );
+            foreach ($responseItems as $responseItem) {
+                $sourceId = (string) $responseItem['source_medicine_id'];
+                $source = $sourceById[$sourceId] ?? null;
+                if (!is_array($source)) {
+                    throw new RuntimeException("恩济增量导入返回未知 source_medicine_id：{$sourceId}");
+                }
+                $action = (string) $responseItem['action'];
+                $medicineCode = (string) $responseItem['medicine_code'];
+                $expectedCode = $selection['active_mapping_codes'][$sourceId] ?? null;
+                if ($expectedCode !== null && !hash_equals($expectedCode, $medicineCode)) {
+                    throw new DomainException(
+                        "本地药材 {$sourceId} 的启用映射编码 {$expectedCode} 与恩济返回 {$medicineCode} 不一致"
+                    );
+                }
+                $remoteCreated += $action === 'created' ? 1 : 0;
+                $remoteExisting += $action === 'existing' ? 1 : 0;
+                $projectionRows[] = [
+                    'local_medicine_id' => $sourceId,
+                    'medicine_code' => $medicineCode,
+                    'name' => (string) $source['name'],
+                    'brand' => (string) ($source['brand'] ?? ''),
+                    'unit' => (string) $source['unit'],
+                    'settlement_price' => (string) $source['settlement_price'],
+                    'retail_price' => (string) $source['retail_price'],
+                    'status' => (int) $source['status'],
+                    'catalog_version' => (int) $responseItem['catalog_version'],
+                ];
+            }
+        }
+
+        $projectionStats = [
+            'catalog_created' => 0,
+            'catalog_updated' => 0,
+            'mapping_created' => 0,
+            'mapping_updated' => 0,
+            'mapping_unchanged' => 0,
+        ];
+        if ($projectionRows !== []) {
+            self::assertLockLease($lockRenewer, $lockToken);
+            $projectionStats = $projectionUpserter === null
+                ? self::upsertProjection($projectionRows, $lockToken)
+                : $projectionUpserter($projectionRows);
+        }
+
+        return array_merge([
+            'run_id' => $runId,
+            'source_count' => count($allItems),
+            'candidate_count' => count($items),
+            'batch_count' => count($batches),
+            'preserved_inactive' => $selection['preserved_inactive'],
+            'remote_created' => $remoteCreated,
+            'remote_existing' => $remoteExisting,
+            'remote_delete_count' => 0,
+            'local_delete_count' => 0,
+        ], $projectionStats);
+    }
+
+    /**
+     * @param list<array<string,mixed>> $items
+     * @return list<array{source_system:string,import_id:string,items:array<int,array<string,mixed>>}>
+     */
+    public static function buildBatches(array $items, int $batchSize, string $runId): array
+    {
+        self::assertBatchSize($batchSize);
+        $runId = self::normalizeRunId($runId);
+        usort($items, static fn (array $left, array $right): int => EjMedicineBootstrapItem::compareIds(
+            (string) ($left['source_medicine_id'] ?? ''),
+            (string) ($right['source_medicine_id'] ?? '')
+        ));
+
+        $batches = [];
+        foreach (array_chunk($items, $batchSize) as $index => $batchItems) {
+            $contentJson = json_encode(
+                $batchItems,
+                JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES
+            );
+            $batches[] = [
+                'source_system' => self::SOURCE_SYSTEM,
+                'import_id' => sprintf(
+                    '%s-%04d-%s',
+                    $runId,
+                    $index + 1,
+                    substr(hash('sha256', $contentJson), 0, 32)
+                ),
+                'items' => $batchItems,
+            ];
+        }
+
+        return $batches;
+    }
+
+    private static function assertBatchSize(int $batchSize): void
+    {
+        if ($batchSize < 1 || $batchSize > 500) {
+            throw new InvalidArgumentException('--batch-size 必须在 1 到 500 之间');
+        }
+    }
+
+    private static function normalizeRunId(string $runId): string
+    {
+        $runId = trim($runId);
+        if ($runId === '') {
+            $runId = self::DEFAULT_RUN_ID;
+        }
+        if (strlen($runId) > 25 || preg_match('/^[A-Za-z0-9][A-Za-z0-9._-]*$/D', $runId) !== 1) {
+            throw new InvalidArgumentException('--run-id 必须为不超过 25 位的字母、数字、点、下划线或短横线');
+        }
+
+        return $runId;
+    }
+
+    /** @return array<int,array<string,mixed>> */
+    private static function loadSourceRows(): array
+    {
+        return Db::name('doctor_medicine')
+            ->field('id,name,unit,settlement_price,retail_price,status')
+            ->where('status', 1)
+            ->whereNull('delete_time')
+            ->order('id', 'asc')
+            ->select()
+            ->toArray();
+    }
+
+    /** @param array<int,int> $localIds @return array<int,array<string,mixed>> */
+    private static function loadMappingRows(array $localIds): array
+    {
+        if ($localIds === []) {
+            return [];
+        }
+
+        return Db::name('ej_medicine_mapping')
+            ->whereIn('local_medicine_id', $localIds)
+            ->field('id,local_medicine_id,medicine_code,status,delete_time')
+            ->order('local_medicine_id', 'asc')
+            ->select()
+            ->toArray();
+    }
+
+    /**
+     * Active mappings are replayed so EJ can recover a missing remote medicine.
+     * A medicine with an inactive or soft-deleted mapping is intentionally
+     * excluded; an operator decision must never be undone by synchronization.
+     *
+     * @param list<array<string,mixed>> $items
+     * @param array<int,array<string,mixed>> $mappingRows
+     * @return array{candidates:list<array<string,mixed>>,mapped_count:int,unmapped_count:int,preserved_inactive:int,active_mapping_codes:array<string,string>}
+     */
+    private static function selectCandidates(array $items, array $mappingRows): array
+    {
+        $byLocalId = [];
+        foreach ($mappingRows as $mapping) {
+            $localId = (int) ($mapping['local_medicine_id'] ?? 0);
+            if ($localId < 1 || isset($byLocalId[$localId])) {
+                throw new DomainException("本地药材 {$localId} 存在重复的恩济映射记录");
+            }
+            $byLocalId[$localId] = $mapping;
+        }
+
+        $candidates = [];
+        $mappedCount = 0;
+        $unmappedCount = 0;
+        $preservedInactive = 0;
+        $activeMappingCodes = [];
+        foreach ($items as $item) {
+            $localId = (int) $item['source_medicine_id'];
+            $mapping = $byLocalId[$localId] ?? null;
+            if ($mapping === null) {
+                ++$unmappedCount;
+                $candidates[] = $item;
+                continue;
+            }
+            if ((int) ($mapping['status'] ?? 0) === 1 && ($mapping['delete_time'] ?? null) === null) {
+                $medicineCode = trim((string) ($mapping['medicine_code'] ?? ''));
+                if ($medicineCode === '') {
+                    throw new DomainException("本地药材 {$localId} 的启用恩济映射编码为空");
+                }
+                ++$mappedCount;
+                $candidates[] = $item;
+                $activeMappingCodes[(string) $localId] = $medicineCode;
+                continue;
+            }
+            ++$preservedInactive;
+        }
+
+        return [
+            'candidates' => $candidates,
+            'mapped_count' => $mappedCount,
+            'unmapped_count' => $unmappedCount,
+            'preserved_inactive' => $preservedInactive,
+            'active_mapping_codes' => $activeMappingCodes,
+        ];
+    }
+
+    /**
+     * @param array<string,mixed> $response
+     * @param array<string,mixed> $payload
+     * @param array<string,bool> $seenCodes
+     * @param array<int,bool> $seenVersions
+     * @return list<array{source_medicine_id:string,medicine_code:string,catalog_version:int,action:string}>
+     */
+    private static function validateImportResponse(
+        array $response,
+        array $payload,
+        array &$seenCodes,
+        array &$seenVersions
+    ): array {
+        $nextSeenCodes = $seenCodes;
+        $nextSeenVersions = $seenVersions;
+        $items = EjMedicineBootstrapService::validateImportResponse(
+            $response,
+            $payload,
+            $nextSeenCodes,
+            $nextSeenVersions
+        );
+
+        $data = $response['body']['data'] ?? null;
+        if (!is_array($data) || !hash_equals(self::SOURCE_SYSTEM, (string) ($data['source_system'] ?? ''))) {
+            throw new RuntimeException('恩济药材导入响应 source_system 不匹配');
+        }
+        $expectedPayloadHash = hash('sha256', json_encode(
+            self::canonicalize($payload),
+            JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES
+        ));
+        $actualPayloadHash = strtolower(trim((string) ($data['payload_hash'] ?? '')));
+        if (
+            preg_match('/^[a-f0-9]{64}$/D', $actualPayloadHash) !== 1
+            || !hash_equals($expectedPayloadHash, $actualPayloadHash)
+        ) {
+            throw new RuntimeException('恩济药材导入响应 payload_hash 不匹配');
+        }
+
+        $seenCodes = $nextSeenCodes;
+        $seenVersions = $nextSeenVersions;
+        return $items;
+    }
+
+    private static function canonicalize(mixed $value): mixed
+    {
+        if (!is_array($value)) {
+            return $value;
+        }
+        if (array_is_list($value)) {
+            return array_map([self::class, 'canonicalize'], $value);
+        }
+        ksort($value, SORT_STRING);
+        foreach ($value as $key => $child) {
+            $value[$key] = self::canonicalize($child);
+        }
+        return $value;
+    }
+
+    /** @param callable(string):bool $lockRenewer */
+    private static function assertLockLease(callable $lockRenewer, string $lockToken): void
+    {
+        if (!$lockRenewer($lockToken)) {
+            throw new DomainException('恩济药材同步锁已失效，请重试');
+        }
+    }
+
+    private static function acquireLock(string $token): bool
+    {
+        $now = time();
+        return Db::name('ej_pharmacy_sync_state')
+            ->where('id', self::STATE_ID)
+            ->where(function ($query) use ($now): void {
+                $query->where('lock_token', '')->whereOr('lock_expires_at', '<', $now);
+            })
+            ->update([
+                'lock_token' => $token,
+                'lock_expires_at' => $now + self::LOCK_TTL,
+                'update_time' => $now,
+            ]) === 1;
+    }
+
+    private static function renewLock(string $token): bool
+    {
+        $now = time();
+        $query = Db::name('ej_pharmacy_sync_state')
+            ->where('id', self::STATE_ID)
+            ->where('lock_token', $token)
+            ->where('lock_expires_at', '>=', $now);
+        $updated = $query->update([
+            'lock_expires_at' => $now + self::LOCK_TTL,
+            'update_time' => $now,
+        ]);
+        if ($updated === 1) {
+            return true;
+        }
+
+        $state = Db::name('ej_pharmacy_sync_state')
+            ->where('id', self::STATE_ID)
+            ->where('lock_token', $token)
+            ->find();
+        return is_array($state) && (int) ($state['lock_expires_at'] ?? 0) >= $now;
+    }
+
+    private static function releaseLock(string $token): void
+    {
+        Db::name('ej_pharmacy_sync_state')
+            ->where('id', self::STATE_ID)
+            ->where('lock_token', $token)
+            ->update([
+                'lock_token' => '',
+                'lock_expires_at' => 0,
+                'update_time' => time(),
+            ]);
+    }
+
+    /**
+     * @param array<int,array<string,mixed>> $rows
+     * @return array{catalog_created:int,catalog_updated:int,mapping_created:int,mapping_updated:int,mapping_unchanged:int}
+     */
+    private static function upsertProjection(array $rows, string $lockToken): array
+    {
+        return Db::transaction(static function () use ($rows, $lockToken): array {
+            $stats = [
+                'catalog_created' => 0,
+                'catalog_updated' => 0,
+                'mapping_created' => 0,
+                'mapping_updated' => 0,
+                'mapping_unchanged' => 0,
+            ];
+            $state = Db::name('ej_pharmacy_sync_state')
+                ->where('id', self::STATE_ID)
+                ->lock(true)
+                ->find();
+            if (
+                !is_array($state)
+                || !hash_equals($lockToken, (string) ($state['lock_token'] ?? ''))
+                || (int) ($state['lock_expires_at'] ?? 0) < time()
+            ) {
+                throw new DomainException('恩济药材同步锁已失效，请重试');
+            }
+            $now = time();
+            foreach ($rows as $row) {
+                $localId = (int) ($row['local_medicine_id'] ?? 0);
+                $medicineCode = trim((string) ($row['medicine_code'] ?? ''));
+                if ($localId < 1 || $medicineCode === '') {
+                    throw new RuntimeException('恩济增量导入投影缺少本地药材 ID 或 medicine_code');
+                }
+
+                $conflict = Db::name('ej_medicine_mapping')
+                    ->where('medicine_code', $medicineCode)
+                    ->where('local_medicine_id', '<>', $localId)
+                    ->lock(true)
+                    ->find();
+                if ($conflict) {
+                    throw new DomainException(
+                        "恩济药材编码 {$medicineCode} 已映射到本地药材 " . (int) $conflict['local_medicine_id']
+                    );
+                }
+
+                $catalogValues = [
+                    'name' => (string) $row['name'],
+                    'brand' => (string) ($row['brand'] ?? ''),
+                    'unit' => (string) $row['unit'],
+                    'settlement_price' => (string) $row['settlement_price'],
+                    'retail_price' => (string) $row['retail_price'],
+                    'status' => (int) $row['status'],
+                    'catalog_version' => (int) $row['catalog_version'],
+                    'remote_deleted' => 0,
+                    'update_time' => $now,
+                ];
+                $catalog = Db::name('ej_medicine_catalog')
+                    ->where('medicine_code', $medicineCode)
+                    ->lock(true)
+                    ->find();
+                if ($catalog) {
+                    Db::name('ej_medicine_catalog')->where('id', (int) $catalog['id'])->update($catalogValues);
+                    ++$stats['catalog_updated'];
+                } else {
+                    Db::name('ej_medicine_catalog')->insert($catalogValues + [
+                        'medicine_code' => $medicineCode,
+                        'create_time' => $now,
+                    ]);
+                    ++$stats['catalog_created'];
+                }
+
+                $mapping = Db::name('ej_medicine_mapping')
+                    ->where('local_medicine_id', $localId)
+                    ->lock(true)
+                    ->find();
+                $mappingValues = [
+                    'medicine_code' => $medicineCode,
+                    'status' => 1,
+                    'operator_id' => 0,
+                    'operator_name' => 'system-incremental',
+                    'delete_time' => null,
+                    'update_time' => $now,
+                ];
+                if (!$mapping) {
+                    Db::name('ej_medicine_mapping')->insert($mappingValues + [
+                        'local_medicine_id' => $localId,
+                        'create_time' => $now,
+                    ]);
+                    ++$stats['mapping_created'];
+                } elseif ((int) $mapping['status'] !== 1 || ($mapping['delete_time'] ?? null) !== null) {
+                    throw new DomainException("本地药材 {$localId} 的恩济映射已被停用，增量同步保持该状态不变");
+                } elseif (hash_equals((string) $mapping['medicine_code'], $medicineCode)) {
+                    ++$stats['mapping_unchanged'];
+                } else {
+                    throw new DomainException(
+                        "本地药材 {$localId} 的启用映射编码 "
+                        . (string) $mapping['medicine_code']
+                        . " 与恩济返回 {$medicineCode} 不一致，增量同步未改写该映射"
+                    );
+                }
+            }
+
+            return $stats;
+        });
+    }
+}
diff --git a/server/config/console.php b/server/config/console.php
index 6de151f83..dd547f59f 100755
--- a/server/config/console.php
+++ b/server/config/console.php
@@ -41,6 +41,7 @@ return [
         'gancao:sync-logistics' => 'app\\command\\GancaoSyncLogisticsRoute',
         'ej-pharmacy:sync-catalog' => 'app\\command\\EjPharmacySyncCatalog',
         'ej-pharmacy:bootstrap-medicines' => 'app\\command\\EjPharmacyBootstrapMedicines',
+        'ej-pharmacy:push-medicines' => 'app\\command\\EjPharmacyPushMedicines',
         // 历史 internal_cost：批量甘草预报价回填（CTM_PREVIEW）
         'tcm:backfill-internal-cost' => 'app\\command\\TcmBackfillPrescriptionOrderInternalCost',
         // 批量回填业务订单签收时间到物流库（导出读库即可，不再逐单 HTTP）
diff --git a/server/tests/pharmacy/incremental_medicine_push.php b/server/tests/pharmacy/incremental_medicine_push.php
new file mode 100644
index 000000000..baf1acbe0
--- /dev/null
+++ b/server/tests/pharmacy/incremental_medicine_push.php
@@ -0,0 +1,342 @@
+<?php
+
+declare(strict_types=1);
+
+require dirname(__DIR__, 2) . '/vendor/autoload.php';
+
+use app\common\service\pharmacy\EjMedicineIncrementalPushService;
+
+$passed = 0;
+$assertSame = static function (mixed $expected, mixed $actual, string $message) use (&$passed): void {
+    if ($expected !== $actual) {
+        throw new RuntimeException($message . '\nExpected: ' . var_export($expected, true) . '\nActual: ' . var_export($actual, true));
+    }
+    ++$passed;
+};
+$assertTrue = static function (bool $actual, string $message) use (&$passed): void {
+    if (!$actual) {
+        throw new RuntimeException($message);
+    }
+    ++$passed;
+};
+$canonicalize = null;
+$canonicalize = static function (mixed $value) use (&$canonicalize): mixed {
+    if (!is_array($value)) {
+        return $value;
+    }
+    if (array_is_list($value)) {
+        return array_map($canonicalize, $value);
+    }
+    ksort($value, SORT_STRING);
+    foreach ($value as $key => $child) {
+        $value[$key] = $canonicalize($child);
+    }
+    return $value;
+};
+$canonicalPayloadHash = static fn (array $payload): string => hash(
+    'sha256',
+    json_encode($canonicalize($payload), JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES)
+);
+
+$sourceRows = [
+    [
+        'id' => 101,
+        'name' => '增量药材甲',
+        'unit' => '克',
+        'settlement_price' => '1.100000',
+        'retail_price' => '1.200000',
+        'status' => 1,
+    ],
+    [
+        'id' => 202,
+        'name' => '增量药材乙',
+        'unit' => '克',
+        'settlement_price' => '2.100000',
+        'retail_price' => '2.200000',
+        'status' => 1,
+    ],
+    [
+        'id' => 303,
+        'name' => '人工停用映射药材',
+        'unit' => '克',
+        'settlement_price' => '3.100000',
+        'retail_price' => '3.200000',
+        'status' => 1,
+    ],
+];
+
+EjMedicineIncrementalPushService::assertCommandGate(false, '', 100);
+$blocked = false;
+try {
+    EjMedicineIncrementalPushService::assertCommandGate(true, '', 100);
+} catch (InvalidArgumentException $exception) {
+    $blocked = str_contains($exception->getMessage(), 'INCREMENTAL_NO_DELETE');
+}
+$assertTrue($blocked, 'apply mode must require the non-destructive confirmation token');
+foreach ([0, 501] as $invalidBatchSize) {
+    try {
+        EjMedicineIncrementalPushService::assertCommandGate(false, '', $invalidBatchSize);
+        throw new RuntimeException('invalid incremental batch size unexpectedly accepted');
+    } catch (InvalidArgumentException $exception) {
+        $assertTrue(str_contains($exception->getMessage(), 'batch-size'), 'invalid batch size must return an actionable error');
+    }
+}
+
+$plan = EjMedicineIncrementalPushService::plan(
+    100,
+    static fn (): array => $sourceRows,
+    static fn (array $localIds): array => [
+        ['id' => 1, 'local_medicine_id' => 101, 'medicine_code' => 'EJ-101', 'status' => 1, 'delete_time' => null],
+        ['id' => 2, 'local_medicine_id' => 303, 'medicine_code' => 'EJ-303', 'status' => 0, 'delete_time' => 123],
+    ]
+);
+$assertSame(3, $plan['source_count'], 'dry-run must count every active local medicine');
+$assertSame(2, $plan['candidate_count'], 'dry-run must replay active mappings and add never-mapped medicines');
+$assertSame(1, $plan['batch_count'], 'dry-run must report the remote batch count');
+$assertSame(1, $plan['mapped_count'], 'dry-run must count existing active mappings');
+$assertSame(1, $plan['unmapped_count'], 'dry-run must identify local medicines without an active mapping');
+$assertSame(1, $plan['preserved_inactive'], 'dry-run must preserve manually disabled or unlinked mappings');
+$assertSame(0, $plan['remote_delete_count'], 'incremental push must never plan remote deletes');
+$assertSame(0, $plan['local_delete_count'], 'incremental push must never plan local projection deletes');
+
+$capturedPayloads = [];
+$capturedProjection = [];
+$heldLockToken = '';
+$lockAcquireCount = 0;
+$lockRenewCount = 0;
+$lockReleaseCount = 0;
+$result = EjMedicineIncrementalPushService::execute(
+    1,
+    'zyt-inc-contract',
+    static fn (): array => $sourceRows,
+    static function (array $payload) use (&$capturedPayloads, $canonicalPayloadHash): array {
+        $capturedPayloads[] = $payload;
+        $sourceId = (string) $payload['items'][0]['source_medicine_id'];
+        $action = $sourceId === '101' ? 'existing' : 'created';
+        return [
+            'http_status' => $action === 'created' ? 201 : 200,
+            'body' => [
+                'code' => 0,
+                'message' => 'ok',
+                'data' => [
+                    'source_system' => 'zyt',
+                    'import_id' => $payload['import_id'],
+                    'payload_hash' => $canonicalPayloadHash($payload),
+                    'idempotent' => false,
+                    'item_count' => 1,
+                    'created_count' => $action === 'created' ? 1 : 0,
+                    'existing_count' => $action === 'existing' ? 1 : 0,
+                    'items' => [[
+                        'source_medicine_id' => $sourceId,
+                        'medicine_code' => 'EJ-' . $sourceId,
+                        'catalog_version' => $sourceId === '101' ? 10 : 11,
+                        'action' => $action,
+                    ]],
+                ],
+            ],
+            'request_id' => 'request-' . $sourceId,
+        ];
+    },
+    static function (array $rows) use (&$capturedProjection): array {
+        $capturedProjection = $rows;
+        return [
+            'catalog_created' => 1,
+            'catalog_updated' => 1,
+            'mapping_created' => 1,
+            'mapping_updated' => 0,
+            'mapping_unchanged' => 1,
+        ];
+    },
+    static fn (array $localIds): array => [
+        ['id' => 1, 'local_medicine_id' => 101, 'medicine_code' => 'EJ-101', 'status' => 1, 'delete_time' => null],
+        ['id' => 2, 'local_medicine_id' => 303, 'medicine_code' => 'EJ-303', 'status' => 0, 'delete_time' => 123],
+    ],
+    static function (string $token) use (&$heldLockToken, &$lockAcquireCount): bool {
+        ++$lockAcquireCount;
+        if ($heldLockToken !== '') {
+            return false;
+        }
+        $heldLockToken = $token;
+        return true;
+    },
+    static function (string $token) use (&$heldLockToken, &$lockRenewCount): bool {
+        ++$lockRenewCount;
+        return $heldLockToken !== '' && hash_equals($heldLockToken, $token);
+    },
+    static function (string $token) use (&$heldLockToken, &$lockReleaseCount): void {
+        ++$lockReleaseCount;
+        if ($heldLockToken !== '' && hash_equals($heldLockToken, $token)) {
+            $heldLockToken = '';
+        }
+    }
+);
+
+$assertSame(2, count($capturedPayloads), 'apply must send deterministic bounded batches');
+$assertTrue(
+    str_starts_with((string) $capturedPayloads[0]['import_id'], 'zyt-inc-contract-0001-'),
+    'incremental import id must include the caller-provided run id and batch ordinal'
+);
+$assertSame(['101', '202'], array_column($capturedProjection, 'local_medicine_id'), 'projection upsert must retain source identity');
+$assertSame(['EJ-101', 'EJ-202'], array_column($capturedProjection, 'medicine_code'), 'projection upsert must use EJ-returned codes');
+$assertSame(1, $result['remote_created'], 'result must report remotely created medicines');
+$assertSame(1, $result['remote_existing'], 'result must report idempotently existing medicines');
+$assertSame(1, $result['preserved_inactive'], 'apply must not reactivate an operator-disabled mapping');
+$assertSame(0, $result['remote_delete_count'], 'apply result must prove no EJ medicine was deleted');
+$assertSame(0, $result['local_delete_count'], 'apply result must prove no ZYT projection was deleted');
+$assertSame(1, $lockAcquireCount, 'apply must acquire one shared EJ synchronization lease');
+$assertSame(5, $lockRenewCount, 'apply must renew its lease around every remote batch and before projection writes');
+$assertSame(1, $lockReleaseCount, 'apply must release its synchronization lease');
+$assertSame('', $heldLockToken, 'successful apply must not leave the synchronization lease held');
+
+$candidateItems = array_merge($capturedPayloads[0]['items'], $capturedPayloads[1]['items']);
+$defaultBatches = EjMedicineIncrementalPushService::buildBatches($candidateItems, 1, '');
+$retryBatches = EjMedicineIncrementalPushService::buildBatches(array_reverse($candidateItems), 1, '');
+$assertSame(
+    array_column($defaultBatches, 'import_id'),
+    array_column($retryBatches, 'import_id'),
+    'default incremental import ids must be deterministic for identical content'
+);
+$assertTrue(
+    str_starts_with((string) $defaultBatches[0]['import_id'], 'zyt-incremental-0001-'),
+    'default run id must be stable so an interrupted apply can safely retry'
+);
+
+$responseFor = static function (
+    array $payload,
+    string $medicineCode,
+    string $sourceSystem = 'zyt',
+    ?string $responsePayloadHash = null
+) use ($canonicalPayloadHash): array {
+    $sourceId = (string) $payload['items'][0]['source_medicine_id'];
+    return [
+        'http_status' => 200,
+        'body' => ['code' => 0, 'message' => 'ok', 'data' => [
+            'source_system' => $sourceSystem,
+            'import_id' => $payload['import_id'],
+            'payload_hash' => $responsePayloadHash ?? $canonicalPayloadHash($payload),
+            'idempotent' => true,
+            'item_count' => 1,
+            'created_count' => 0,
+            'existing_count' => 1,
+            'items' => [[
+                'source_medicine_id' => $sourceId,
+                'medicine_code' => $medicineCode,
+                'catalog_version' => 99,
+                'action' => 'existing',
+            ]],
+        ]],
+        'request_id' => 'validation-' . $sourceId,
+    ];
+};
+$alwaysAcquire = static fn (string $token): bool => $token !== '';
+$alwaysRenew = static fn (string $token): bool => $token !== '';
+$releaseNoop = static function (string $token): void {};
+
+foreach ([
+    'source_system' => static fn (array $payload): array => $responseFor($payload, 'EJ-202', 'other'),
+    'payload_hash' => static fn (array $payload): array => $responseFor($payload, 'EJ-202', 'zyt', str_repeat('0', 64)),
+] as $field => $invalidImporter) {
+    $projectionCalled = false;
+    try {
+        EjMedicineIncrementalPushService::execute(
+            1,
+            'response-validation',
+            static fn (): array => [$sourceRows[1]],
+            $invalidImporter,
+            static function () use (&$projectionCalled): array {
+                $projectionCalled = true;
+                return [];
+            },
+            static fn (array $localIds): array => [],
+            $alwaysAcquire,
+            $alwaysRenew,
+            $releaseNoop
+        );
+        throw new RuntimeException("invalid {$field} unexpectedly accepted");
+    } catch (RuntimeException $exception) {
+        $assertTrue(str_contains($exception->getMessage(), $field), "incremental response must reject invalid {$field}");
+    }
+    $assertSame(false, $projectionCalled, "invalid {$field} must block local projection writes");
+}
+
+$projectionCalled = false;
+try {
+    EjMedicineIncrementalPushService::execute(
+        1,
+        'mapping-mismatch',
+        static fn (): array => [$sourceRows[0]],
+        static fn (array $payload): array => $responseFor($payload, 'EJ-DIFFERENT'),
+        static function () use (&$projectionCalled): array {
+            $projectionCalled = true;
+            return [];
+        },
+        static fn (array $localIds): array => [
+            ['id' => 1, 'local_medicine_id' => 101, 'medicine_code' => 'EJ-101', 'status' => 1, 'delete_time' => null],
+        ],
+        $alwaysAcquire,
+        $alwaysRenew,
+        $releaseNoop
+    );
+    throw new RuntimeException('active mapping code mismatch unexpectedly accepted');
+} catch (DomainException $exception) {
+    $assertTrue(str_contains($exception->getMessage(), '不一致'), 'active mapping replay must reject a different EJ code');
+}
+$assertSame(false, $projectionCalled, 'active mapping mismatch must not overwrite its local projection');
+
+$sourceLoaded = false;
+$failedAcquireReleased = false;
+try {
+    EjMedicineIncrementalPushService::execute(
+        1,
+        'concurrent-run',
+        static function () use (&$sourceLoaded): array {
+            $sourceLoaded = true;
+            return [];
+        },
+        null,
+        null,
+        null,
+        static fn (string $token): bool => false,
+        $alwaysRenew,
+        static function (string $token) use (&$failedAcquireReleased): void { $failedAcquireReleased = true; }
+    );
+    throw new RuntimeException('concurrent incremental run unexpectedly acquired the lease');
+} catch (DomainException $exception) {
+    $assertTrue(str_contains($exception->getMessage(), '正在执行'), 'concurrent apply must fail with a business error');
+}
+$assertSame(false, $sourceLoaded, 'concurrent apply must stop before reading or pushing source medicines');
+$assertSame(false, $failedAcquireReleased, 'a failed lock acquisition must not release another run lease');
+
+$leaseRenewals = 0;
+$leaseFailureReleased = false;
+$projectionCalled = false;
+try {
+    EjMedicineIncrementalPushService::execute(
+        1,
+        'lease-loss',
+        static fn (): array => [$sourceRows[1]],
+        static fn (array $payload): array => $responseFor($payload, 'EJ-202'),
+        static function () use (&$projectionCalled): array {
+            $projectionCalled = true;
+            return [];
+        },
+        static fn (array $localIds): array => [],
+        $alwaysAcquire,
+        static function () use (&$leaseRenewals): bool {
+            ++$leaseRenewals;
+            return $leaseRenewals === 1;
+        },
+        static function (string $token) use (&$leaseFailureReleased): void { $leaseFailureReleased = true; }
+    );
+    throw new RuntimeException('expired synchronization lease unexpectedly allowed projection');
+} catch (DomainException $exception) {
+    $assertTrue(str_contains($exception->getMessage(), '锁已失效'), 'lease loss must return an actionable retry error');
+}
+$assertSame(false, $projectionCalled, 'lease loss after a remote response must block local projection writes');
+$assertSame(true, $leaseFailureReleased, 'lease loss must still release the caller-owned lock token');
+
+$serviceSource = (string) file_get_contents(dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineIncrementalPushService.php');
+$assertTrue(!str_contains($serviceSource, '->delete('), 'incremental service must not call model delete');
+$assertTrue(!str_contains($serviceSource, '->delete()'), 'incremental service must not call model delete without arguments');
+
+echo "incremental EJ medicine push tests passed: {$passed}\n";
diff --git a/server/tests/pharmacy/route_contracts.php b/server/tests/pharmacy/route_contracts.php
index 2b8aaa14c..613322a96 100644
--- a/server/tests/pharmacy/route_contracts.php
+++ b/server/tests/pharmacy/route_contracts.php
@@ -39,9 +39,13 @@ $assertTrue(
 $bootstrapItemPath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineBootstrapItem.php';
 $bootstrapServicePath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineBootstrapService.php';
 $bootstrapCommandPath = dirname(__DIR__, 2) . '/app/command/EjPharmacyBootstrapMedicines.php';
+$incrementalServicePath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjMedicineIncrementalPushService.php';
+$incrementalCommandPath = dirname(__DIR__, 2) . '/app/command/EjPharmacyPushMedicines.php';
 $assertTrue(is_file($bootstrapItemPath), 'ZYT must provide a dedicated EJ medicine bootstrap item normalizer');
 $assertTrue(is_file($bootstrapServicePath), 'ZYT must provide an atomic EJ medicine bootstrap service');
 $assertTrue(is_file($bootstrapCommandPath), 'ZYT must provide the ej-pharmacy:bootstrap-medicines command');
+$assertTrue(is_file($incrementalServicePath), 'ZYT must provide a non-destructive EJ medicine incremental push service');
+$assertTrue(is_file($incrementalCommandPath), 'ZYT must provide the ej-pharmacy:push-medicines command');

 $clientSource = (string) file_get_contents(dirname(__DIR__, 2) . '/app/common/service/pharmacy/EjPharmacyClient.php');
 $configSource = (string) file_get_contents(dirname(__DIR__, 2) . '/config/ej_pharmacy.php');
@@ -62,6 +66,34 @@ $assertTrue(
         && str_contains((string) file_get_contents($bootstrapCommandPath), 'RESET_TEST_CATALOG'),
     'bootstrap command must be registered and guard destructive replacement with the exact confirmation token'
 );
+$incrementalServiceSource = (string) file_get_contents($incrementalServicePath);
+$incrementalCommandSource = (string) file_get_contents($incrementalCommandPath);
+$assertTrue(
+    str_contains($consoleSource, "'ej-pharmacy:push-medicines'")
+        && str_contains($incrementalCommandSource, 'INCREMENTAL_NO_DELETE')
+        && str_contains($incrementalCommandSource, "addOption('apply'")
+        && str_contains($incrementalCommandSource, 'dry-run'),
+    'incremental push command must be registered, dry-run by default, and require the non-delete confirmation token'
+);
+$assertTrue(
+    !str_contains($incrementalServiceSource, '->delete(')
+        && str_contains($incrementalServiceSource, "'remote_delete_count' => 0")
+        && str_contains($incrementalServiceSource, "'local_delete_count' => 0"),
+    'incremental medicine push must preserve EJ-only medicines and unrelated local projections'
+);
+$assertTrue(
+    str_contains($incrementalServiceSource, "private const DEFAULT_RUN_ID = 'zyt-incremental'")
+        && str_contains($incrementalServiceSource, 'private static function acquireLock(')
+        && str_contains($incrementalServiceSource, "->where('lock_token', '')")
+        && str_contains($incrementalServiceSource, "'lock_expires_at'"),
+    'incremental medicine apply must use deterministic retry ids and share the EJ synchronization lease'
+);
+$assertTrue(
+    str_contains($incrementalServiceSource, "['source_system']")
+        && str_contains($incrementalServiceSource, "['payload_hash']")
+        && str_contains($incrementalServiceSource, '增量同步未改写该映射'),
+    'incremental medicine apply must validate response identity and never rewrite an active mapping to a different code'
+);
 $assertTrue(
     str_contains($configSource, "'catalog_sync_enabled'")
         && str_contains($configSource, "env('EJ_PHARMACY_CATALOG_SYNC_ENABLED', false)"),
### EJ /Users/long/Work/ej through 495b02341ac940d7f7b3dac5254fb6714b41311b
diff --git a/server/app/common/service/pharmacy/MedicineImportService.php b/server/app/common/service/pharmacy/MedicineImportService.php
index ab0fbc3..0942365 100644
--- a/server/app/common/service/pharmacy/MedicineImportService.php
+++ b/server/app/common/service/pharmacy/MedicineImportService.php
@@ -7,6 +7,7 @@ namespace app\common\service\pharmacy;
 use app\common\model\pharmacy\Medicine;
 use app\common\model\pharmacy\MedicineImportBatch;
 use app\common\model\pharmacy\MedicineSource;
+use app\common\model\pharmacy\Sequence;
 use app\common\model\pharmacy\Stock;
 use app\common\model\pharmacy\Warehouse;
 use DomainException;
@@ -62,11 +63,9 @@ final class MedicineImportService
             if (!hash_equals((string) $batch->payload_hash, $input['payload_hash'])) {
                 throw new DomainException('Medicine import identity conflicts with a different payload');
             }
-            return self::batchReplay($input, $sourceRows, $medicineRows);
+            return self::batchReplay($input, $sourceRows, $medicineRows, $sequence);
         }

-        self::assertCatalogBelongsToBootstrap($medicineRows, $sourceRows);
-
         $createdCount = 0;
         $existingCount = 0;
         $results = [];
@@ -74,17 +73,12 @@ final class MedicineImportService
             $sourceId = $item['source_medicine_id'];
             $itemHash = MedicineImportPayload::itemHash($item);
             if (isset($sourceRows[$sourceId])) {
-                $results[] = self::existingResult($sourceRows[$sourceId], $itemHash, $medicineRows);
+                $results[] = self::existingResult($sourceRows[$sourceId], $item, $itemHash, $medicineRows, $sequence);
                 ++$existingCount;
                 continue;
             }

-            $catalogVersion = CatalogVersionAllocator::next(
-                static fn (): int => (int) $sequence->current_value,
-                static function (int $next) use ($sequence): void {
-                    $sequence->save(['current_value' => $next]);
-                }
-            );
+            $catalogVersion = self::nextCatalogVersion($sequence);
             $medicine = Medicine::create([
                 'medicine_code' => 'EJ' . strtoupper(bin2hex(random_bytes(6))),
                 'name' => $item['name'],
@@ -149,28 +143,18 @@ final class MedicineImportService
         return self::response($input, $results, $createdCount, $existingCount, false);
     }

-    /** @param array<int,array<string,mixed>> $medicineRows @param array<string,array<string,mixed>> $sourceRows */
-    private static function assertCatalogBelongsToBootstrap(array $medicineRows, array $sourceRows): void
-    {
-        $medicineIds = array_map('intval', array_keys($medicineRows));
-        $mappedMedicineIds = array_map(
-            static fn (array $row): int => (int) $row['medicine_id'],
-            array_values($sourceRows)
-        );
-        $unmappedMedicineIds = array_diff($medicineIds, $mappedMedicineIds);
-        if ($unmappedMedicineIds !== []) {
-            throw new DomainException('Medicine import preflight found unrelated existing catalog rows');
-        }
-    }
-
     /**
      * @param array<string,mixed> $input
      * @param array<string,array<string,mixed>> $sourceRows
      * @param array<int,array<string,mixed>> $medicineRows
      * @return array<string,mixed>
      */
-    private static function batchReplay(array $input, array $sourceRows, array $medicineRows): array
-    {
+    private static function batchReplay(
+        array $input,
+        array $sourceRows,
+        array $medicineRows,
+        Sequence $sequence
+    ): array {
         $results = [];
         foreach ($input['items'] as $item) {
             $sourceId = $item['source_medicine_id'];
@@ -179,26 +163,55 @@ final class MedicineImportService
             }
             $results[] = self::existingResult(
                 $sourceRows[$sourceId],
+                $item,
                 MedicineImportPayload::itemHash($item),
-                $medicineRows
+                $medicineRows,
+                $sequence
             );
         }
         return self::response($input, $results, 0, count($results), true);
     }

-    /** @param array<string,mixed> $source @param array<int,array<string,mixed>> $medicineRows */
-    private static function existingResult(array $source, string $itemHash, array $medicineRows): array
-    {
+    /**
+     * @param array<string,mixed> $source
+     * @param array<string,mixed> $item
+     * @param array<int,array<string,mixed>> $medicineRows
+     */
+    private static function existingResult(
+        array $source,
+        array $item,
+        string $itemHash,
+        array &$medicineRows,
+        Sequence $sequence
+    ): array {
         if (!hash_equals((string) $source['payload_hash'], $itemHash)) {
             throw new DomainException('Source medicine identity conflicts with a different payload');
         }
-        $medicine = $medicineRows[(int) $source['medicine_id']] ?? null;
-        if ($medicine === null || $medicine['delete_time'] !== null) {
-            throw new DomainException('Source medicine mapping points to a deleted or missing medicine');
+        $medicineId = (int) $source['medicine_id'];
+        $medicine = $medicineRows[$medicineId] ?? null;
+        if ($medicine === null) {
+            throw new DomainException('Source medicine mapping points to a missing medicine');
         }
         if (!hash_equals((string) $source['medicine_code'], (string) $medicine['medicine_code'])) {
             throw new RuntimeException('Source medicine mapping code is inconsistent');
         }
+
+        if ($medicine['delete_time'] !== null || (int) $medicine['status'] !== (int) $item['status']) {
+            $catalogVersion = self::nextCatalogVersion($sequence);
+            $updateTime = time();
+            Db::name('pharmacy_medicine')->where('id', $medicineId)->update([
+                'delete_time' => null,
+                'status' => (int) $item['status'],
+                'catalog_version' => $catalogVersion,
+                'update_time' => $updateTime,
+            ]);
+            $medicine['delete_time'] = null;
+            $medicine['status'] = (int) $item['status'];
+            $medicine['catalog_version'] = $catalogVersion;
+            $medicine['update_time'] = $updateTime;
+            $medicineRows[$medicineId] = $medicine;
+        }
+
         return [
             'source_medicine_id' => (string) $source['source_medicine_id'],
             'medicine_code' => (string) $medicine['medicine_code'],
@@ -207,6 +220,16 @@ final class MedicineImportService
         ];
     }

+    private static function nextCatalogVersion(Sequence $sequence): int
+    {
+        return CatalogVersionAllocator::next(
+            static fn (): int => (int) $sequence->current_value,
+            static function (int $next) use ($sequence): void {
+                $sequence->save(['current_value' => $next]);
+            }
+        );
+    }
+
     /** @param array<string,mixed> $input @param list<array<string,mixed>> $items */
     private static function response(
         array $input,
diff --git a/server/tests/pharmacy/medicine_import_mysql_integration.php b/server/tests/pharmacy/medicine_import_mysql_integration.php
index 34b6b7a..9464b18 100644
--- a/server/tests/pharmacy/medicine_import_mysql_integration.php
+++ b/server/tests/pharmacy/medicine_import_mysql_integration.php
@@ -2,6 +2,7 @@

 declare(strict_types=1);

+use app\adminapi\logic\pharmacy\MedicineLogic;
 use app\command\PharmacyResetTestCatalog;
 use app\common\service\pharmacy\MedicineCatalogResetService;
 use app\common\service\pharmacy\MedicineImportPayload;
@@ -297,6 +298,21 @@ try {
     $assertSame(2, $count('pharmacy_medicine_import_batch'), 'source replay in another batch must record that batch once');
     $assertSame(1, $count('pharmacy_medicine'), 'source replay must not add a medicine');
     $assertSame(1, $count('pharmacy_stock'), 'source replay must not add stock');
+    $assertSame(
+        $medicineOne,
+        Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find(),
+        'enabled mapped medicine replay must preserve the medicine row byte for byte'
+    );
+    $assertSame(
+        $sourceOne,
+        Db::name('pharmacy_medicine_source')->where('id', (int) $sourceOne['id'])->find(),
+        'enabled mapped medicine replay must preserve its source mapping byte for byte'
+    );
+    $assertSame(
+        $stockOne,
+        Db::name('pharmacy_stock')->where('id', (int) $stockOne['id'])->find(),
+        'enabled mapped medicine replay must preserve its stock row byte for byte'
+    );
     $normalizedDifferentBatchReplay = MedicineImportPayload::normalize($differentBatchReplay);
     $secondBatchRow = Db::name('pharmacy_medicine_import_batch')
         ->where('merchant_id', 9001)
@@ -340,6 +356,104 @@ try {
     $assertSame(2, $count('pharmacy_medicine_import_batch'), 'source conflict must roll back the batch');
     ++$passed;

+    $medicineCountBeforeRestore = $count('pharmacy_medicine');
+    $sourceCountBeforeRestore = $count('pharmacy_medicine_source');
+    $stockCountBeforeRestore = $count('pharmacy_stock');
+    $sourceBeforeRestore = Db::name('pharmacy_medicine_source')->where('id', (int) $sourceOne['id'])->find();
+    $stockBeforeRestore = Db::name('pharmacy_stock')->where('id', (int) $stockOne['id'])->find();
+    $assertSame(true, MedicineLogic::delete((int) $medicineOne['id']), 'admin delete must soft-delete the mapped medicine');
+    $softDeleted = Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find();
+    $assertSame(0, (int) $softDeleted['status'], 'admin delete must disable the mapped medicine');
+    $assertTrue($softDeleted['delete_time'] !== null, 'admin delete must retain the mapped medicine as soft-deleted');
+    $sequenceAfterSoftDelete = (int) Db::name('pharmacy_sequence')
+        ->where('sequence_name', 'medicine_catalog')
+        ->value('current_value');
+
+    $restoreSoftDeletedBatch = $batchOne;
+    $restoreSoftDeletedBatch['import_id'] = 'integration-batch-restore-soft-deleted';
+    $restoredSoftDeletedResult = MedicineImportService::import(9001, $restoreSoftDeletedBatch);
+    $restoredSoftDeleted = Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find();
+    $assertSame(false, $restoredSoftDeletedResult['idempotent'], 'soft-delete restoration in a new batch must not be idempotent');
+    $assertSame(0, $restoredSoftDeletedResult['created_count'], 'soft-delete restoration must not create a medicine');
+    $assertSame(1, $restoredSoftDeletedResult['existing_count'], 'soft-delete restoration must count the mapping as existing');
+    $assertSame('existing', $restoredSoftDeletedResult['items'][0]['action'], 'soft-delete restoration must preserve the existing action');
+    $assertSame((int) $medicineOne['id'], (int) $restoredSoftDeleted['id'], 'soft-delete restoration must reuse the original medicine id');
+    $assertSame((string) $medicineOne['medicine_code'], (string) $restoredSoftDeleted['medicine_code'], 'soft-delete restoration must reuse the original medicine code');
+    $assertSame(null, $restoredSoftDeleted['delete_time'], 'soft-delete restoration must clear delete_time');
+    $assertSame(1, (int) $restoredSoftDeleted['status'], 'soft-delete restoration must re-enable the medicine');
+    $assertSame($sequenceAfterSoftDelete + 1, (int) $restoredSoftDeleted['catalog_version'], 'soft-delete restoration must publish a new catalog version');
+    $assertSame((int) $restoredSoftDeleted['catalog_version'], $restoredSoftDeletedResult['items'][0]['catalog_version'], 'soft-delete response must return the restored catalog version');
+    $assertSame($medicineCountBeforeRestore, $count('pharmacy_medicine'), 'soft-delete restoration must not add a medicine row');
+    $assertSame($sourceCountBeforeRestore, $count('pharmacy_medicine_source'), 'soft-delete restoration must not add a source mapping');
+    $assertSame($stockCountBeforeRestore, $count('pharmacy_stock'), 'soft-delete restoration must not add a stock row');
+    $assertSame($sourceBeforeRestore, Db::name('pharmacy_medicine_source')->where('id', (int) $sourceOne['id'])->find(), 'soft-delete restoration must not mutate the source mapping');
+    $assertSame($stockBeforeRestore, Db::name('pharmacy_stock')->where('id', (int) $stockOne['id'])->find(), 'soft-delete restoration must not mutate stock');
+    $softDeletedComparable = $softDeleted;
+    $restoredSoftDeletedComparable = $restoredSoftDeleted;
+    foreach (['delete_time', 'status', 'catalog_version', 'update_time'] as $restoredField) {
+        unset($softDeletedComparable[$restoredField], $restoredSoftDeletedComparable[$restoredField]);
+    }
+    $assertSame($softDeletedComparable, $restoredSoftDeletedComparable, 'soft-delete restoration must preserve every other medicine field');
+    ++$passed;
+
+    $assertTrue(is_array(MedicineLogic::edit([
+        'id' => (int) $restoredSoftDeleted['id'],
+        'name' => (string) $restoredSoftDeleted['name'],
+        'brand' => (string) $restoredSoftDeleted['brand'],
+        'unit' => (string) $restoredSoftDeleted['unit'],
+        'settlement_price' => (string) $restoredSoftDeleted['settlement_price'],
+        'retail_price' => (string) $restoredSoftDeleted['retail_price'],
+        'status' => 0,
+    ])), 'admin edit must disable the mapped medicine without deleting it');
+    $disabledMedicine = Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find();
+    $assertSame(null, $disabledMedicine['delete_time'], 'admin edit must leave the mapped medicine undeleted');
+    $assertSame(0, (int) $disabledMedicine['status'], 'admin edit must disable the mapped medicine');
+    $sequenceAfterDisable = (int) Db::name('pharmacy_sequence')
+        ->where('sequence_name', 'medicine_catalog')
+        ->value('current_value');
+
+    $restoreDisabledBatch = $batchOne;
+    $restoreDisabledBatch['import_id'] = 'integration-batch-restore-disabled';
+    $restoredDisabledResult = MedicineImportService::import(9001, $restoreDisabledBatch);
+    $restoredDisabled = Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find();
+    $assertSame(false, $restoredDisabledResult['idempotent'], 'disabled restoration in a new batch must not be idempotent');
+    $assertSame(0, $restoredDisabledResult['created_count'], 'disabled restoration must not create a medicine');
+    $assertSame(1, $restoredDisabledResult['existing_count'], 'disabled restoration must count the mapping as existing');
+    $assertSame('existing', $restoredDisabledResult['items'][0]['action'], 'disabled restoration must preserve the existing action');
+    $assertSame((int) $disabledMedicine['id'], (int) $restoredDisabled['id'], 'disabled restoration must reuse the original medicine id');
+    $assertSame((string) $disabledMedicine['medicine_code'], (string) $restoredDisabled['medicine_code'], 'disabled restoration must reuse the original medicine code');
+    $assertSame(null, $restoredDisabled['delete_time'], 'disabled restoration must leave delete_time clear');
+    $assertSame(1, (int) $restoredDisabled['status'], 'disabled restoration must re-enable the medicine');
+    $assertSame($sequenceAfterDisable + 1, (int) $restoredDisabled['catalog_version'], 'disabled restoration must publish a new catalog version');
+    $assertSame((int) $restoredDisabled['catalog_version'], $restoredDisabledResult['items'][0]['catalog_version'], 'disabled response must return the restored catalog version');
+    $assertSame($medicineCountBeforeRestore, $count('pharmacy_medicine'), 'disabled restoration must not add a medicine row');
+    $assertSame($sourceCountBeforeRestore, $count('pharmacy_medicine_source'), 'disabled restoration must not add a source mapping');
+    $assertSame($stockCountBeforeRestore, $count('pharmacy_stock'), 'disabled restoration must not add a stock row');
+    $assertSame($sourceBeforeRestore, Db::name('pharmacy_medicine_source')->where('id', (int) $sourceOne['id'])->find(), 'disabled restoration must not mutate the source mapping');
+    $assertSame($stockBeforeRestore, Db::name('pharmacy_stock')->where('id', (int) $stockOne['id'])->find(), 'disabled restoration must not mutate stock');
+    ++$passed;
+
+    $assertSame(true, MedicineLogic::delete((int) $medicineOne['id']), 'admin delete must prepare the batch replay restoration case');
+    $sequenceBeforeBatchReplayRestore = (int) Db::name('pharmacy_sequence')
+        ->where('sequence_name', 'medicine_catalog')
+        ->value('current_value');
+    $batchCountBeforeReplayRestore = $count('pharmacy_medicine_import_batch');
+    $replayRestoredResult = MedicineImportService::import(9001, $restoreDisabledBatch);
+    $replayRestored = Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find();
+    $assertSame(true, $replayRestoredResult['idempotent'], 'same completed batch replay must remain idempotent');
+    $assertSame(0, $replayRestoredResult['created_count'], 'batch replay restoration must not create a medicine');
+    $assertSame(1, $replayRestoredResult['existing_count'], 'batch replay restoration must count the mapping as existing');
+    $assertSame('existing', $replayRestoredResult['items'][0]['action'], 'batch replay restoration must preserve the existing action');
+    $assertSame((int) $medicineOne['id'], (int) $replayRestored['id'], 'batch replay restoration must reuse the original medicine id');
+    $assertSame((string) $medicineOne['medicine_code'], (string) $replayRestored['medicine_code'], 'batch replay restoration must reuse the original medicine code');
+    $assertSame(null, $replayRestored['delete_time'], 'batch replay restoration must clear delete_time');
+    $assertSame(1, (int) $replayRestored['status'], 'batch replay restoration must re-enable the medicine');
+    $assertSame($sequenceBeforeBatchReplayRestore + 1, (int) $replayRestored['catalog_version'], 'batch replay restoration must publish a new catalog version');
+    $assertSame($batchCountBeforeReplayRestore, $count('pharmacy_medicine_import_batch'), 'batch replay restoration must not add a batch row');
+    $assertSame($sourceBeforeRestore, Db::name('pharmacy_medicine_source')->where('id', (int) $sourceOne['id'])->find(), 'batch replay restoration must not mutate the source mapping');
+    $assertSame($stockBeforeRestore, Db::name('pharmacy_stock')->where('id', (int) $stockOne['id'])->find(), 'batch replay restoration must not mutate stock');
+    ++$passed;
+
     $continuation = [
         'source_system' => 'zyt',
         'import_id' => 'integration-batch-0004',
@@ -360,7 +474,47 @@ try {
     $assertSame(0, (int) $medicineTwo['status'], 'continued import disabled status must persist');
     ++$passed;

-    Db::name('pharmacy_medicine')->insert([
+    $assertTrue(is_array(MedicineLogic::edit([
+        'id' => (int) $replayRestored['id'],
+        'name' => (string) $replayRestored['name'],
+        'brand' => (string) $replayRestored['brand'],
+        'unit' => (string) $replayRestored['unit'],
+        'settlement_price' => (string) $replayRestored['settlement_price'],
+        'retail_price' => (string) $replayRestored['retail_price'],
+        'status' => 0,
+    ])), 'admin edit must prepare the transactional restoration rollback case');
+    $disabledBeforeRollback = Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find();
+    $sequenceBeforeRollback = (int) Db::name('pharmacy_sequence')
+        ->where('sequence_name', 'medicine_catalog')
+        ->value('current_value');
+    $batchCountBeforeRollback = $count('pharmacy_medicine_import_batch');
+    $rollbackBatch = [
+        'source_system' => 'zyt',
+        'import_id' => 'integration-batch-restore-rollback',
+        'items' => [$itemOne, $continuation['items'][0]],
+    ];
+    $rollbackBatch['items'][1]['retail_price'] = '9.9999';
+    $restoreRolledBack = false;
+    try {
+        MedicineImportService::import(9001, $rollbackBatch);
+    } catch (DomainException $exception) {
+        $restoreRolledBack = str_contains($exception->getMessage(), 'Source medicine identity conflicts');
+    }
+    $assertTrue($restoreRolledBack, 'later source conflict must reject a batch after an earlier restoration attempt');
+    $assertSame(
+        $disabledBeforeRollback,
+        Db::name('pharmacy_medicine')->where('id', (int) $medicineOne['id'])->find(),
+        'source conflict must roll back the earlier medicine restoration byte for byte'
+    );
+    $assertSame(
+        $sequenceBeforeRollback,
+        (int) Db::name('pharmacy_sequence')->where('sequence_name', 'medicine_catalog')->value('current_value'),
+        'source conflict must roll back the catalog version allocation'
+    );
+    $assertSame($batchCountBeforeRollback, $count('pharmacy_medicine_import_batch'), 'source conflict must not persist a partial restoration batch');
+    ++$passed;
+
+    $unrelatedMedicineId = Db::name('pharmacy_medicine')->insertGetId([
         'medicine_code' => 'IT_UNRELATED',
         'name' => 'Unrelated old medicine',
         'brand' => '',
@@ -375,14 +529,13 @@ try {
     $unrelatedAttempt = $continuation;
     $unrelatedAttempt['import_id'] = 'integration-batch-0005';
     $unrelatedAttempt['items'][0]['source_medicine_id'] = '103';
-    $rejected = false;
-    try {
-        MedicineImportService::import(9001, $unrelatedAttempt);
-    } catch (DomainException $exception) {
-        $rejected = str_contains($exception->getMessage(), 'unrelated existing catalog');
-    }
-    $assertTrue($rejected, 'unmapped old medicine must fail bootstrap preflight');
-    $assertSame(2, $count('pharmacy_medicine_source'), 'preflight rejection must not add a source mapping');
+    $unrelatedBefore = Db::name('pharmacy_medicine')->where('id', $unrelatedMedicineId)->find();
+    $incrementalResult = MedicineImportService::import(9001, $unrelatedAttempt);
+    $unrelatedAfter = Db::name('pharmacy_medicine')->where('id', $unrelatedMedicineId)->find();
+    $assertSame('created', $incrementalResult['items'][0]['action'], 'unrelated EJ medicines must not block incremental ZYT additions');
+    $assertSame(3, $count('pharmacy_medicine_source'), 'incremental import must add only the new ZYT source mapping');
+    $assertSame($unrelatedBefore, $unrelatedAfter, 'incremental import must preserve unrelated EJ medicines byte for byte');
+    $assertSame(4, $count('pharmacy_medicine'), 'incremental import must append without replacing the EJ catalog');
     ++$passed;

     config([
diff --git a/server/tests/pharmacy/route_contracts.php b/server/tests/pharmacy/route_contracts.php
index 39deace..7ce525c 100644
--- a/server/tests/pharmacy/route_contracts.php
+++ b/server/tests/pharmacy/route_contracts.php
@@ -94,8 +94,9 @@ foreach ([
 }
 $assertTrue(!str_contains($medicineImportService, 'StockLedger'), 'zero-stock bootstrap must not append a stock ledger movement');
 $assertTrue(
-    str_contains($medicineImportService, 'array_diff($medicineIds, $mappedMedicineIds)'),
-    'bootstrap preflight must reject only medicines without a source mapping and allow deterministic continuation'
+    !str_contains($medicineImportService, 'assertCatalogBelongsToBootstrap')
+        && !str_contains($medicineImportService, 'unrelated existing catalog rows'),
+    'incremental medicine import must preserve EJ medicines that were created by other sources'
 );

 $catalogLockPath = dirname(__DIR__, 2) . '/app/common/service/pharmacy/MedicineCatalogLock.php';
