Files
2026-09-10 15:19:17 +08:00

415 lines
24 KiB
PHP

<?php
declare(strict_types=1);
namespace app\common\service\prescriptionai;
use DomainException;
use think\facade\Db;
/** Database-backed outbox, immutable result storage and fenced work leases. */
final class PrescriptionAiStore
{
/** Cached per connection for rolling deployment before the additive migration arrives. */
public static function supportsProgress(): bool
{
static $columns = [];
$query = Db::name('prescription_ai_task');
$connection = $query->getConnection();
$key = spl_object_id($connection) . ':' . $query->getConfig('database');
if (!array_key_exists($key, $columns)) {
$columns[$key] = isset($query->getFields()['progress_json']);
}
return $columns[$key];
}
public static function enabled(): bool
{
return (bool) config('prescription_analysis.enabled', false);
}
public static function recordSaved(array $rx, int $actorId, ?array $info = null, array $options = []): ?int
{
if (!self::enabled() || !PrescriptionAiPolicy::isManual($rx)) {
return null;
}
$info = $info ?? PrescriptionAiAccess::actor($actorId);
if (!$info || !PrescriptionAiAccess::canGenerate($actorId, $info)) {
return null;
}
// Caller holds the prescription mutation transaction; failure must propagate.
return Db::transaction(static function () use ($rx, $actorId, $options): ?int {
$id = (int) $rx['id'];
$subject = Db::name('prescription_ai_subject')->where('prescription_id', $id)->lock(true)->find();
$hash = PrescriptionAiPolicy::fingerprint($rx);
if ($subject && hash_equals((string) $subject['clinical_hash'], $hash)
&& empty($options['restored'])) {
return (int) $subject['latest_batch_id'];
}
$revision = $subject ? (int) $subject['revision'] + 1 : 1;
$now = time();
if (!$subject) {
Db::name('prescription_ai_subject')->insert([
'prescription_id' => $id, 'revision' => $revision, 'clinical_hash' => $hash,
'first_batch_id' => 0, 'latest_batch_id' => 0, 'updated_at' => $now,
]);
} else {
self::invalidate($id, 'prescription_changed');
Db::name('prescription_ai_subject')->where('prescription_id', $id)->update([
'revision' => $revision, 'clinical_hash' => $hash, 'updated_at' => $now,
]);
}
$trigger = $subject ? 'clinical_change' : (string) ($options['trigger'] ?? 'first_manual');
if ($subject && !config('prescription_analysis.auto_refresh_prescription', true)) {
return null;
}
return self::insertBatch($rx, $revision, $actorId, $trigger,
hash('sha256', "saved:{$id}:{$revision}"), $options);
});
}
public static function enqueue(array $rx, int $actorId, string $trigger, string $eventKey, array $options = []): int
{
return Db::transaction(static function () use ($rx, $actorId, $trigger, $eventKey, $options): int {
$rx = Db::name('tcm_prescription')->where('id', (int) $rx['id'])->lock(true)->find();
if (!self::enabled() || !$rx || !PrescriptionAiPolicy::isManual($rx)) {
throw new DomainException('该处方不能安排分析');
}
$subject = Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->lock(true)->find();
if (!$subject || !hash_equals($subject['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
$info = PrescriptionAiAccess::actor($actorId);
$id = self::recordSaved($rx, $actorId, $info, ['trigger' => $trigger] + $options);
if (!$id) {
throw new DomainException('无法为该处方安排分析');
}
return $id;
}
$existing = Db::name('prescription_ai_batch')->where('event_key', $eventKey)->value('id');
if ($existing) {
return (int) $existing;
}
$latest = Db::name('prescription_ai_batch')->where('id', (int) $subject['latest_batch_id'])->find();
if ($trigger === 'manual_refresh' && $latest && $latest['validity'] === 'current'
&& in_array($latest['status'], ['preparing', 'waiting_sources', 'queued', 'running'], true)) {
return (int) $latest['id'];
}
self::invalidate((int) $rx['id'], 'source_updated');
return self::insertBatch($rx, (int) $subject['revision'], $actorId, $trigger, $eventKey, $options);
});
}
private static function insertBatch(array $rx, int $revision, int $actorId, string $trigger, string $key, array $options): int
{
$now = time();
$diagnosisId = (int) ($rx['diagnosis_id'] ?? 0);
$patientId = $diagnosisId > 0 ? (int) Db::name('tcm_diagnosis')->where('id', $diagnosisId)
->whereNull('delete_time')->value('patient_id') : 0;
$bindingValid = $patientId > 0 && ((int) ($rx['patient_id'] ?? 0) === 0 || (int) $rx['patient_id'] === $patientId);
$aiAssisted = array_key_exists('ai_assisted', $options)
? (in_array($options['ai_assisted'], [true, 1, '1'], true) ? 'yes'
: (in_array($options['ai_assisted'], [false, 0, '0'], true) ? 'no' : 'unknown')) : 'unknown';
$id = (int) Db::name('prescription_ai_batch')->insertGetId([
'event_key' => $key, 'prescription_id' => (int) $rx['id'], 'prescription_revision' => $revision,
'clinical_hash' => PrescriptionAiPolicy::fingerprint($rx), 'diagnosis_id' => $diagnosisId,
'patient_id' => $patientId, 'doctor_id' => (int) ($rx['creator_id'] ?? 0), 'actor_id' => $actorId,
'trigger_type' => $trigger, 'reason' => mb_substr((string) ($options['reason'] ?? ''), 0, 500),
'ai_assisted' => $aiAssisted, 'status' => $bindingValid ? 'preparing' : 'blocked',
'validity' => 'current', 'comparison_type' => 'latest_context', 'baseline_eligible' => 0,
'prescription_cipher' => (new PrescriptionAiCipher())->encrypt($rx, 'prescription'),
'decision_at' => $now, 'wait_until' => $now + (int) config('prescription_analysis.transcript_wait_seconds', 300),
'next_run_at' => $now + ($trigger === 'clinical_change' ? (int) config('prescription_analysis.debounce_seconds', 60) : 0),
'error_code' => $bindingValid ? '' : 'PATIENT_BINDING_REQUIRED', 'created_at' => $now, 'updated_at' => $now,
]);
$subject = Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->find();
Db::name('prescription_ai_subject')->where('prescription_id', (int) $rx['id'])->update([
'latest_batch_id' => $id, 'first_batch_id' => (int) ($subject['first_batch_id'] ?? 0) ?: $id, 'updated_at' => $now,
]);
return $id;
}
public static function invalidate(int $rxId, string $validity): void
{
if (!self::enabled()) {
return;
}
$ids = Db::name('prescription_ai_batch')->where('prescription_id', $rxId)->where('validity', 'current')->column('id');
if ($ids === []) {
return;
}
$now = time();
Db::name('prescription_ai_batch')->whereIn('id', $ids)->update([
'validity' => $validity, 'lock_token' => '', 'lock_until' => 0, 'updated_at' => $now,
]);
Db::name('prescription_ai_batch')->whereIn('id', $ids)
->whereIn('status', ['preparing', 'waiting_sources', 'retry_wait', 'queued', 'running'])->update(['status' => 'cancelled']);
Db::name('prescription_ai_task')->whereIn('batch_id', $ids)->whereIn('status', PrescriptionAiPolicy::ACTIVE_TASKS)
->update(['status' => 'cancelled', 'lock_token' => '', 'lock_until' => 0, 'updated_at' => $now]);
$taskIds = Db::name('prescription_ai_task')->whereIn('batch_id', $ids)->column('id');
if ($taskIds !== []) {
Db::name('prescription_ai_attempt')->whereIn('task_id', $taskIds)->where('status', 'running')
->update(['status' => 'cancelled', 'finished_at' => $now]);
}
}
public static function claimBatch(): ?array
{
$now = time();
$ids = Db::name('prescription_ai_batch')->where('validity', 'current')
->whereIn('status', ['preparing', 'waiting_sources', 'retry_wait'])->where('next_run_at', '<=', $now)
->where('lock_until', '<=', $now)->order('id')->limit(20)->column('id');
foreach ($ids as $id) {
$row = Db::transaction(static function () use ($id, $now): ?array {
$row = Db::name('prescription_ai_batch')->where('id', $id)->lock(true)->find();
if (!$row || $row['validity'] !== 'current' || (int) $row['lock_until'] > $now
|| !in_array($row['status'], ['preparing', 'waiting_sources', 'retry_wait'], true)
|| (int) $row['next_run_at'] > $now) {
return null;
}
$row['lock_token'] = bin2hex(random_bytes(16));
$row['lock_until'] = $now + (int) config('prescription_analysis.lease_seconds', 600);
Db::name('prescription_ai_batch')->where('id', $id)->update([
'lock_token' => $row['lock_token'], 'lock_until' => $row['lock_until'], 'updated_at' => $now,
]);
return $row;
});
if ($row) {
return $row;
}
}
return null;
}
public static function finishPreparation(array $batch, array $context): bool
{
return Db::transaction(static function () use ($batch, $context): bool {
$current = Db::name('prescription_ai_batch')->where('id', $batch['id'])->lock(true)->find();
if (!self::owns($current, $batch)) {
return false;
}
$now = time();
if (!empty($context['wait_for_transcript']) && $now < (int) $current['wait_until']) {
self::releasePreparation($batch, ['status' => 'waiting_sources', 'next_run_at' => $now + 15]);
return true;
}
$exclusions = (array) ($context['baseline_exclusion_reasons'] ?? []);
if ($current['ai_assisted'] !== 'no') {
$exclusions[] = $current['ai_assisted'] === 'yes' ? 'ai_assisted' : 'assistance_unknown';
}
if ($current['trigger_type'] !== 'first_manual' && $current['trigger_type'] !== 'blank_to_manual') {
$exclusions[] = 'not_original_submission';
}
$eligible = !empty($context['baseline_eligible']) && $exclusions === [];
Db::name('prescription_ai_batch')->where('id', $batch['id'])->update([
'context_cipher' => (new PrescriptionAiCipher())->encrypt($context, 'context'),
'access_cipher' => (new PrescriptionAiCipher())->encrypt(
['source_access_manifest' => $context['source_access_manifest'] ?? []], 'access:' . $batch['id']),
'source_hash' => (string) ($context['source_hash'] ?? ''),
'source_diagnosis_ids_json' => PrescriptionAiPolicy::canonical($context['source_diagnosis_ids'] ?? []),
'source_summary_json' => PrescriptionAiPolicy::canonical($context['source_summary'] ?? []),
'missing_json' => PrescriptionAiPolicy::canonical($context['missing'] ?? []),
'comparison_type' => (string) ($context['comparison_type'] ?? 'latest_context'),
'baseline_eligible' => (int) $eligible,
'baseline_exclusions_json' => PrescriptionAiPolicy::canonical(array_values(array_unique($exclusions))),
'cutoff_at' => (int) ($context['cutoff_at'] ?? $now), 'status' => 'queued',
'coverage_status' => empty($context['missing']) ? 'pending' : 'partial',
'lock_token' => '', 'lock_until' => 0, 'error_code' => '', 'updated_at' => $now,
]);
foreach (PrescriptionAiPolicy::MODELS as $model) {
Db::name('prescription_ai_task')->insert([
'batch_id' => $batch['id'], 'model_key' => $model, 'status' => 'queued',
'next_run_at' => $now, 'updated_at' => $now,
]);
}
return true;
});
}
public static function releasePreparation(array $batch, array $values): bool
{
return (bool) Db::name('prescription_ai_batch')->where('id', $batch['id'])
->where('lock_token', $batch['lock_token'])->where('validity', 'current')->update($values + [
'lock_token' => '', 'lock_until' => 0, 'updated_at' => time(),
]);
}
public static function claimTask(string $model): ?array
{
if (!in_array($model, PrescriptionAiPolicy::MODELS, true)) {
throw new DomainException('未知模型');
}
$now = time();
$ids = Db::name('prescription_ai_task')->where('model_key', $model)->whereIn('status', PrescriptionAiPolicy::ACTIVE_TASKS)
->where('next_run_at', '<=', $now)->where('lock_until', '<=', $now)->order('id')->limit(30)->column('id');
foreach ($ids as $id) {
$result = Db::transaction(static function () use ($id, $model, $now): ?array {
// One locked daily model counter serializes concurrency and budget reservations.
$key = date('Y-m-d', $now) . ':' . $model;
Db::name('prescription_ai_limit')->extra('IGNORE')->insert(['limit_key' => $key, 'used_count' => 0, 'updated_at' => $now]);
$budget = Db::name('prescription_ai_limit')->where('limit_key', $key)->lock(true)->find();
$batchId = Db::name('prescription_ai_task')->where('id', $id)->value('batch_id');
// All mutations lock batch before task, matching invalidation and completion.
$batch = $batchId ? Db::name('prescription_ai_batch')->where('id', $batchId)->lock(true)->find() : null;
$row = Db::name('prescription_ai_task')->where('id', $id)->lock(true)->find();
if (!$row || !in_array($row['status'], PrescriptionAiPolicy::ACTIVE_TASKS, true)
|| (int) $row['lock_until'] > $now || (int) $row['next_run_at'] > $now) {
return null;
}
if (!$batch || $batch['validity'] !== 'current') {
Db::name('prescription_ai_task')->where('id', $id)->update(['status' => 'cancelled', 'updated_at' => $now]);
return null;
}
$limit = max(1, (int) config('prescription_analysis.max_attempts', 3));
if ($row['status'] === 'running') {
Db::name('prescription_ai_attempt')->where('task_id', $id)->where('status', 'running')
->update(['status' => 'expired', 'error_code' => 'LEASE_EXPIRED', 'finished_at' => $now]);
}
if ((int) $row['attempts'] >= $limit) {
Db::name('prescription_ai_task')->where('id', $id)->update([
'status' => 'failed', 'error_code' => 'LEASE_EXPIRED', 'finished_at' => $now, 'updated_at' => $now,
]);
self::refreshBatch((int) $row['batch_id']);
return null;
}
if ((int) $budget['used_count'] >= max(1, (int) config('prescription_analysis.daily_model_tasks', 200))) {
Db::name('prescription_ai_task')->where('id', $id)->update([
'status' => 'retry_wait', 'error_code' => 'BUDGET_PAUSED',
'next_run_at' => strtotime('tomorrow', $now), 'updated_at' => $now,
]);
return null;
}
$running = Db::name('prescription_ai_task')->where('model_key', $model)->where('status', 'running')
->where('lock_until', '>', $now)->count();
if ($running >= max(1, (int) config('prescription_analysis.max_parallel_per_model', 1))) {
return null;
}
$row['attempts'] = (int) $row['attempts'] + 1;
$row['total_attempts'] = (int) $row['total_attempts'] + 1;
$row['status'] = 'running';
$row['lock_token'] = bin2hex(random_bytes(16));
$row['lock_until'] = $now + (int) config('prescription_analysis.lease_seconds', 600);
$row['started_at'] = $now;
$row['finished_at'] = 0;
$public = self::supportsProgress() ? ['progress_json' => PrescriptionAiPolicy::canonical(
PrescriptionAiProgress::advance([], 'preparing', 'running', null, null, $now))] : [];
Db::name('prescription_ai_task')->where('id', $id)->update(array_intersect_key($row, array_flip([
'attempts', 'total_attempts', 'status', 'lock_token', 'lock_until', 'started_at', 'finished_at',
])) + $public + ['error_code' => '', 'updated_at' => $now]);
Db::name('prescription_ai_attempt')->insert([
'task_id' => $id, 'attempt_no' => $row['total_attempts'], 'status' => 'running', 'started_at' => $now,
]);
Db::name('prescription_ai_limit')->where('limit_key', $key)->inc('used_count')->update(['updated_at' => $now]);
Db::name('prescription_ai_batch')->where('id', $row['batch_id'])->update(['status' => 'running', 'updated_at' => $now]);
return $row;
});
if ($result) {
return $result;
}
}
return null;
}
public static function checkpoint(array $task, array $progress, bool $persistCache = true): bool
{
$now = time();
$values = ['lock_until' => $now + (int) config('prescription_analysis.lease_seconds', 600), 'updated_at' => $now];
if (self::supportsProgress()) {
$values['progress_json'] = PrescriptionAiPolicy::canonical(PrescriptionAiProgress::sanitize($progress['public'] ?? null));
}
// Stage notifications do not rewrite the growing encrypted model cache.
if ($persistCache) {
$values['progress_cipher'] = (new PrescriptionAiCipher())->encrypt($progress, 'progress:' . $task['id']);
}
$owned = static fn () => Db::name('prescription_ai_task')->where('id', $task['id'])->where('status', 'running')
->where('lock_token', $task['lock_token'])->where('lock_until', '>', time());
$changed = $owned()->update($values);
// Identical metadata twice in one second (also old-schema heartbeats) can be a no-op.
return $changed > 0 || $owned()->count() > 0;
}
public static function complete(array $task, array $output, array $comparison): bool
{
return Db::transaction(static function () use ($task, $output, $comparison): bool {
$rxId = Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->value('prescription_id');
$rx = Db::name('tcm_prescription')->where('id', $rxId)->lock(true)->find();
$batch = Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->lock(true)->find();
$row = Db::name('prescription_ai_task')->where('id', $task['id'])->lock(true)->find();
if (!self::owns($row, $task) || $row['status'] !== 'running') {
return false;
}
if (!$batch || $batch['validity'] !== 'current' || !$rx || !PrescriptionAiPolicy::isManual($rx)
|| !hash_equals($batch['clinical_hash'], PrescriptionAiPolicy::fingerprint($rx))) {
return false;
}
$coverage = $output['coverage'] ?? [];
$coverageStatus = is_array($coverage) ? (string) ($coverage['status'] ?? 'partial') : 'partial';
$coverageStatus = in_array($coverageStatus, ['complete', 'full'], true) ? 'complete' : 'partial';
$body = ['report' => $output['report'] ?? [], 'candidate' => $output['candidate'] ?? null,
'comparison' => $comparison, 'coverage' => $coverage, 'usage' => $output['usage'] ?? []];
$now = time();
$resultId = (int) Db::name('prescription_ai_result')->insertGetId([
'batch_id' => $task['batch_id'], 'model_key' => $task['model_key'],
'body_cipher' => (new PrescriptionAiCipher())->encrypt($body, 'result:' . $task['batch_id'] . ':' . $task['model_key']),
'score' => $comparison['score'] ?? null, 'herb_score' => $comparison['herb_score'] ?? null,
'comparison_status' => (string) ($comparison['status'] ?? 'not_comparable'),
'comparison_reason_code' => (string) ($comparison['reason_code'] ?? ''), 'coverage_status' => $coverageStatus,
'model_name' => mb_substr((string) ($output['model_name'] ?? ''), 0, 100),
'prompt_version' => mb_substr((string) ($output['prompt_version'] ?? ''), 0, 100),
'algorithm_version' => mb_substr((string) ($comparison['algorithm_version'] ?? ''), 0, 100), 'generated_at' => $now,
'dictionary_version' => mb_substr((string) ($comparison['dictionary_version'] ?? ''), 0, 100),
]);
$public = self::supportsProgress() ? ['progress_json' => PrescriptionAiPolicy::canonical(
PrescriptionAiProgress::advance([], 'completed', 'completed', null, null, $now))] : [];
Db::name('prescription_ai_task')->where('id', $task['id'])->update($public + [
'status' => 'success', 'result_id' => $resultId, 'finished_at' => $now, 'updated_at' => $now,
'lock_token' => '', 'lock_until' => 0, 'error_code' => '', 'progress_cipher' => null,
]);
Db::name('prescription_ai_attempt')->where('task_id', $task['id'])->where('attempt_no', $row['total_attempts'])
->update(['status' => 'success', 'finished_at' => $now]);
self::refreshBatch((int) $task['batch_id']);
return true;
});
}
public static function fail(array $task, string $code, bool $retryable): void
{
Db::transaction(static function () use ($task, $code, $retryable): void {
Db::name('prescription_ai_batch')->where('id', $task['batch_id'])->lock(true)->find();
$current = Db::name('prescription_ai_task')->where('id', $task['id'])->lock(true)->find();
if (!self::owns($current, $task) || $current['status'] !== 'running') {
return;
}
$now = time();
$code = preg_match('/^[A-Z0-9_]{1,64}$/D', $code) ? $code : 'INTERNAL_ERROR';
$next = PrescriptionAiPolicy::retryAt((int) $task['attempts'], $now, $retryable,
max(1, (int) config('prescription_analysis.max_attempts', 3)));
Db::name('prescription_ai_task')->where('id', $task['id'])->where('lock_token', $task['lock_token'])
->where('status', 'running')->update([
'status' => $next === null ? 'failed' : 'retry_wait', 'next_run_at' => $next ?? $now,
'error_code' => $code, 'lock_token' => '', 'lock_until' => 0, 'finished_at' => $now, 'updated_at' => $now,
]);
self::refreshBatch((int) $task['batch_id']);
Db::name('prescription_ai_attempt')->where('task_id', $task['id'])->where('attempt_no', $current['total_attempts'])
->update(['status' => 'failed', 'error_code' => $code, 'finished_at' => $now]);
});
}
public static function refreshBatch(int $batchId): void
{
$states = Db::name('prescription_ai_task')->where('batch_id', $batchId)->column('status');
$coverage = Db::name('prescription_ai_result')->where('batch_id', $batchId)->column('coverage_status');
Db::name('prescription_ai_batch')->where('id', $batchId)->where('validity', 'current')->update([
'status' => PrescriptionAiPolicy::aggregate($states),
'coverage_status' => count($coverage) === 2 && $coverage === ['complete', 'complete'] ? 'complete' : 'partial',
'updated_at' => time(),
]);
}
private static function owns(?array $current, array $claim): bool
{
return $current && !empty($claim['lock_token'])
&& hash_equals((string) $current['lock_token'], (string) $claim['lock_token'])
&& (int) $current['lock_until'] > time()
&& (!isset($current['validity']) || $current['validity'] === 'current');
}
}