settings = $settings; $this->provider = $provider; $this->transport = $transport; } /** Caller supplies current authorization/lease + durable encrypted checkpoint CAS on EVERY callback. */ public function run(string $path, array $task, callable $heartbeat, array $checkpoint = []): array { if (($this->provider['driver'] ?? '') !== 'asr_then_llm') { throw new FollowupAudioException('CONFIG_INVALID'); } foreach (['asr', 'extraction'] as $stage) { if (!FollowupAudioProviderConfig::secureEndpoint($this->provider[$stage]['base_url'], $this->provider['allow_loopback_tunnel'])) { throw new FollowupAudioException('HTTPS_REQUIRED'); } } $snapshot = json_decode((string) ($task['upstream_ids_json'] ?? '{}'), true); if (!hash_equals($this->provider['fingerprint'], (string) ($snapshot['provider_fingerprint'] ?? ''))) { throw new FollowupAudioException('PROVIDER_CONFIGURATION_CHANGED'); } self::beat($heartbeat, []); FollowupAudioPolicy::strictRecordedAt((string) ($task['recorded_at'] ?? '')); $media = new FollowupAudioPipelineMedia($this->settings); $audio = $media->inspect($path, (string) ($task['sha256'] ?? '')); if (abs((float) $task['duration_seconds'] - $audio['duration']) > 0.08) { throw new FollowupAudioException('AUDIO_INVALID'); } $chunkMs = (int) ($audio['channels'] === 2 ? $this->provider['stereo_chunk_seconds'] : $this->provider['chunk_seconds']) * 1000; FollowupAudioPipelineMedia::assertChunkPlan($audio['duration'], $chunkMs, $audio['channels']); $state = $checkpoint; if ($state === []) { if ((int) ($task['upstream_started_at'] ?? 0) > 0) { throw new FollowupAudioException('RECONCILIATION_REQUIRED', true); } $next = FollowupAudioPipelineCheckpoint::initial($task, $this->provider['fingerprint'], $chunkMs, $audio['channels']); $this->save($state, $next, $task, $heartbeat, 'transcribing'); } else { FollowupAudioPipelineCheckpoint::validate($state, $task); if ($state['schema_version'] !== 2 || $state['channel_policy'] !== $this->provider['channel_policy']) { throw new FollowupAudioException('PIPELINE_POLICY_CHANGED', true); } if ($state['source_channels'] !== $audio['channels'] || $state['chunk_ms'] !== $chunkMs) { throw new FollowupAudioException('AUDIO_INVALID', true); } } if (FollowupAudioPipelineCheckpoint::hasIntent($state)) { throw new FollowupAudioException('RECONCILIATION_REQUIRED', true); } $directory = sys_get_temp_dir() . '/followup-asr-' . bin2hex(random_bytes(16)); if (!mkdir($directory, 0700)) { throw new FollowupAudioException('AUDIO_NORMALIZATION_FAILED'); } $chunkPath = $directory . '/chunk.wav'; try { foreach ($state['chunks'] as $index => $chunk) { if (in_array($chunk['state'], ['complete', 'local_silence'], true)) { continue; } self::beat($heartbeat, []); $this->assertSource($audio['path'], $task['sha256']); $media->chunk($audio, $chunk, $chunkPath, $heartbeat); $this->assertSource($audio['path'], $task['sha256']); $silence = FollowupAudioPipelineMedia::digitalSilence($chunkPath); if ($silence !== null) { $next = $state; $next['chunks'][$index] += $silence + ['text' => '', 'source' => 'local_silence']; $next['chunks'][$index]['state'] = 'local_silence'; $this->save($state, $next, $task, $heartbeat, 'transcribing'); unlink($chunkPath); continue; } $next = $state; unset($next['chunks'][$index]['upstream_ids']); $next['chunks'][$index]['state'] = 'intent'; $next['chunks'][$index]['request_id'] = 'fa-' . bin2hex(random_bytes(16)); $this->save($state, $next, $task, $heartbeat, 'transcribing', true); $response = $this->request('asr', $this->transcriptionRequest($chunkPath, $state['chunks'][$index]['request_id']), $heartbeat); if ($response['rejected']) { $next = $state; $next['chunks'][$index]['state'] = 'rejected'; if ($response['upstream_ids'] !== []) { $next['chunks'][$index]['upstream_ids'] = $response['upstream_ids']; } $this->save($state, $next, $task, $heartbeat, 'transcribing'); throw new FollowupAudioException('ASR_REJECTED'); } $body = $response['body']; if (!is_string($body['text'] ?? null) || strlen($body['text']) > 200000 || !mb_check_encoding($body['text'], 'UTF-8')) { throw new FollowupAudioException('ASR_RESPONSE_INVALID', true); } $next = $state; $next['chunks'][$index]['state'] = 'complete'; $next['chunks'][$index]['text'] = $body['text']; if ($response['upstream_ids'] !== []) { $next['chunks'][$index]['upstream_ids'] = $response['upstream_ids']; } $this->save($state, $next, $task, $heartbeat, 'transcribing'); unlink($chunkPath); } $segments = FollowupAudioPipelineCheckpoint::segments($state); $transcript = implode("\n", array_column($segments, 'text')); self::assertTranscriptQuality($transcript, $segments); if ($state['extraction']['state'] !== 'complete') { $requestId = 'fa-' . bin2hex(random_bytes(16)); $payload = $this->extractRequest($transcript, $segments, $task['recorded_at'], FollowupAudioFields::catalog(), $requestId); self::beat($heartbeat, []); $this->assertSource($audio['path'], $task['sha256']); $next = $state; $next['extraction'] = ['state' => 'intent', 'request_id' => $requestId]; $this->save($state, $next, $task, $heartbeat, 'extracting', true); $response = $this->request('extraction', ['json' => $payload], $heartbeat); if ($response['rejected']) { $next = $state; $next['extraction']['state'] = 'rejected'; if ($response['upstream_ids'] !== []) { $next['extraction']['upstream_ids'] = $response['upstream_ids']; } $this->save($state, $next, $task, $heartbeat, 'extracting'); throw new FollowupAudioException('EXTRACTION_REJECTED'); } if (($this->provider['extraction']['protocol'] ?? 'openai') === 'dify_chat') { $body = $response['body']; $finish = $body['finish_reason'] ?? $body['metadata']['finish_reason'] ?? $body['metadata']['usage']['finish_reason'] ?? null; $answer = ($finish === null || $finish === 'stop') && empty($body['refusal']) && empty($body['tool_calls']) && is_string($body['answer'] ?? null) ? $body['answer'] : ''; } else { $choices = $response['body']['choices'] ?? []; $choice = is_array($choices) && count($choices) === 1 ? ($choices[0] ?? []) : []; $answer = ($choice['finish_reason'] ?? '') === 'stop' && empty($choice['message']['refusal']) && empty($choice['message']['tool_calls']) && is_string($choice['message']['content'] ?? null) ? $choice['message']['content'] : ''; } // A received invalid answer is still a known completed request; retain it encrypted, never silently reissue it. $next = $state; $next['extraction']['state'] = 'complete'; $next['extraction']['answer'] = $answer; if ($response['upstream_ids'] !== []) { $next['extraction']['upstream_ids'] = $response['upstream_ids']; } $this->save($state, $next, $task, $heartbeat, 'validating'); } $this->assertSource($audio['path'], $task['sha256']); self::beat($heartbeat, ['stage' => 'validating']); $receivedVersion = json_decode($state['extraction']['answer'], true)['schema_version'] ?? ''; if ($receivedVersion !== ($this->provider['output_schema'] ?? FollowupAudioExtractionSchema::VERSION)) { // The standalone validator can read historical V1, but a new V2 request never silently falls back to V1. throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } $result = self::validateAnswer($state['extraction']['answer'], $transcript, $segments, $task['recorded_at']); $result['uncertainties'][] = ($audio['channels'] === 2 ? '左右声道已独立转写;同一时间窗可能交叠,不代表话轮或角色已识别。' : '单声道转写未进行说话人分离。') . '患者、家属和客服归属须人工回听核对。'; return $result; } finally { if (is_file($chunkPath)) { unlink($chunkPath); } rmdir($directory); } } /** Pure production request builder for authorized cached-ASR acceptance; performs no network or state mutation. */ public function extractRequest(string $transcript, array $segments, string $recordedAt, array $catalog, ?string $requestId = null): array { $part = $this->provider['extraction']; $mode = $part['response_format'] ?? 'json_schema'; $maxTokens = $part['max_tokens'] ?? 8192; $thinking = $part['enable_thinking'] ?? null; if (!in_array($mode, ['json_schema', 'json_object', 'prompt_json'], true) || !is_int($maxTokens) || $maxTokens < 256 || $maxTokens > 8192 || ($thinking !== null && !is_bool($thinking))) { throw new FollowupAudioException('CONFIG_INVALID'); } $prompt = FollowupAudioTranscriptPrompt::build($transcript, $segments, $recordedAt, $catalog); self::assertTranscriptQuality($transcript, $segments); if (($part['protocol'] ?? 'openai') === 'dify_chat') { if ($mode !== 'prompt_json' || empty($part['binding_revision']) || $thinking !== null) { throw new FollowupAudioException('CONFIG_INVALID'); } return ['inputs' => new \stdClass(), 'query' => $prompt, 'response_mode' => 'blocking', 'user' => $this->opaqueUser($requestId ?? hash('sha256', $transcript)), 'auto_generate_name' => false]; } $payload = ['model' => $part['model'], 'stream' => false, 'temperature' => 0, 'max_tokens' => $maxTokens, 'messages' => [['role' => 'user', 'content' => $prompt]]]; if ($mode === 'json_schema') { $payload['response_format'] = self::buildResponseFormat($catalog, FollowupAudioTranscriptPrompt::citations($segments)); } elseif ($mode === 'json_object') { $payload['response_format'] = ['type' => 'json_object']; } // Explicit vendor extension only; portable unset config sends no chat_template_kwargs at all. if ($thinking !== null) { $payload['chat_template_kwargs'] = ['enable_thinking' => $thinking]; } return $payload; } public function transcriptionRequest(string $chunkPath, string $requestId): array { $file = new \CURLFile($chunkPath, 'audio/wav', 'chunk.wav'); if (($this->provider['asr']['protocol'] ?? 'openai') === 'dify') { return ['multipart' => ['file' => $file, 'user' => $this->opaqueUser($requestId)]]; } return ['multipart' => ['model' => $this->provider['asr']['model'], 'response_format' => 'json', 'file' => $file]]; } private function opaqueUser(string $requestId): string { return 'fa-opaque-' . substr(hash('sha256', $this->provider['fingerprint'] . ':' . $requestId), 0, 48); } public static function buildResponseFormat(array $catalog, array $citations): array { return FollowupAudioExtractionSchema::responseFormat($catalog, $citations); } /** Conservative ASR failure guard, not a claim of medical or transcription accuracy. */ private static function assertTranscriptQuality(string $transcript, array $segments): void { if (trim($transcript) === '') { throw new FollowupAudioException('ASR_TRANSCRIPT_EMPTY'); } foreach ($segments as $segment) { $text = preg_replace('/[\s\p{P}]+/u', '', $segment['text']); // Severe decoder loops are retained for manual review, never forwarded as reliable facts. if (!is_string($text) || preg_match('/(.)\1{39,}/u', $text) === 1 || preg_match('/(.{2,40})\1{19,}/u', $text) === 1) { throw new FollowupAudioException('ASR_QUALITY_REVIEW_REQUIRED'); } } } private function save(array &$state, array $next, array $task, callable $heartbeat, string $stage, bool $intent = false): void { $revision = $state['revision'] ?? 0; $next['revision'] = $revision + 1; FollowupAudioPipelineCheckpoint::transition($state, $next, $task, $revision); $fields = ['stage' => $stage, 'pipeline_checkpoint' => ['expected_revision' => $revision, 'state' => $next]]; if ($intent) { $fields['upstream_started_at'] = time(); } self::beat($heartbeat, $fields); $state = $next; } private static function beat(callable $heartbeat, array $fields): void { try { $ok = $heartbeat($fields); } catch (\Throwable $e) { throw new FollowupAudioException('LEASE_LOST', true); } if ($ok !== true) { throw new FollowupAudioException('LEASE_LOST', true); } } private function assertSource(string $path, string $hash): void { if (!hash_equals($hash, (string) hash_file('sha256', $path))) { throw new FollowupAudioException('AUDIO_INVALID'); } } /** Exact evidence must be in the cited canonical ASR segment. Model offsets and audio claims are forbidden. */ public static function validateAnswer(string $answer, string $transcript, array $segments, string $recordedAt, ?array $catalog = null): array { try { $raw = json_decode($answer, true, 64, JSON_THROW_ON_ERROR); } catch (\Throwable $e) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } if (!is_array($raw) || !in_array($raw['schema_version'] ?? '', ['followup-audio-transcript-v1', FollowupAudioExtractionSchema::VERSION], true) || array_diff(array_keys($raw), ['schema_version', 'summary', 'uncertainties', 'items']) || !is_string($raw['summary'] ?? null) || strlen($raw['summary']) > 60000 || !is_array($raw['items'] ?? null) || !array_is_list($raw['items']) || count($raw['items']) > 500 || !is_array($raw['uncertainties'] ?? null) || !array_is_list($raw['uncertainties']) || count($raw['uncertainties']) > 200) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } foreach ($raw['uncertainties'] as $text) { if (!is_string($text) || strlen($text) > 10000) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } } $byId = array_column($segments, null, 'id'); $v2 = $raw['schema_version'] === FollowupAudioExtractionSchema::VERSION; $citations = []; if ($v2) { $shape = json_decode($answer); if (!is_array($shape->items ?? null) || !is_array($shape->uncertainties ?? null)) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } if ($transcript !== implode("\n", array_column($segments, 'text'))) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } try { $citations = array_column(FollowupAudioTranscriptPrompt::citations($segments), null, 'id'); } catch (\Throwable $error) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } $catalog = $catalog ?? FollowupAudioFields::catalog(); } foreach ($raw['items'] as &$item) { if ($v2) { if (!is_array($item) || array_diff(['kind', 'values', 'record_date', 'record_time', 'date_text', 'time_text', 'time_period', 'time_estimated', 'needs_review', 'evidence_ids'], array_keys($item)) || array_key_exists('evidence', $item) || !is_array($item['evidence_ids'] ?? null) || !array_is_list($item['evidence_ids']) || count($item['evidence_ids']) < 1 || count($item['evidence_ids']) > FollowupAudioTranscriptPrompt::MAX_CITATIONS) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } $quotes = []; $seen = []; foreach ($item['evidence_ids'] as $citationId) { if (!is_string($citationId) || !isset($citations[$citationId]) || isset($seen[$citationId])) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } $citation = $citations[$citationId]; $seen[$citationId] = true; if (!isset($byId[$citation['segment_id']]) || !str_contains($byId[$citation['segment_id']]['text'], $citation['text'])) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } $quotes[] = ['segment_id' => $citation['segment_id'], 'text' => $citation['text']]; } unset($item['evidence_ids']); $item['evidence'] = $quotes; } if (!is_array($item) || array_diff(array_keys($item), ['kind', 'values', 'record_date', 'record_time', 'date_text', 'time_text', 'time_period', 'time_estimated', 'needs_review', 'evidence']) || !is_string($item['kind'] ?? null) || !is_array($item['values'] ?? null) || $item['values'] === [] || !is_string($item['date_text'] ?? null) || !is_string($item['time_text'] ?? null) || !is_bool($item['time_estimated'] ?? null) || !is_bool($item['needs_review'] ?? null) || !is_array($item['evidence'] ?? null) || !array_is_list($item['evidence']) || $item['evidence'] === []) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } foreach (['record_date', 'record_time', 'time_period'] as $key) { if (isset($item[$key]) && !is_string($item[$key])) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } } if ($v2) { FollowupAudioExtractionSchema::validateValues($item['kind'], $item['values'], $catalog); } try { FollowupAudioFields::validateValues($item['kind'], $item['values']); } catch (\Throwable $e) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } foreach ($item['evidence'] as &$evidence) { if (!is_array($evidence) || array_diff(array_keys($evidence), ['segment_id', 'text']) || !is_string($evidence['segment_id'] ?? null) || !isset($byId[$evidence['segment_id']]) || !is_string($evidence['text'] ?? null) || trim($evidence['text']) === '' || !str_contains($byId[$evidence['segment_id']]['text'], $evidence['text'])) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } $segment = $byId[$evidence['segment_id']]; $evidence += ['start_ms' => $segment['start_ms'], 'end_ms' => $segment['end_ms'], 'position_type' => 'segment']; if (array_key_exists('channel', $segment)) { $evidence['channel'] = $segment['channel']; } } unset($evidence); } unset($item); $raw['transcript'] = $transcript; $raw['transcript_segments'] = $segments; try { return FollowupAudioPolicy::normalizeExtraction($raw, $recordedAt); } catch (\Throwable $e) { throw new FollowupAudioException('UPSTREAM_SCHEMA_INVALID'); } } private function request(string $stage, array $payload, callable $heartbeat): array { $timeout = (int) ($this->settings[$stage . '_request_timeout'] ?? $this->settings['request_timeout'] ?? 240); $limit = (int) ($this->settings['max_response_bytes'] ?? 8388608); if ($timeout < 1 || $timeout > 300 || $limit < 1024 || $limit > 8388608) { throw new FollowupAudioException('CONFIG_INVALID'); } $part = $this->provider[$stage]; $protocol = $part['protocol'] ?? 'openai'; $endpoint = $stage === 'asr' ? ($protocol === 'dify' ? '/audio-to-text' : '/audio/transcriptions') : ($protocol === 'dify_chat' ? '/chat-messages' : '/chat/completions'); $spec = ['url' => $part['base_url'] . $endpoint, 'api_key' => $part['api_key'], 'timeout' => $timeout, 'stage' => $stage] + $payload; try { $response = $this->transport ? ($this->transport)($spec, $heartbeat) : (($this->provider['http_transport'] ?? 'curl') === 'openssl_stream' ? FollowupAudioStreamTransport::request($spec, $heartbeat, $limit, $this->provider['allow_loopback_tunnel']) : $this->curl($spec, $heartbeat, $limit)); } catch (\Throwable $e) { throw new FollowupAudioException('UPSTREAM_UNCERTAIN', true); } $http = (int) ($response['http_code'] ?? 0); $candidate = is_string($response['body'] ?? null) && strlen($response['body']) <= $limit ? json_decode($response['body'], true) : null; $ids = []; if (is_array($candidate)) { foreach (['task_id', 'message_id', 'conversation_id'] as $key) { if (is_string($candidate[$key] ?? null) && preg_match('/^[A-Za-z0-9_.:-]{1,190}$/D', $candidate[$key])) { $ids[$key] = $candidate[$key]; } } if (!isset($ids['message_id']) && $stage === 'extraction' && is_string($candidate['id'] ?? null) && preg_match('/^[A-Za-z0-9_.:-]{1,190}$/D', $candidate['id'])) { $ids['message_id'] = $candidate['id']; } } if ($ids !== []) { $fields = ['upstream_ids_json' => json_encode($ids, JSON_THROW_ON_ERROR)]; if (isset($ids['message_id']) || isset($ids['task_id'])) { $fields['upstream_run_id'] = $ids['message_id'] ?? $ids['task_id']; } self::beat($heartbeat, $fields); // Preserve observed opaque IDs even when the reply is uncertain/rejected. } if (($response['errno'] ?? 0) !== 0 || $http === 0 || $http >= 500 || $http === 408 || !is_string($response['body'] ?? null) || strlen($response['body']) > $limit) { throw new FollowupAudioException('UPSTREAM_UNCERTAIN', true); } if (in_array($http, [400, 401, 403, 404, 413, 415, 422, 429], true)) { return ['rejected' => true, 'body' => [], 'upstream_ids' => $ids]; } if ($http < 200 || $http >= 300) { throw new FollowupAudioException('UPSTREAM_UNCERTAIN', true); } try { $body = json_decode($response['body'], true, 64, JSON_THROW_ON_ERROR); } catch (\Throwable $e) { throw new FollowupAudioException('UPSTREAM_UNCERTAIN', true); } if (!is_array($body) || !empty($body['error']) || !empty($body['code']) || ($body['event'] ?? '') === 'error') { throw new FollowupAudioException('UPSTREAM_UNCERTAIN', true); } return ['rejected' => false, 'body' => $body, 'upstream_ids' => $ids]; } private function curl(array $spec, callable $heartbeat, int $limit): array { if (!function_exists('curl_init')) { throw new FollowupAudioException('CURL_UNAVAILABLE'); } $curl = curl_init(); $body = ''; $headers = ['Accept: application/json', 'Authorization: Bearer ' . $spec['api_key']]; $post = $spec['multipart'] ?? null; if (isset($spec['json'])) { $post = json_encode($spec['json'], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR); $headers[] = 'Content-Type: application/json'; } $last = microtime(true); curl_setopt_array($curl, [CURLOPT_URL => $spec['url'], CURLOPT_POST => true, CURLOPT_POSTFIELDS => $post, CURLOPT_HTTPHEADER => $headers, CURLOPT_CONNECTTIMEOUT => min(10, $spec['timeout']), CURLOPT_TIMEOUT => $spec['timeout'], CURLOPT_FOLLOWLOCATION => false, CURLOPT_SSL_VERIFYPEER => true, CURLOPT_SSL_VERIFYHOST => 2, CURLOPT_NOPROGRESS => false, CURLOPT_XFERINFOFUNCTION => static function (...$unused) use ($heartbeat, &$last): int { if (microtime(true) - $last < 5) { return 0; } $last = microtime(true); try { return $heartbeat([]) === true ? 0 : 1; } catch (\Throwable $e) { return 1; } }, CURLOPT_WRITEFUNCTION => static function ($handle, string $bytes) use (&$body, $limit): int { if (strlen($body) + strlen($bytes) > $limit) { return 0; } $body .= $bytes; return strlen($bytes); }]); curl_exec($curl); $errno = curl_errno($curl); $http = (int) curl_getinfo($curl, CURLINFO_HTTP_CODE); curl_close($curl); return ['body' => $body, 'errno' => $errno, 'http_code' => $http]; } }