<?php

declare(strict_types=1);

/** Disposable local MySQL only. Never initializes the app or loads production config. */
require dirname(__DIR__) . '/vendor/autoload.php';
require dirname(__DIR__) . '/vendor/topthink/framework/src/helper.php';

use app\common\service\prescriptionai\PrescriptionAiStore as Store;
use app\common\service\prescriptionai\PrescriptionAiPolicy as Policy;
use app\common\service\prescriptionai\PrescriptionAiCipher as Cipher;
use app\common\service\prescriptionai\PrescriptionAiProgress as Progress;
use app\common\service\prescriptionai\PrescriptionAiRequest as SaveRequest;
use app\adminapi\logic\tcm\PrescriptionAiLogic as Api;
use think\Container;
use think\facade\Db;

$port = (int) getenv('ZYT_AI_TEST_MYSQL_PORT');
if ($port <= 0) {
    throw new RuntimeException('Set ZYT_AI_TEST_MYSQL_PORT to an isolated local empty-password MySQL instance');
}
$child = ($argv[1] ?? '') === '--claim';
$legacyProgressSchema = in_array('--legacy-progress-schema', $argv, true);
$database = $child ? (string) getenv('ZYT_AI_TEST_DATABASE') : 'prescription_ai_test_' . bin2hex(random_bytes(6));
if (!preg_match('/^prescription_ai_test_[a-f0-9]{12}$/D', $database)) {
    throw new RuntimeException('Only disposable test database names are allowed');
}
$pdo = new PDO("mysql:host=127.0.0.1;port={$port};charset=utf8mb4", 'root', '', [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
if (!$child) {
    $pdo->exec("CREATE DATABASE `{$database}` CHARACTER SET utf8mb4");
}
$pdo->exec("USE `{$database}`");
$app = new think\App(); // no initialize()
$manager = new think\DbManager();
$manager->setConfig(['default' => 'mysql', 'auto_timestamp' => true, 'datetime_format' => false,
    'connections' => ['mysql' => ['type' => 'mysql', 'hostname' => '127.0.0.1', 'hostport' => $port,
        'database' => $database, 'username' => 'root', 'password' => '', 'charset' => 'utf8mb4',
        'prefix' => 'zyt_', 'fields_strict' => true]]]);
Container::getInstance()->instance('think\DbManager', $manager);
$config = new think\Config();
$config->set(['enabled' => true, 'encryption_key' => str_repeat('isolated-test-', 4), 'debounce_seconds' => 0,
    'lease_seconds' => 600, 'max_attempts' => 3, 'max_manual_retries' => 2, 'max_parallel_per_model' => 1,
    'daily_model_tasks' => 200, 'transcript_wait_seconds' => 300], 'prescription_analysis');
Container::getInstance()->instance('config', $config);
if ($child) {
    $claim = Store::claimTask($argv[2] ?? 'qwen');
    echo json_encode(['id' => $claim['id'] ?? null]) . "\n";
    exit(0);
}
$checks = 0;
$expect = static function (bool $ok, string $why) use (&$checks): void {
    if (!$ok) { throw new RuntimeException($why); }
    $checks++;
};
$root = ['root' => 1, 'admin_id' => 1, 'id' => 1, 'role_id' => [], 'dept_id' => [], 'name' => 'Test'];
try {
    $pdo->exec('CREATE TABLE zyt_system_menu (id INT PRIMARY KEY AUTO_INCREMENT,pid INT,type VARCHAR(5),name VARCHAR(100),icon VARCHAR(50),sort INT,perms VARCHAR(100),paths VARCHAR(100),component VARCHAR(100),selected VARCHAR(100),params VARCHAR(100),is_cache INT,is_show INT,is_disable INT,create_time INT,update_time INT)');
    $pdo->exec('CREATE TABLE zyt_system_role_menu (role_id INT,menu_id INT,UNIQUE KEY(role_id,menu_id))');
    $pdo->exec('CREATE TABLE zyt_admin (id INT PRIMARY KEY,name VARCHAR(50),root INT,disable INT,delete_time INT NULL)');
    $pdo->exec('CREATE TABLE zyt_admin_role (admin_id INT,role_id INT)');
    $pdo->exec('CREATE TABLE zyt_admin_dept (admin_id INT,dept_id INT)');
    $pdo->exec('CREATE TABLE zyt_admin_jobs (admin_id INT,jobs_id INT)');
    $pdo->exec("INSERT INTO zyt_admin VALUES(1,'Test',1,0,NULL)");
    $pdo->exec('CREATE TABLE zyt_tcm_diagnosis (id INT PRIMARY KEY,patient_id INT,assistant_id INT DEFAULT 1,delete_time INT NULL)');
    $pdo->exec('INSERT INTO zyt_tcm_diagnosis(id,patient_id) VALUES(1,100),(2,200)');
    $pdo->exec('CREATE TABLE zyt_tcm_prescription (id INT PRIMARY KEY AUTO_INCREMENT,diagnosis_id INT DEFAULT 1,patient_id INT DEFAULT 100,creator_id INT DEFAULT 1,is_system_auto INT DEFAULT 0,void_status INT DEFAULT 0,delete_time INT NULL,herbs TEXT,prescription_type VARCHAR(30) DEFAULT "饮片",update_time INT DEFAULT 0) ENGINE=InnoDB');
    foreach (['sn','prescription_name','dosage_unit','patient_name','phone','visit_no','prescription_date','pulse','pulse_condition',
        'tongue','tongue_image','clinical_diagnosis','case_record','dose_unit','aux_usage','usage_instruction','usage_time','usage_way',
        'dietary_taboo','usage_notes','doctor_name','doctor_signature','visible_role_ids','audit_by_name','audit_remark','void_by_name'] as $field) {
        $pdo->exec("ALTER TABLE zyt_tcm_prescription ADD `$field` TEXT NULL");
    }
    foreach (['appointment_id','assistant_id','gender','age','dosage_bag_count','need_decoction','bags_per_dose','dose_count',
        'usage_days','times_per_day','template_id','is_shared','audit_status','audit_time','audit_by','void_time','void_by','create_time'] as $field) {
        $pdo->exec("ALTER TABLE zyt_tcm_prescription ADD `$field` INT DEFAULT 0");
    }
    $pdo->exec('ALTER TABLE zyt_tcm_prescription ADD dosage_amount DECIMAL(10,2) NULL, ADD amount DECIMAL(10,2) DEFAULT 0');
    $pdo->exec('CREATE TABLE zyt_tcm_prescription_order (id INT PRIMARY KEY,prescription_id INT,source_prescription_id INT,prescription_audit_status INT,fulfillment_status INT,delete_time INT NULL)');
    $pdo->exec('CREATE TABLE zyt_doctor_medicine (id INT PRIMARY KEY,name VARCHAR(100),unit VARCHAR(20),status INT,delete_time INT NULL)');
    $pdo->exec("INSERT INTO zyt_doctor_medicine VALUES(1,'测试药材','g',1,NULL)");
    $migration = file_get_contents(dirname(__DIR__) . '/database/migrations/2026_09_09_prescription_ai_analysis.sql');
    foreach (explode(';', preg_replace('/^--.*$/m', '', $migration)) as $statement) {
        if (trim($statement) !== '') { $pdo->exec($statement); }
    }
    // Migrations are rerunnable including role grants.
    foreach (explode(';', preg_replace('/^--.*$/m', '', $migration)) as $statement) {
        if (trim($statement) !== '') { $pdo->exec($statement); }
    }
    $expect((int) Db::name('system_menu')->count() === 7, 'idempotent permission migration');
    if (!$legacyProgressSchema) {
        $progressMigration = file_get_contents(dirname(__DIR__) . '/database/migrations/2026_09_10_prescription_ai_progress.sql');
        for ($i = 0; $i < 2; $i++) {
            foreach (explode(';', preg_replace('/^--.*$/m', '', $progressMigration)) as $statement) {
                if (trim($statement) !== '') { $pdo->exec($statement); }
            }
        }
        $expect((int) $pdo->query("SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE()
            AND TABLE_NAME = 'zyt_prescription_ai_task' AND COLUMN_NAME = 'progress_json'")->fetchColumn() === 1,
            'additive progress migration is rerunnable');
    }
    $expect(Store::supportsProgress() === !$legacyProgressSchema, 'old and migrated schema detected without reading model caches');
    $fixture = static function (array $extra = []) use ($root): array {
        $id = (int) Db::name('tcm_prescription')->insertGetId($extra + [
            'herbs' => json_encode([['medicine_id' => 1, 'name' => '测试药材', 'dosage' => 10]], JSON_UNESCAPED_UNICODE),
            'update_time' => time(),
        ]);
        return Db::name('tcm_prescription')->where('id', $id)->find();
    };
    $save = static function (array $rx, array $options = []) use ($root): ?int {
        return Db::transaction(static function () use ($rx, $options, $root): ?int {
            $fresh = Db::name('tcm_prescription')->where('id', $rx['id'])->lock(true)->find();
            return Store::recordSaved($fresh, 1, $root, ['ai_assisted' => false] + $options);
        });
    };
    $context = ['source' => ['clinical' => 'synthetic record'], 'source_hash' => hash('sha256', 'fixed'),
        'source_diagnosis_ids' => [1], 'source_summary' => ['diagnosis_count' => 1], 'missing' => [],
        'baseline_eligible' => false, 'baseline_exclusion_reasons' => ['SOURCE_HISTORY_VERSIONS_UNAVAILABLE'],
        'comparison_type' => 'latest_context', 'cutoff_at' => time(), 'wait_for_transcript' => false];
    $rx = $fixture();
    $context['source_access_manifest'] = ['schema_version' => 'prescription-source-access-v1', 'patient_id' => 100,
        'target' => ['prescription_id' => (int) $rx['id'], 'diagnosis_id' => 1],
        'records' => [['source_kind' => 'diagnoses', 'id' => 1, 'source_id' => 'diagnoses:1', 'diagnosis_id' => 1, 'patient_id' => 100, 'staff' => []]]];
    $batchId = $save($rx);
    $expect($batchId > 0 && $save($rx) === $batchId, 'same clinical content enqueues once');
    $blank = $fixture(['is_system_auto' => 1, 'herbs' => '[]']);
    $expect($save($blank) === null, 'blank prescriptions do not enqueue');
    $expect((int) Db::name('prescription_ai_task')->count() === 0, 'no model call or task before snapshot');
    $claim = Store::claimBatch();
    $expect((int) $claim['id'] === $batchId && Store::claimBatch() === null, 'preparation lease prevents duplicate claim');
    $expect(Store::finishPreparation($claim, $context), 'snapshot prepares');
    $expect(!Store::finishPreparation($claim, $context), 'preparation cannot run twice');
    $expect((int) Db::name('prescription_ai_task')->count() === 2, 'exactly two model tasks');
    $stored = Db::name('prescription_ai_batch')->find($batchId);
    $expect(!str_contains($stored['context_cipher'], 'synthetic record'), 'context encrypted');
    $expect((new Cipher())->decrypt($stored['context_cipher'], 'context')['source_hash'] === $context['source_hash'], 'context decrypts exactly');
    $expect((int) $stored['baseline_eligible'] === 0, 'uncertain historical provenance excluded');
    $q = Store::claimTask('qwen');
    $o = Store::claimTask('openai');
    $expect($q !== null && $o !== null && Store::claimTask('qwen') === null, 'models run independently, claim exclusive');
    $expect(Store::checkpoint($q, ['stage' => 'first', 'steps' => ['private fixture']]), 'checkpoint persists');
    $expect(!str_contains(Db::name('prescription_ai_task')->where('id', $q['id'])->value('progress_cipher'), 'private fixture'), 'checkpoint encrypted');
    $beforeProgressCipher = Db::name('prescription_ai_task')->where('id', $q['id'])->value('progress_cipher');
    $progressPayload = ['public' => Progress::advance([], 'text', 'waiting', 1, 3), 'steps' => ['must-not-replace-cache'],
        'prompt' => 'never-public'];
    $expect(Store::checkpoint($q, $progressPayload, false) && Store::checkpoint($q, $progressPayload, false),
        'same-second metadata no-op still recognizes a valid lease');
    $expect(Db::name('prescription_ai_task')->where('id', $q['id'])->value('progress_cipher') === $beforeProgressCipher,
        'metadata-only checkpoint leaves encrypted model cache unchanged');
    if (!$legacyProgressSchema) {
        $publicJson = Db::name('prescription_ai_task')->where('id', $q['id'])->value('progress_json');
        $expect(!str_contains($publicJson, 'never-public') && !str_contains($publicJson, 'must-not-replace-cache')
            && json_decode($publicJson, true)['completed_units'] === 1, 'database stores only sanitized scalar progress');
    }
    $modelQueries = [];
    $captureModels = true;
    Db::listen(static function (string $sql) use (&$modelQueries, &$captureModels): void {
        if ($captureModels && str_contains($sql, 'prescription_ai_task') && preg_match('/^SELECT/i', $sql)) { $modelQueries[] = $sql; }
    });
    $liveStatus = Api::statuses([$rx['id']], 1, $root)['items'][0];
    $liveDetail = Api::detail($batchId, 1, $root);
    $liveReports = Api::reports(['prescription_id' => $rx['id']], 1, $root);
    $captureModels = false;
    $expect($modelQueries !== [], 'list/detail/history task select queries observed');
    foreach ($modelQueries as $sql) {
        $expect(!str_contains($sql, 'progress_cipher') && !preg_match('/SELECT\s+\*/i', $sql),
            'polling selects bounded task fields, not encrypted model cache');
    }
    $expect($liveStatus['models']['qwen']['progress']['stage'] === ($legacyProgressSchema ? 'unknown' : 'text')
        && $liveDetail['models']['qwen']['progress']['stage'] === ($legacyProgressSchema ? 'unknown' : 'text'),
        'status and detail present measured progress, with honest old-schema fallback');
    $expect(isset($liveStatus['progress']) && isset($liveReports['lists'][0]['models']['qwen']['progress']),
        'batch and history include public progress');
    $output = ['report' => ['summary' => 'synthetic report'], 'candidate' => ['status' => 'available_for_review'],
        'coverage' => ['status' => 'complete'], 'model_name' => 'test', 'prompt_version' => 'test-v1'];
    $comparison = ['status' => 'comparable', 'score' => 80, 'herb_score' => 100, 'algorithm_version' => 'test-v1'];
    $expect(Store::complete($q, $output, $comparison), 'first model completes');
    $expect(!Store::complete($q, $output, $comparison), 'duplicate result callback fenced');
    Store::fail($o, 'UPSTREAM_TIMEOUT', false);
    $expect(Db::name('prescription_ai_batch')->where('id', $batchId)->value('status') === 'partial', 'one failure preserves other result');
    $statuses = Api::statuses([$rx['id']], 1, $root);
    $expect($statuses['items'][0]['models']['qwen']['score'] === 80.0, 'list returns persisted numeric score');
    $expect($statuses['items'][0]['models']['openai']['score'] === null, 'failed score is null rather than zero');
    $detail = Api::detail($batchId, 1, $root);
    $expect($detail['models']['qwen']['report']['summary'] === 'synthetic report', 'authorized detail decrypts');
    $expect($detail['models']['qwen']['progress']['stage'] === 'completed'
        && $statuses['items'][0]['models']['openai']['progress']['stage'] === 'failed', 'persisted task terminal state overrides stale stage');
    $resultRow = Db::name('prescription_ai_result')->where('batch_id', $batchId)->where('model_key', 'qwen')->find();
    $resultBody = (new Cipher())->decrypt($resultRow['body_cipher'], 'result:' . $batchId . ':qwen');
    $resultBody['progress'] = ['stage' => 'poisoned', 'notice' => 'model supplied text'];
    Db::name('prescription_ai_result')->where('id', $resultRow['id'])->update([
        'body_cipher' => (new Cipher())->encrypt($resultBody, 'result:' . $batchId . ':qwen'),
    ]);
    $expect(Api::detail($batchId, 1, $root)['models']['qwen']['progress']['stage'] === 'completed',
        'report body cannot override trusted task progress');
    Api::review($batchId, 'qwen', 'not_adopted', 'test comment', 1, $root);
    $expect(Api::detail($batchId, 1, $root)['models']['qwen']['review']['status'] === 'not_adopted', 'review independent of prescription');
    Api::retry($batchId, 'openai', 1, $root);
    $o2 = Store::claimTask('openai');
    $expect((int) $o2['total_attempts'] === 2 && (int) $o2['attempts'] === 1, 'manual retry preserves lifetime attempts');
    $expect(Store::complete($o2, $output, $comparison), 'failed model retries without repeating successful model');
    $expect(Db::name('prescription_ai_batch')->where('id', $batchId)->value('status') === 'success', 'both success aggregate');
    $stats = Api::statistics([], 1, $root);
    $expect($stats['doctors'][0]['models']['qwen']['mean'] === null, 'nonbaseline scores never become doctor accuracy');
    Db::name('prescription_ai_batch')->where('id', $batchId)->update(['baseline_eligible' => 1, 'baseline_exclusions_json' => '[]']);
    $eligibleStats = Api::statistics([], 1, $root);
    $expect($eligibleStats['doctors'][0]['models']['qwen']['mean'] === 80.0, 'synthetic qualified baseline has empty exclusion reason');
    Db::name('prescription_ai_batch')->where('id', $batchId)->update(['baseline_eligible' => 0, 'baseline_exclusions_json' => '["SOURCE_HISTORY_VERSIONS_UNAVAILABLE"]']);
    $expect((int) Db::name('prescription_ai_attempt')->count() === 3, 'attempt history retained');
    $payload = ['request_key' => '12345678-abcd-1234-abcd-123456789012', 'herbs' => [['name' => 'fixture']]];
    Db::transaction(static function () use ($payload, $rx, $expect): void {
        $expect(SaveRequest::replay($payload, 1, true) === null, 'first request reserved');
        SaveRequest::complete($payload, 1, (int) $rx['id']);
    });
    $expect(SaveRequest::replay($payload, 1) === (int) $rx['id'], 'lost save response replays same prescription');
    try { SaveRequest::replay($payload + ['extra' => 'changed'], 1); $expect(false, 'changed content cannot reuse key'); }
    catch (DomainException $e) { $checks++; }
    $countBefore = (int) Db::name('prescription_ai_batch')->count();
    try {
        Db::transaction(static function () use ($fixture, $save): void {
            $save($fixture());
            throw new RuntimeException('rollback fixture');
        });
    } catch (RuntimeException $e) {}
    $expect((int) Db::name('prescription_ai_batch')->count() === $countBefore, 'prescription and outbox roll back together');
    Db::name('tcm_prescription')->where('id', $rx['id'])->update(['herbs' => '[{"name":"changed","dosage":20}]']);
    $newBatch = $save($rx);
    $expect($newBatch !== $batchId, 'clinical change creates immutable new batch');
    $expect(Db::name('prescription_ai_batch')->where('id', $batchId)->value('validity') === 'prescription_changed', 'previous version invalidated');
    $expect((int) Db::name('prescription_ai_result')->count() === 2, 'historic model results immutable');
    $newClaim = Store::claimBatch();
    Store::finishPreparation($newClaim, $context);
    $lease = Store::claimTask('qwen');
    Db::name('prescription_ai_task')->where('id', $lease['id'])->update(['lock_until' => time() - 1]);
    $replacement = Store::claimTask('qwen');
    $expect($replacement !== null && $replacement['lock_token'] !== $lease['lock_token'], 'expired lease recovered');
    $expect(!Store::checkpoint($lease, []) && !Store::complete($lease, $output, $comparison), 'old worker cannot write after lease steal');
    $expect(!Store::checkpoint($lease, $progressPayload, false), 'metadata-only progress is fenced after lease steal');
    $expect(Db::name('prescription_ai_attempt')->where('task_id', $lease['id'])->where('attempt_no', 1)->value('status') === 'expired', 'expired attempt audited');
    putenv('ZYT_AI_TEST_DATABASE=' . $database);
    $pipes = [];
    $process = proc_open([PHP_BINARY, __FILE__, '--claim', 'qwen'], [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes);
    $childOutput = stream_get_contents($pipes[1]);
    $childError = stream_get_contents($pipes[2]);
    fclose($pipes[1]); fclose($pipes[2]);
    $exit = proc_close($process);
    $expect($exit === 0 && $childError === '' && json_decode($childOutput, true)['id'] === null, 'second PHP process cannot duplicate live lease');
    Db::transaction(static function () use ($rx): void {
        Db::name('tcm_prescription')->where('id', $rx['id'])->lock(true)->find();
        Db::name('tcm_prescription')->where('id', $rx['id'])->update(['void_status' => 1]);
        Store::invalidate((int) $rx['id'], 'voided');
    });
    $expect(!Store::complete($replacement, $output, $comparison), 'void during generation cannot publish');
    $expect(Db::name('prescription_ai_task')->where('id', $replacement['id'])->value('status') === 'cancelled', 'void cancels tasks');
    $editParams = ['id' => $blank['id'], 'herbs' => [['medicine_id' => 1, 'name' => '测试药材', 'dosage' => 10]],
        'prescription_date' => '2026-09-09', 'clinical_diagnosis' => 'fixture', 'usage_instruction' => 'fixture'];
    $expect(\app\adminapi\logic\tcm\PrescriptionLogic::edit($editParams, 1), 'real blank-to-manual hook: ' . \app\adminapi\logic\tcm\PrescriptionLogic::getError());
    $blankBatch = Db::name('prescription_ai_batch')->where('prescription_id', $blank['id'])->find();
    $expect($blankBatch['trigger_type'] === 'blank_to_manual' && (int) $blankBatch['patient_id'] === 100, 'blank hook carries authoritative binding');
    $expect(\app\adminapi\logic\tcm\PrescriptionLogic::edit($editParams, 1), 'same edit can replay');
    $expect((int) Db::name('prescription_ai_batch')->where('prescription_id', $blank['id'])->count() === 1, 'same edit does not generate twice');
    $expect(\app\adminapi\logic\tcm\PrescriptionLogic::void((int) $blank['id'], 1, 'Test'), 'real void hook succeeds');
    $expect(Db::name('prescription_ai_batch')->where('id', $blankBatch['id'])->value('validity') === 'voided', 'real void hook invalidates');
    $addParams = ['request_key' => 'add-request-1234567890123456', 'diagnosis_id' => 0, 'patient_id' => 100,
        'patient_name' => 'fixture', 'gender' => 1, 'age' => 50, 'clinical_diagnosis' => 'fixture',
        'herbs' => [['medicine_id' => 1, 'name' => '测试药材', 'dosage' => 10]], 'doctor_signature' => 'fixture'];
    $added = \app\adminapi\logic\tcm\PrescriptionLogic::add($addParams, 1, $root);
    $expect($added !== null, 'real direct-manual save hook: ' . \app\adminapi\logic\tcm\PrescriptionLogic::getError());
    $expect(\app\adminapi\logic\tcm\PrescriptionLogic::add($addParams, 1, $root) === $added, 'real direct save request replay returns same id');
    $expect(Db::name('prescription_ai_batch')->where('prescription_id', $added)->value('error_code') === 'PATIENT_BINDING_REQUIRED', 'unbound direct prescription has explicit blocked analysis');
    $pdo->exec("CREATE TRIGGER test_outbox_failure BEFORE INSERT ON zyt_prescription_ai_batch FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='test fixture failure'");
    $prescriptionsBefore = (int) Db::name('tcm_prescription')->count();
    $addParams['request_key'] = 'rollback-request-123456789012';
    $expect(\app\adminapi\logic\tcm\PrescriptionLogic::add($addParams, 1, $root) === null, 'outbox storage failure aborts real save');
    $expect((int) Db::name('tcm_prescription')->count() === $prescriptionsBefore, 'real save and request reservation roll back with outbox');
    $expect(!str_contains(\app\adminapi\logic\tcm\PrescriptionLogic::getError(), 'SQLSTATE'), 'outbox error text cannot expose SQL');
    $pdo->exec('DROP TRIGGER test_outbox_failure');
    if (defined('PRESCRIPTION_AI_PIPELINE_FIXTURE')) {
        $pipelineRx = $fixture();
        $pipelineBatch = $save($pipelineRx);
        $worker = new \app\common\service\prescriptionai\PrescriptionAiWorker();
        $expect($worker->prepareOne(), 'real coordinator freezes fixture evidence');
        $comparingObservations = [];
        $captureComparing = !$legacyProgressSchema;
        Db::listen(static function (string $sql) use (&$comparingObservations, &$captureComparing, $pipelineBatch): void {
            if (!$captureComparing || !str_starts_with($sql, 'UPDATE') || !str_contains($sql, 'progress_json') || !str_contains($sql, 'comparing')) {
                return;
            }
            foreach (Db::name('prescription_ai_task')->where('batch_id', $pipelineBatch)->where('status', 'running')->select()->toArray() as $row) {
                if ((json_decode($row['progress_json'] ?? '{}', true)['stage'] ?? '') !== 'comparing') { continue; }
                $cache = (new Cipher())->decrypt($row['progress_cipher'], 'progress:' . $row['id']);
                $comparingObservations[] = ['status' => $row['status'], 'cache_stage' => $cache['stage'] ?? '',
                    'has_result' => (bool) Db::name('prescription_ai_result')->where('batch_id', $pipelineBatch)->where('model_key', $row['model_key'])->count()];
            }
        });
        $expect($worker->runOne('qwen'), 'first real model worker');
        Db::name('doctor_medicine')->insert(['id' => 2, 'name' => 'new catalog fixture', 'unit' => 'g', 'status' => 1]);
        $expect($worker->runOne('openai'), 'second real model worker');
        $captureComparing = false;
        if (!$legacyProgressSchema) {
            $expect(count($comparingObservations) === 2, 'each worker publishes comparing before it persists a result');
            foreach ($comparingObservations as $observation) {
                $expect($observation === ['status' => 'running', 'cache_stage' => 'fixture', 'has_result' => false],
                    'comparison retains the encrypted checkpoint and never announces task completion before the result transaction');
            }
        }
        $pipelineStatus = Api::statuses([$pipelineRx['id']], 1, $root)['items'][0];
        $expect($pipelineStatus['models']['qwen']['score'] === 100.0, 'raw DB JSON prescription reaches real comparator');
        $expect($pipelineStatus['models']['openai']['score'] === 100.0, 'second model score persisted');
        $inputs = \app\common\service\prescriptionai\PrescriptionAiGenerator::$inputs;
        $expect($inputs['qwen']['source_hash'] === $inputs['openai']['source_hash'] && $inputs['qwen']['source'] === $inputs['openai']['source'], 'same frozen evidence supplied to both');
        $expect($inputs['qwen']['dictionary_version'] === $inputs['openai']['dictionary_version']
            && count($inputs['openai']['_comparison_catalog']) === 1, 'catalog changes between branches do not alter frozen dictionary');
        $expect(\app\common\service\prescriptionai\PrescriptionAiContext::$builds === 1, 'evidence read once per batch');
        $revokedRx = $fixture();
        $revokedBatch = $save($revokedRx);
        $worker->prepareOne();
        \app\common\service\prescriptionai\PrescriptionAiGenerator::$revokeDuringCall = true;
        $worker->runOne('qwen');
        $expect((int) Db::name('prescription_ai_result')->where('batch_id', $revokedBatch)->count() === 0, 'source permission revoked during call prevents publication');
        \app\common\service\prescriptionai\PrescriptionAiContext::$allowed = true;
    }
    $before = (int) Db::name('prescription_ai_batch')->count();
    $config->set(['enabled' => false], 'prescription_analysis');
    $expect($save($fixture()) === null && (int) Db::name('prescription_ai_batch')->count() === $before, 'feature disabled makes no outbox writes');
    $expect(Api::statuses([1], 1, $root) === ['enabled' => false, 'items' => []], 'disabled list has graceful compatibility');
    echo "Prescription AI queue: {$checks} checks passed\n";
} finally {
    $pdo->exec("DROP DATABASE `{$database}`");
}