<?php

declare(strict_types=1);

/** 真实业务链路与多进程并发，仅允许目录指纹一致的本项目隔离 MySQL。 */
$port = (int) getenv('ASSISTANT_SERVICE_TEST_PORT');
$expectedDirectory = str_replace('\\', '/', (string) getenv('ASSISTANT_SERVICE_TEST_DATADIR'));
if (getenv('ASSISTANT_SERVICE_TEST_ALLOW_DISPOSABLE') !== '1' || $port !== 15339
    || strtolower(rtrim($expectedDirectory, '/')) !== 'd:/web/zyt/.spool/assistant-service-mysql') {
    throw new RuntimeException('必须显式指定本项目隔离实例端口 15339 和目录指纹');
}
$pdo = new PDO("mysql:host=127.0.0.1;port={$port};charset=utf8mb4", 'root', '', [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
$actual = str_replace('\\', '/', (string) $pdo->query('SELECT @@datadir')->fetchColumn());
if (strtolower(rtrim($actual, '/')) !== strtolower(rtrim($expectedDirectory, '/'))) {
    throw new RuntimeException('实际数据目录不匹配，拒绝运行');
}
$child = ($argv[1] ?? '') === '--child';
$database = $child ? (string) ($argv[2] ?? '') : 'assistant_service_flow_' . bin2hex(random_bytes(6));
if (!preg_match('/^assistant_service_flow_[a-f0-9]{12}$/D', $database)) { throw new RuntimeException('非法测试数据库名称'); }

require dirname(__DIR__) . '/vendor/autoload.php';
require dirname(__DIR__) . '/vendor/topthink/framework/src/helper.php';

use app\adminapi\logic\tcm\AssistantServiceLogic as Logic;
use app\adminapi\validate\tcm\AssistantServiceValidate as Validate;
use app\common\service\assistantservice\AssistantServiceAccess as Access;
use app\common\service\assistantservice\AssistantServiceContext as Context;
use app\common\service\assistantservice\AssistantServiceWorker as Worker;
use think\Container;
use think\facade\Db;

new think\App(); // 不 initialize，不加载 .env 与任何业务连接。
set_exception_handler(static function (Throwable $e): void {
    fwrite(STDERR, $e->getMessage() . PHP_EOL . $e->getTraceAsString() . PHP_EOL); exit(1);
});
$manager = new think\DbManager();
$manager->setConfig(['default' => 'isolated', 'connections' => ['isolated' => [
    '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);
think\Model::setDb($manager);
$config = new think\Config();
$config->set(['enabled' => true, 'profile' => 'qwen', 'timeout' => 120, 'lease_seconds' => 300, 'daily_job_limit' => 100,
    'max_source_rows' => 200, 'max_source_chars' => 30000, 'max_manual_chars' => 10000, 'max_knowledge_rows' => 40], 'assistant_service');
$config->set(['enable' => true, 'base_url' => 'https://example.invalid', 'models' => [
    'qwen' => ['name' => 'synthetic', 'api_key' => 'synthetic-not-used']]], 'prescription_ai');
$config->set(['data_scope' => ['enabled' => true, 'no_dept_fallback_self' => true]], 'project');
Container::getInstance()->instance('config', $config);
$call = static fn (string $action, array $p = [], int $actor = 11): array => Logic::$action(Validate::parameters($action, $p), $actor);

if ($child) {
    $pdo->exec("USE `{$database}`");
    if ((string) $pdo->query('SELECT marker FROM flow_test_guard')->fetchColumn() !== $database) { throw new RuntimeException('缺少父进程测试标记'); }
    Db::execute('SET SESSION innodb_lock_wait_timeout=5');
    $action = (string) ($argv[3] ?? '');
    if (!in_array($action, ['analyze', 'adoptAction'], true)) { throw new RuntimeException('未知子进程任务'); }
    $params = json_decode(base64_decode((string) ($argv[4] ?? ''), true), true, 16, JSON_THROW_ON_ERROR);
    $at = (float) ($argv[5] ?? 0);
    if ($at > microtime(true) + 5) { throw new RuntimeException('非法测试屏障'); }
    while (microtime(true) < $at) { usleep(10000); }
    $response = $call($action, $params);
    echo json_encode(['id' => $response['id'], 'todo_id' => $response['todo_id'] ?? null], JSON_THROW_ON_ERROR);
    exit(0);
}

$checks = 0;
$ok = static function (bool $condition, string $label) use (&$checks): void { if (!$condition) { throw new RuntimeException($label); } $checks++; };
$deny = static function (callable $call, string $label) use ($ok): void {
    try { $call(); } catch (DomainException $e) { $ok(true, $label); return; } $ok(false, $label);
};
$runSql = static function (string $path) use ($pdo): void {
    $sql = preg_replace('/^--.*$/m', '', (string) file_get_contents($path));
    foreach (explode(';', (string) $sql) as $statement) { if (trim($statement) !== '') { $cursor = $pdo->query($statement); $cursor->closeCursor(); } }
};
$parallel = static function (string $action, array $params) use ($database, $ok): array {
    $processes = []; $startAt = microtime(true) + 0.8;
    for ($i = 0; $i < 2; $i++) {
        $process = proc_open([PHP_BINARY, __FILE__, '--child', $database, $action,
            base64_encode(json_encode($params, JSON_THROW_ON_ERROR)), (string) $startAt],
            [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes, __DIR__, null,
            ['bypass_shell' => true, 'create_new_console' => false]);
        if (!is_resource($process)) { throw new RuntimeException('无法启动本地并发测试子进程'); }
        fclose($pipes[0]); $processes[] = [$process, $pipes];
    }
    $results = []; $errors = [];
    // 必须回收两个子进程，再对结果断言，避免失败时数据库仍有写入者。
    foreach ($processes as [$process, $pipes]) {
        $out = stream_get_contents($pipes[1]); $error = stream_get_contents($pipes[2]);
        fclose($pipes[1]); fclose($pipes[2]); $exit = proc_close($process);
        $errors[] = [$exit, $error]; $results[] = json_decode($out, true);
    }
    foreach ($errors as [$exit, $error]) { $ok($exit === 0, 'concurrent child failed: ' . $error); }
    return $results;
};

$pdo->exec("CREATE DATABASE `{$database}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci");
try {
    $pdo->exec("USE `{$database}`");
    $pdo->exec('CREATE TABLE flow_test_guard(marker VARCHAR(100) PRIMARY KEY) ENGINE=InnoDB');
    $pdo->exec("INSERT INTO flow_test_guard VALUES('{$database}')");
    $schemas = [
        'admin' => 'id INT PRIMARY KEY,name VARCHAR(64),root INT DEFAULT 0,disable INT DEFAULT 0,delete_time INT NULL',
        'admin_role' => 'admin_id INT,role_id INT,PRIMARY KEY(admin_id,role_id)',
        'admin_dept' => 'admin_id INT,dept_id INT,PRIMARY KEY(admin_id,dept_id)',
        'dept' => 'id INT PRIMARY KEY,pid INT,status INT DEFAULT 1,delete_time INT NULL',
        'system_role' => 'id INT PRIMARY KEY,name VARCHAR(64),data_scope INT,delete_time INT NULL',
        'system_role_menu' => 'role_id INT,menu_id INT,PRIMARY KEY(role_id,menu_id)',
        'system_menu' => 'id INT PRIMARY KEY,perms VARCHAR(100),is_disable INT DEFAULT 0',
        'doctor_appointment' => 'id INT PRIMARY KEY,patient_id INT,doctor_id INT,status INT',
        'tcm_diagnosis' => 'id INT PRIMARY KEY,patient_id INT,assistant_id INT,patient_name VARCHAR(64),phone VARCHAR(32),status INT DEFAULT 1,delete_time INT NULL',
        'tcm_im_chat_message' => 'id INT PRIMARY KEY,patient_id INT,diagnosis_id INT,from_account VARCHAR(64),to_account VARCHAR(64),doctor_peer_account VARCHAR(64),msg_type VARCHAR(30),text TEXT,msg_time INT',
    ];
    foreach ($schemas as $name => $schema) { $pdo->exec('CREATE TABLE zyt_' . $name . ' (' . $schema . ') ENGINE=InnoDB'); }
    $runSql(dirname(__DIR__) . '/sql/1.9.20260505/add_tracking_note_table.sql');
    $runSql(dirname(__DIR__) . '/sql/1.9.20260505/add_tcm_diagnosis_todo.sql');
    $runSql(dirname(__DIR__) . '/sql/1.9.20261109/01_assistant_service_tables.sql');
    Db::execute('SET SESSION innodb_lock_wait_timeout=5');
    Db::execute('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
    $pdo->exec('SET SESSION innodb_lock_wait_timeout=5');
    Db::name('admin')->insertAll([['id' => 1, 'name' => 'root', 'root' => 1], ['id' => 11, 'name' => 'assistant', 'root' => 0],
        ['id' => 12, 'name' => 'other', 'root' => 0], ['id' => 13, 'name' => 'manager', 'root' => 0], ['id' => 14, 'name' => 'doctor', 'root' => 0]]);
    Db::name('system_role')->insertAll([['id' => 1, 'name' => '医生', 'data_scope' => 1], ['id' => 2, 'name' => '医助', 'data_scope' => 4],
        ['id' => 3, 'name' => '经理', 'data_scope' => 3], ['id' => 9, 'name' => '功能角色', 'data_scope' => 4]]);
    Db::name('admin_role')->insertAll([['admin_id' => 11, 'role_id' => 2], ['admin_id' => 11, 'role_id' => 9],
        ['admin_id' => 12, 'role_id' => 2], ['admin_id' => 13, 'role_id' => 3], ['admin_id' => 14, 'role_id' => 1]]);
    Db::name('dept')->insertAll([['id' => 10, 'pid' => 0], ['id' => 20, 'pid' => 0]]);
    Db::name('admin_dept')->insertAll([['admin_id' => 11, 'dept_id' => 10], ['admin_id' => 12, 'dept_id' => 20], ['admin_id' => 13, 'dept_id' => 10]]);
    foreach (Access::PERMISSIONS as $i => $permission) {
        Db::name('system_menu')->insert(['id' => $i + 1, 'perms' => 'tcm.assistantService/' . $permission]);
        foreach ([1, 2, 3, 9] as $role) { Db::name('system_role_menu')->insert(['role_id' => $role, 'menu_id' => $i + 1]); }
    }
    Db::name('tcm_diagnosis')->insertAll([['id' => 101, 'patient_id' => 201, 'assistant_id' => 11, 'patient_name' => '合成患者甲', 'phone' => '13800138000'],
        ['id' => 102, 'patient_id' => 202, 'assistant_id' => 12, 'patient_name' => '合成患者乙', 'phone' => '13900139000']]);
    Db::name('doctor_appointment')->insert(['id' => 1, 'patient_id' => 101, 'doctor_id' => 14, 'status' => 1]);
    $record = $call('record', ['diagnosis_id' => 101, 'content' => '患者希望核对周末预约时段。', 'record_type' => 'service_note', 'occurred_at' => time()]);
    $knowledge = $call('saveKnowledge', ['question' => '如何预约？', 'answer' => '请由医助核对当前可预约时段。', 'category' => '预约', 'source_note' => '合成流程'], 1);
    $knowledge = $call('publishKnowledge', $knowledge + ['status' => 'published'], 1);
    $ok($call('capabilities')['ready'], 'actual schema is ready');
    $ok($call('patients')['count'] === 1 && $call('patients', [], 13)['count'] === 1, 'MySQL MyPatient scope works');
    Db::name('system_role')->insert(['id' => 7, 'name' => '已删除主管', 'data_scope' => 1, 'delete_time' => 1]);
    Db::name('admin_role')->insert(['admin_id' => 11, 'role_id' => 7]);
    $ok($call('patients')['count'] === 1, 'deleted manager-role residue neither expands nor blocks current scope');
    $legacy = Db::name('tcm_diagnosis')->alias('d')->whereNull('d.delete_time')->where('d.status', 1);
    \app\adminapi\logic\firstvisit\MyPatientLogic::applyScope($legacy, 14, Access::actor(14));
    $ok((int) $legacy->count() === 1, 'legacy MyPatient default-call behavior preserved');

    $jobs = $parallel('analyze', ['diagnosis_id' => 101, 'scenario' => 'followup']);
    $ok($jobs[0]['id'] === $jobs[1]['id'] && Db::name('assistant_service_job')->count() === 1, 'parallel enqueue dedupes to one job');
    $jobId = $jobs[0]['id']; $invocations = 0;
    $generator = static function ($type, $scenario, $context, $knowledge, $question) use (&$invocations): array {
        $invocations++;
        $evidence = [$context['sources'][0]['id']];
        if ($type === 'reply') {
            return ['ok' => true, 'result' => ['reply' => '我会协助您核对可预约时段。', 'knowledge_refs' => [array_intersect_key($knowledge[0], ['id' => true, 'version' => true])],
                'evidence_ids' => $evidence, 'needs_confirmation' => true, 'reason' => '请人工核对时段后使用'], 'message_id' => 'synthetic-reply'];
        }
        $quality = [];
        foreach (['identity', 'needs', 'explanation', 'next_step', 'followup', 'escalation'] as $key) {
            $quality[] = ['key' => $key, 'label' => '人工复核', 'status' => 'insufficient', 'reason' => '服务记录不能归责', 'evidence_ids' => [], 'actor_admin_id' => 0];
        }
        return ['ok' => true, 'result' => ['summary' => [['text' => '患者希望核对预约时段', 'evidence_ids' => $evidence]],
            'needs' => [], 'unresolved' => [], 'tags' => [], 'suggested_actions' => [
                ['key' => 'confirm_slot', 'text' => '核对预约时段并回复患者', 'evidence_ids' => $evidence],
                ['key' => 'later_followup', 'text' => '确认后再次跟进', 'evidence_ids' => $evidence]], 'quality_checks' => $quality], 'message_id' => 'synthetic-analysis'];
    };
    $worker = new Worker();
    $ok($worker->runOnce($generator), 'worker claims API-created job');
    $ready = $call('job', ['id' => $jobId]);
    $ok($ready['status'] === 'ready' && $invocations === 1 && count($ready['result']['quality_checks']) === 6, 'real engine validates worker result');
    $ok(!$worker->runOnce($generator) && $invocations === 1, 'worker does not re-execute completed task');
    $reply = $call('draftReply', ['diagnosis_id' => 101, 'scenario' => 'followup', 'question' => '如何预约？']);
    $worker->runOnce($generator);
    $ok($call('job', ['id' => $reply['id']])['result']['knowledge_refs'][0]['version'] === $knowledge['version'], 'reply API-worker preserves published knowledge version');
    $review = $call('reviewQuality', ['job_id' => $jobId, 'item_key' => 'identity', 'decision' => 'not_applicable', 'version' => 1, 'note' => '合成复核'], 13);
    $ok($review['id'] > 0, 'supervisor review follows worker result');
    $adopt = ['job_id' => $jobId, 'action_key' => 'confirm_slot', 'version' => 1, 'due_at' => time() + 3600];
    $actions = $parallel('adoptAction', $adopt);
    $ok($actions[0] === $actions[1] && Db::name('assistant_service_action')->count() === 1 && Db::name('tcm_diagnosis_todo')->count() === 1,
        'parallel adoption creates exactly one action and one reminder');
    $call('completeAction', ['id' => $actions[0]['id'], 'state' => 'completed', 'note' => '已核对并回复']);
    $ok((int) Db::name('tcm_diagnosis_todo')->where('id', $actions[0]['todo_id'])->value('status') === 2, 'completion atomically cancels pending reminder');

    $pdo->exec("CREATE TRIGGER fail_action_insert BEFORE INSERT ON zyt_assistant_service_action FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='synthetic rollback'");
    $before = Db::name('tcm_diagnosis_todo')->count(); $failed = false;
    try { $call('adoptAction', array_replace($adopt, ['action_key' => 'later_followup'])); }
    catch (Throwable $e) { $failed = str_contains($e->getMessage(), 'synthetic rollback'); }
    $ok($failed && Db::name('tcm_diagnosis_todo')->count() === $before, 'sidecar insertion failure rolls back old reminder on InnoDB');
    $pdo->exec('DROP TRIGGER fail_action_insert');

    // 每次先在 ORM 连接建立 RR 一致性快照，再由第二连接提交撤权。
    Db::startTrans(); Db::name('admin_role')->select();
    $pdo->exec('DELETE FROM zyt_admin_role WHERE admin_id=11 AND role_id=2');
    $deny(fn () => Access::diagnosis(101, 11), 'role revocation beats existing RR snapshot'); Db::rollback();
    $pdo->exec('INSERT INTO zyt_admin_role VALUES(11,2)');
    Db::startTrans(); Db::name('system_role')->select();
    $pdo->exec('UPDATE zyt_system_role SET data_scope=4 WHERE id=3');
    $deny(fn () => Access::diagnosis(101, 13), 'role data scope change beats RR snapshot'); Db::rollback();
    $pdo->exec('UPDATE zyt_system_role SET data_scope=3 WHERE id=3');
    Db::startTrans(); Db::name('admin_dept')->select();
    $pdo->exec('UPDATE zyt_admin_dept SET dept_id=20 WHERE admin_id=11');
    $deny(fn () => Access::diagnosis(101, 13), 'team member department removal beats RR snapshot'); Db::rollback();
    $pdo->exec('UPDATE zyt_admin_dept SET dept_id=10 WHERE admin_id=11');
    Db::startTrans(); Db::name('admin_dept')->select();
    $pdo->exec('UPDATE zyt_admin_dept SET dept_id=20 WHERE admin_id=13');
    $deny(fn () => Access::diagnosis(101, 13), 'manager own department change beats RR snapshot'); Db::rollback();
    $pdo->exec('UPDATE zyt_admin_dept SET dept_id=10 WHERE admin_id=13');
    Db::startTrans(); Db::name('system_role_menu')->select();
    $pdo->exec('DELETE FROM zyt_system_role_menu WHERE role_id IN(2,9) AND menu_id=2');
    $deny(fn () => Access::diagnosis(101, 11, 'analyze'), 'menu revocation beats RR snapshot'); Db::rollback();
    $pdo->exec('INSERT INTO zyt_system_role_menu VALUES(2,2),(9,2)');
    Db::startTrans(); Db::name('admin')->select();
    $pdo->exec('UPDATE zyt_admin SET disable=1 WHERE id=11');
    $deny(fn () => Access::diagnosis(101, 11), 'actor disabled beats RR snapshot'); Db::rollback();
    $pdo->exec('UPDATE zyt_admin SET disable=0 WHERE id=11');
    Db::startTrans(); Db::name('tcm_diagnosis')->select();
    $pdo->exec('UPDATE zyt_tcm_diagnosis SET assistant_id=12 WHERE id=101');
    $deny(fn () => Access::diagnosis(101, 11), 'patient reassignment beats RR snapshot'); Db::rollback();
    $pdo->exec('UPDATE zyt_tcm_diagnosis SET assistant_id=11 WHERE id=101');
    Db::startTrans(); Db::name('doctor_appointment')->select();
    $pdo->exec('UPDATE zyt_doctor_appointment SET status=2 WHERE id=1');
    $deny(fn () => Access::diagnosis(101, 14), 'doctor appointment revocation beats RR snapshot'); Db::rollback();
    $pdo->exec('UPDATE zyt_doctor_appointment SET status=1 WHERE id=1');

    $beforeHash = Context::build(101, 11)['source_hash'];
    Db::startTrans(); Db::name('assistant_service_record')->select();
    $pdo->exec("UPDATE zyt_assistant_service_record SET content='患者改为核对下周预约时段。' WHERE id=" . $record['id']);
    $ok(Context::build(101, 11)['source_hash'] !== $beforeHash, 'context reads current source despite old RR snapshot');
    $deny(fn () => $call('adoptAction', array_replace($adopt, ['action_key' => 'later_followup'])), 'source change blocks adoption inside existing RR transaction');
    Db::rollback();

    // 生成期间修改来源：真实 worker 提交前复核，禁止旧证据落为 ready。
    $changed = $call('analyze', ['diagnosis_id' => 101, 'scenario' => 'revisit']);
    $worker->runOnce(static function (...$args) use ($generator, $pdo, $record): array {
        $answer = $generator(...$args);
        $pdo->exec("UPDATE zyt_assistant_service_record SET content='患者再次调整服务时间。' WHERE id=" . $record['id']);
        return $answer;
    });
    $ok(Db::name('assistant_service_job')->where('id', $changed['id'])->value('status') === 'stale', 'worker refuses source changed during generation');
    $fresh = $call('analyze', ['diagnosis_id' => 101, 'scenario' => 'initial']); $worker->runOnce($generator);
    Db::startTrans(); Db::name('assistant_service_knowledge')->select();
    $pdo->exec("UPDATE zyt_assistant_service_knowledge SET version=version+1,answer='合成更新预约流程' WHERE id=" . $knowledge['id']);
    $deny(fn () => $call('adoptAction', array_replace($adopt, ['job_id' => $fresh['id']])), 'knowledge version change beats existing RR snapshot'); Db::rollback();
    $ok($call('jobs', ['status' => 'ready'])['count'] === 0, 'ready filter excludes rows freshly invalidated by source or knowledge');
    $stale = $call('jobs', ['status' => 'stale', 'page_size' => 1]);
    $ok($stale['count'] === 4 && count($stale['lists']) === 1 && $stale['lists'][0]['status'] === 'stale', 'stale filter computes exact count before pagination');
    $retryParams = ['diagnosis_id' => 101, 'days' => 7, 'scenario' => 'initial'];
    $retryJob = $call('analyze', $retryParams);
    $worker->runOnce(static fn (): array => ['ok' => false, 'error_code' => 'CONFIG_INVALID']);
    $ok(Db::name('assistant_service_job')->where('id', $retryJob['id'])->value('status') === 'failed', 'known-not-generated configuration failure is terminal failed');
    $retried = $call('analyze', $retryParams); $worker->runOnce($generator);
    $ok($retried['id'] === $retryJob['id'] && $retried['version'] === 2 && $call('job', ['id' => $retryJob['id']])['status'] === 'ready',
        'explicit retry of known config failure completes same versioned job without duplicate row');
    Db::name('assistant_service_job')->where('id', $retryJob['id'])->update(['status' => 'needs_reconciliation', 'error_code' => 'UPSTREAM_UNCERTAIN']);
    $ok($call('analyze', $retryParams)['status'] === 'needs_reconciliation', 'unknown upstream state is never requeued');

    // 主管目标医助入口：使用真实 MySQL 表与 Worker，沿用显式目录指纹防护。
    $target = ['assistant_id' => 11, 'days' => 30];
    $ok(array_column($call('staff', [], 13)['lists'], 'id') === [11], 'manager staff list excludes other department and manager identity');
    $deny(fn () => $call('staffOverview', ['assistant_id' => 12, 'days' => 30], 13), 'staff overview rejects other department');
    Db::name('admin')->insert(['id' => 15, 'name' => 'second assistant', 'root' => 0]);
    Db::name('admin_role')->insert(['admin_id' => 15, 'role_id' => 2]);
    Db::name('admin_dept')->insert(['admin_id' => 15, 'dept_id' => 10]);
    Db::name('tcm_diagnosis')->insertAll([
        ['id' => 103, 'patient_id' => 201, 'assistant_id' => 15, 'patient_name' => '同患者第二诊单', 'phone' => ''],
        ['id' => 104, 'patient_id' => 204, 'assistant_id' => 11, 'patient_name' => '无来源诊单', 'phone' => ''],
    ]);
    foreach ([[1, 'doctor_11', '目标医助真实发言'], [2, 'doctor_15', '其他医助私有发言']] as [$id, $from, $text]) {
        Db::name('tcm_im_chat_message')->insert(['id' => $id, 'patient_id' => 201, 'diagnosis_id' => 101,
            'from_account' => $from, 'to_account' => 'patient_201', 'doctor_peer_account' => 'doctor_11', 'msg_type' => 'text', 'text' => $text, 'msg_time' => time()]);
    }
    $targetContext = $call('context', $target + ['diagnosis_id' => 101], 13);
    $ok($targetContext['subject_admin_id'] === 11 && !in_array('im:2', array_column($targetContext['sources'], 'id'), true), 'same patient target context excludes other assistant on MySQL');
    $ok($call('staffOverview', $target, 13)['metrics']['im_messages'] === 1, 'same message counted once across multiple diagnoses on MySQL');
    $before = (int) Db::name('assistant_service_job')->count();
    $deny(fn () => $call('analyzeStaff', $target + ['scenario' => 'followup', 'diagnosis_ids' => [101, 104]], 13), 'target batch missing-source failure');
    $ok((int) Db::name('assistant_service_job')->count() === $before, 'InnoDB target batch rolls back earlier nested enqueue');
    $batch = $call('analyzeStaff', $target + ['scenario' => 'followup', 'diagnosis_ids' => [101, 103]], 13);
    $again = $call('analyzeStaff', $target + ['scenario' => 'followup', 'diagnosis_ids' => [101, 103]], 13);
    $ok($batch['count'] === 2 && array_column($batch['jobs'], 'id') === array_column($again['jobs'], 'id'), 'target bulk queue deduplication on MySQL');
    $targetGenerator = static function (...$args) use ($generator, $ok): array {
        $ok($args[2]['subject_admin_id'] === 11 && !in_array('im:2', array_column($args[2]['sources'], 'id'), true), 'worker preserves target through real MySQL context');
        return $generator(...$args);
    };
    $worker->runOnce($targetGenerator); $worker->runOnce($targetGenerator);
    $targetJob = $batch['jobs'][0];
    $ok($call('job', ['id' => $targetJob['id']], 13)['status'] === 'ready', 'manager-created target task reaches ready');
    $targetAction = $call('adoptAction', ['job_id' => $targetJob['id'], 'action_key' => 'confirm_slot', 'version' => 1, 'due_at' => time() + 3600], 13);
    $ok((int) Db::name('assistant_service_action')->where('id', $targetAction['id'])->value('assignee_id') === 11
        && (int) Db::name('tcm_diagnosis_todo')->where('id', $targetAction['todo_id'])->value('creator_id') === 13,
        'target action assignee and actual reminder creator are separate on MySQL');
    $ok($call('actions', ['assistant_id' => 11], 13)['count'] === 2 && $call('actions', ['assistant_id' => 15], 13)['count'] === 0, 'execution filters by target across patients');

    echo 'ASSISTANT_SERVICE_MYSQL_FLOW assertions=' . $checks . ' PASS real_access_context_logic_worker=1 concurrent_processes=2 network_model_calls=0' . PHP_EOL;
} finally {
    try { Db::rollback(); } catch (Throwable $ignored) {}
    if (preg_match('/^assistant_service_flow_[a-f0-9]{12}$/D', $database)) { $pdo->exec("DROP DATABASE `{$database}`"); }
}