ZYT/zyt
3
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of http://git.zhenyangtang.com.cn/ZYT/zyt

Browse Source
This commit is contained in:
gr
2026-10-23 09:27:45 +08:00
parent 4880e96de3 1cbe8bfe54
commit 1224b02e40
34 changed files with 2866 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/common/service/DifyChatService.php
+337 -35
View File
@@ -23,6 +23,10 @@ class DifyChatService
private const DEFAULT_MAX_FILES = 3;
private const CANCEL_METADATA_SECONDS = 2.0;
private const CANCEL_DRAIN_SECONDS = 2.5;
/**
* 上游明确以“这批附件我处理不了”拒绝整次请求时使用的状态码。
* 命中后会去掉附件重试一次，避免一张舌象图让整份病历分析失败。
@@ -42,7 +46,17 @@ class DifyChatService
array $options = []
): array
{
// Cancellable callers collect a real upstream stream internally. Dify's stop
// API cannot stop blocking mode. Legacy blocking callers are unchanged.
// Only the server's pre-delta compatibility retry opts into real blocking;
// its abort closes transport and deliberately leaves remote stop unconfirmed.
if (is_callable($options['should_abort'] ?? null) && empty($options['blocking_compatibility'])) {
return self::streamChat($profile, $inputs, $query, $user,
static fn (string $delta): bool => true,
$options['should_abort'], $files, $options);
}
$config = config('prescription_ai') ?: [];
$shouldAbort = is_callable($options['should_abort'] ?? null) ? $options['should_abort'] : null;
if (empty($config['enable'])) {
return self::error('CONFIG_DISABLED', 'AI 报告功能未启用');
}
@@ -122,6 +136,7 @@ class DifyChatService
$lastSpec = [];
foreach ($requestSpecs as $index => $requestSpec) {
if ($shouldAbort !== null && $shouldAbort()) return self::error('AI_ASSISTANT_CANCELLED', '本次请求已取消');
// The legacy compatible path cannot transmit documents. Strict callers must
// receive an explicit gap, never a successful text-only fallback.
if ($strictFiles && !self::strictProtocolSupportsFiles($requestSpec['protocol'], $attempt['files'])) {
@@ -141,8 +156,12 @@ class DifyChatService
$requestSpec['url'],
$requestSpec['payload'],
$apiKey,
$remainingTimeout
$remainingTimeout,
$options + ['attachment_items' => self::attachmentItems($attempt['files'], $attempt['omitted'], $requestSpec['protocol'])]
);
if (!empty($response['client_aborted']) || ($shouldAbort !== null && $shouldAbort())) {
return self::error('AI_ASSISTANT_CANCELLED', '本次请求已取消', self::elapsedMilliseconds($startedAt));
}
$lastResponse = $response;
$lastSpec = $requestSpec;
$fileRejected = $fileRejected || self::isFileRejection($response, $attempt['files']);
@@ -207,7 +226,8 @@ class DifyChatService
string $user,
callable $onDelta,
?callable $shouldAbort = null,
array $files = []
array $files = [],
array $options = []
): array {
$config = config('prescription_ai') ?: [];
if (empty($config['enable'])) {
@@ -229,7 +249,7 @@ class DifyChatService
return self::error('CONFIG_INVALID', 'AI 服务配置无效');
}
$timeout = (int) ($config['timeout'] ?? 0);
$timeout = (int) ($options['timeout'] ?? $config['timeout'] ?? 0);
if (!self::isValidTimeout($timeout)) {
return self::error('CONFIG_INVALID', 'AI 服务超时配置无效');
}
@@ -242,18 +262,28 @@ class DifyChatService
return self::error('CONFIG_INVALID', 'AI 模型配置无效');
}
$ancientBookSources = ClinicalKnowledgeReference::ancientBookSources($query);
$evidenceOnly = ($options['knowledge_mode'] ?? '') === 'evidence';
$ancientBookSources = $evidenceOnly ? [] : ClinicalKnowledgeReference::ancientBookSources($query);
try {
$query = ClinicalKnowledgeReference::augmentQueryWithAncientBookSources($query, $ancientBookSources);
if (!$evidenceOnly) {
$query = ClinicalKnowledgeReference::augmentQueryWithAncientBookSources($query, $ancientBookSources);
}
} catch (\RuntimeException $error) {
return self::error('SKILL_UNAVAILABLE', '临床 AI 参考资料不可用');
}
$normalized = self::normalizeFiles($files, self::maxFiles($config, $modelConfig));
$strictFiles = !empty($options['strict_files']);
$normalized = self::normalizeFiles($files, self::maxFiles($config, $modelConfig), !$strictFiles);
if ($strictFiles && !self::strictFilesComplete($files, $normalized)) {
return self::error('STRICT_FILES_INVALID_OR_LIMIT', '附件无效或超出单批上限，须分批处理');
}
$startedAt = microtime(true);
$formatted = null;
foreach (self::buildAttemptPlan($normalized['kept'], $normalized['dropped']) as $attempt) {
$attemptPlan = $strictFiles
? [['files' => $normalized['kept'], 'omitted' => []]]
: self::buildAttemptPlan($normalized['kept'], $normalized['dropped']);
foreach ($attemptPlan as $attempt) {
$fileRejected = false;
$inputRejected = false;
@@ -275,6 +305,12 @@ class DifyChatService
$lastSpec = [];
foreach ($requestSpecs as $index => $requestSpec) {
if ($shouldAbort !== null && $shouldAbort()) {
return self::error('AI_ASSISTANT_CANCELLED', '本次请求已取消');
}
if ($strictFiles && !self::strictProtocolSupportsFiles($requestSpec['protocol'], $attempt['files'])) {
return self::error('FILE_TYPE_UNSUPPORTED', '当前模型接口不支持此类原始附件');
}
$elapsedSeconds = (int) floor(microtime(true) - $startedAt);
$remainingTimeout = $timeout - $elapsedSeconds;
if ($remainingTimeout < self::MIN_TIMEOUT) {
@@ -292,7 +328,10 @@ class DifyChatService
$apiKey,
$remainingTimeout,
$onDelta,
$shouldAbort
$shouldAbort,
$options + ['profile' => $profile, 'user' => $user,
'source_summary' => $options['source_summary'] ?? [],
'attachment_items' => self::attachmentItems($attempt['files'], $attempt['omitted'], $requestSpec['protocol'])]
);
$lastResponse = $response;
$lastSpec = $requestSpec;
@@ -317,9 +356,17 @@ class DifyChatService
$lastResponse = $lastResponse ?? self::emptyStreamResponse(0);
$formatted = self::formatStreamResponse($lastResponse, $startedAt);
if (!empty($lastResponse['client_aborted']) || !empty($lastResponse['callback_error'])
|| ($shouldAbort !== null && $shouldAbort())) {
return self::error('AI_ASSISTANT_CANCELLED', '本次请求已取消', self::elapsedMilliseconds($startedAt));
}
self::logUpstreamFailure($lastSpec, $lastResponse, $query, $attempt['files'], $formatted);
if (!empty($formatted['ok'])) {
$formatted['ancient_book_sources'] = $ancientBookSources;
if ($strictFiles) {
$formatted['transmitted_file_count'] = count($attempt['files']);
$formatted['attachment_transport'] = (string) ($lastSpec['protocol'] ?? '');
}
return $formatted;
}
if (
@@ -340,6 +387,38 @@ class DifyChatService
return $formatted ?? self::error('UPSTREAM_REJECTED', '模型未能处理本次请求', self::elapsedMilliseconds($startedAt));
}
/** The task id is obtained from streaming metadata; user must match the original request. */
public static function stopTask(string $profile, string $taskId, string $user): bool
{
if (preg_match('/^[a-zA-Z0-9_-]{1,128}$/', $taskId) !== 1 || $user === '') return false;
$config = config('prescription_ai') ?: [];
$model = self::resolveProfileConfig($config, $profile);
$base = trim((string) ($config['base_url'] ?? ''));
$key = (string) ($model['api_key'] ?? '');
if (!self::isValidBaseUrl($base) || trim($key) === '' || strpbrk($key, "\r\n") !== false) return false;
$endpoint = self::buildEndpoint(rtrim($base, '/'), 'chat-messages');
$response = self::sendRequest($endpoint . '/' . rawurlencode($taskId) . '/stop', ['user' => $user], trim($key), 1);
$body = json_decode((string) ($response['body'] ?? ''), true);
return ($response['errno'] ?? -1) === 0 && ($response['http_code'] ?? 0) === 200
&& ($body['result'] ?? '') === 'success';
}
/** Safe labels only; submitted means request references sent, never remote file reading. */
private static function attachmentItems(array $files, array $omitted, string $protocol): array
{
$labels = ['image' => '图片附件', 'document' => '文档附件', 'audio' => '音频附件', 'video' => '视频附件', 'custom' => '其他附件'];
$items = [];
$counts = [];
foreach (array_merge($files, $omitted) as $index => $file) {
$kind = $file['type'] ?? 'custom';
$counts[$kind] = ($counts[$kind] ?? 0) + 1;
$transmitted = $index < count($files) && ($protocol === 'dify' || $kind === 'image');
$items[] = ['label' => ($labels[$kind] ?? '附件') . $counts[$kind], 'kind' => $kind,
'status' => $transmitted ? 'submitting' : 'omitted'];
}
return $items;
}
/**
* @param array<string,mixed> $config
* @return array<string,mixed>|null
@@ -383,6 +462,9 @@ class DifyChatService
'query' => self::withAttachmentManifest($query, $omitted),
'response_mode' => $streaming ? 'streaming' : 'blocking',
'user' => $user,
// The workstation does not use Dify conversation titles. Avoid a
// separate background LLM call that may outlive the answer's stop.
'auto_generate_name' => false,
],
];
if ($files !== []) {
@@ -740,7 +822,8 @@ class DifyChatService
string $url,
array $payload,
string $apiKey,
int $timeout
int $timeout,
array $options = []
): array {
$body = json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_INVALID_UTF8_SUBSTITUTE);
if ($body === false) {
@@ -751,7 +834,14 @@ class DifyChatService
if ($ch === false) {
return ['body' => '', 'errno' => -2, 'http_code' => 0];
}
curl_setopt_array($ch, [
$shouldAbort = is_callable($options['should_abort'] ?? null) ? $options['should_abort'] : null;
$clientAborted = false;
$notify = self::transferProgressNotifier($options);
$submitted = false;
$markSubmitted = static function () use (&$submitted, $notify): void {
if (!$submitted) { $submitted = true; $notify('submitted'); }
};
$curlOptions = [
CURLOPT_URL => $url,
CURLOPT_POST => true,
CURLOPT_POSTFIELDS => $body,
@@ -765,20 +855,85 @@ class DifyChatService
'Accept: application/json',
'Authorization: Bearer ' . $apiKey,
],
]);
];
if (isset($options['timeout_ms'])) {
$milliseconds = max(1, min($timeout * 1000, (int) $options['timeout_ms']));
$curlOptions[CURLOPT_TIMEOUT_MS] = $milliseconds;
$curlOptions[CURLOPT_CONNECTTIMEOUT_MS] = min(1000, $milliseconds);
}
if ($shouldAbort !== null) {
// Dify blocking mode exposes no streaming task id. Compatibility mode
// can close its transfer promptly, but cannot claim remote stop acceptance.
$notify('submitting');
$progressOption = self::curlProgressOption();
if ($progressOption === null) {
curl_close($ch);
return ['body' => '', 'errno' => -3, 'http_code' => 0, 'client_aborted' => true];
}
$curlOptions[CURLOPT_NOPROGRESS] = false;
$curlOptions[$progressOption] = static function (...$values) use ($shouldAbort, &$clientAborted, $markSubmitted): int {
if (($values[3] ?? 0) > 0 && ($values[4] ?? 0) >= $values[3]) $markSubmitted();
if ($shouldAbort()) { $clientAborted = true; return 1; }
return 0;
};
$curlOptions[CURLOPT_HEADERFUNCTION] = static function ($handle, string $line) use ($markSubmitted): int {
if (preg_match('/^HTTP\/\S+\s+2\d{2}(?:\s|$)/i', trim($line))) $markSubmitted();
return strlen($line);
};
}
curl_setopt_array($ch, $curlOptions);
$responseBody = curl_exec($ch);
$errno = curl_errno($ch);
$httpCode = (int) curl_getinfo($ch, CURLINFO_HTTP_CODE);
curl_close($ch);
$responseBody = '';
$errno = -3;
$httpCode = 0;
try {
$responseBody = curl_exec($ch);
$errno = curl_errno($ch);
$httpCode = (int) curl_getinfo($ch, CURLINFO_HTTP_CODE);
if ($shouldAbort !== null && $shouldAbort()) $clientAborted = true;
} catch (\Throwable $ignored) {
$clientAborted = $shouldAbort !== null;
} finally {
curl_close($ch);
}
if ($shouldAbort !== null) {
if ($clientAborted && is_callable($options['on_stop'] ?? null)) ($options['on_stop'])(false);
if (is_callable($options['on_transfer_end'] ?? null)) ($options['on_transfer_end'])();
if (!$clientAborted && ($errno !== 0 || $httpCode >= 400)) $notify('failed');
}
return [
'body' => is_string($responseBody) ? $responseBody : '',
'errno' => $errno,
'http_code' => $httpCode,
'client_aborted' => $clientAborted,
];
}
private static function transferProgressNotifier(array $options): callable
{
return static function (string $status) use ($options): void {
if (!is_callable($options['on_progress'] ?? null)) return;
$items = $options['attachment_items'] ?? [];
foreach ($items as &$item) {
if ($item['status'] !== 'omitted') $item['status'] = $status;
}
unset($item);
($options['on_progress'])(['phase' => 'attachments', 'status' => $status,
'message' => $status === 'submitted' ? '资料正文与附件引用已提交，等待模型；远程附件读取进度不可观测'
: ($status === 'failed' ? '本次资料请求提交失败' : '正在提交资料正文与附件引用'),
'source_summary' => $options['source_summary'] ?? [],
'source_form' => $options['source_form'] ?? 'full_context',
'evidence_cache_hit' => !empty($options['evidence_cache_hit']),
'attachment_transfer' => 'remote_url_references',
'attachment_count' => count(array_filter($items, static fn (array $item): bool => $item['status'] !== 'omitted')),
'items' => $items]);
if ($status === 'submitted') ($options['on_progress'])([
'phase' => 'upstream', 'status' => 'waiting', 'message' => '资料请求已提交，正在等待模型正文',
]);
};
}
/**
* libcurl 7.32+ exposes CURLOPT_XFERINFOFUNCTION, while CentOS/RHEL 7 commonly
* ships libcurl 7.29 with only CURLOPT_PROGRESSFUNCTION. Resolve the option
@@ -810,7 +965,8 @@ class DifyChatService
string $apiKey,
int $timeout,
callable $onDelta,
?callable $shouldAbort
?callable $shouldAbort,
array $options = []
): array {
$body = json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_INVALID_UTF8_SUBSTITUTE);
if ($body === false) {
@@ -824,10 +980,74 @@ class DifyChatService
$buffer = '';
$state = self::newStreamState();
$state['on_task'] = $options['on_task'] ?? null;
$state['on_terminal'] = $options['on_terminal'] ?? null;
$notify = self::transferProgressNotifier($options);
$notify('submitting');
$submitted = false;
$markSubmitted = static function () use (&$submitted, $notify): void {
if (!$submitted) { $submitted = true; $notify('submitted'); }
};
$cancelAt = null;
$stopAttempted = false;
$stopAccepted = false;
$attemptStop = static function () use ($options, $protocol, $url, $payload, $apiKey,
&$state, &$cancelAt, &$stopAttempted, &$stopAccepted): void {
if ($protocol !== 'dify' || $state['task_id'] === '' || $state['upstream_terminal_confirmed']) return;
$remainingMs = (int) floor((self::CANCEL_DRAIN_SECONDS - (hrtime(true) / 1e9 - $cancelAt)) * 1000);
if ($remainingMs <= 0) return;
$performStop = static function () use ($url, $payload, $apiKey, &$state, $remainingMs): bool {
$response = self::sendRequest(rtrim($url, '/') . '/' . rawurlencode($state['task_id']) . '/stop',
['user' => (string) ($payload['user'] ?? '')], $apiKey, 1,
['timeout_ms' => min(1000, $remainingMs)]);
$body = json_decode((string) ($response['body'] ?? ''), true);
return ($response['errno'] ?? -1) === 0 && ($response['http_code'] ?? 0) === 200
&& ($body['result'] ?? '') === 'success';
};
if (is_callable($options['on_cancel_task'] ?? null)) {
// Both workers use the same atomic task claim; a pending owner
// keeps this generation connection alive while the peer stops it.
$outcome = ($options['on_cancel_task'])($state['task_id'], $performStop);
$stopAccepted = $stopAccepted || (is_array($outcome) && !empty($outcome['accepted']));
} elseif (!$stopAttempted) {
$stopAttempted = true;
$stopAccepted = $performStop();
}
};
$abort = static function () use ($shouldAbort, $protocol, &$state, &$cancelAt, $attemptStop): bool {
if ($cancelAt === null && !$state['callback_error'] && ($shouldAbort === null || !$shouldAbort())) return false;
$cancelAt = $cancelAt ?? hrtime(true) / 1e9;
$state['client_aborted'] = true;
$state['discard_deltas'] = true;
if ($protocol !== 'dify') return true;
// One absolute cancellation budget includes stop I/O and metadata
// discovery. Silent headers/body cannot prolong it: the multi pump
// polls even when libcurl's progress callback has no network events.
$budget = $state['task_id'] === '' ? self::CANCEL_METADATA_SECONDS : self::CANCEL_DRAIN_SECONDS;
if (hrtime(true) / 1e9 - $cancelAt >= $budget) return true;
$attemptStop();
// workflow_finished confirms execution state, but Dify can still
// yield message_end and persist cleanup after that frame. Preserve
// the connection until natural EOF or this same absolute budget.
return $state['callback_error'] || hrtime(true) / 1e9 - $cancelAt >= $budget;
};
$forward = static function (string $delta) use ($shouldAbort, $onDelta, &$state, &$cancelAt): bool {
if ($cancelAt !== null || ($shouldAbort !== null && $shouldAbort())) {
$state['discard_deltas'] = true;
return true; // Consume metadata/terminal frames without accumulating cancelled text.
}
$accepted = $onDelta($delta) !== false;
if (!$accepted && $shouldAbort !== null && $shouldAbort()) {
$state['discard_deltas'] = true;
return true;
}
return $accepted;
};
$responseCode = 0;
$header = static function ($handle, string $line) use (&$responseCode): int {
$header = static function ($handle, string $line) use (&$responseCode, $markSubmitted): int {
if (preg_match('/^HTTP\/\S+\s+(\d{3})(?:\s|$)/i', trim($line), $matches) === 1) {
$responseCode = (int) $matches[1];
if ($responseCode >= 200 && $responseCode < 300) $markSubmitted();
}
return strlen($line);
};
@@ -836,24 +1056,21 @@ class DifyChatService
&$buffer,
&$state,
&$responseCode,
$onDelta,
$shouldAbort
$forward,
$abort
): int {
if ($shouldAbort !== null && $shouldAbort()) {
$state['client_aborted'] = true;
return 0;
}
if ($responseCode < 200 || $responseCode >= 300) {
// Never decode or forward an error response body. Besides preventing
// leakage, this keeps 404/405 protocol fallback side-effect free.
return strlen($chunk);
}
self::consumeStreamBytes($protocol, $buffer, $chunk, $state, $onDelta);
self::consumeStreamBytes($protocol, $buffer, $chunk, $state, $forward);
if ($abort()) { $state['client_aborted'] = true; return 0; }
return $state['callback_error'] ? 0 : strlen($chunk);
};
$progress = static function (...$unused) use (&$state, $shouldAbort): int {
if ($shouldAbort !== null && $shouldAbort()) {
$state['client_aborted'] = true;
$progress = static function (...$values) use ($abort, $markSubmitted): int {
if (($values[3] ?? 0) > 0 && ($values[4] ?? 0) >= $values[3]) $markSubmitted();
if ($abort()) {
return 1;
}
return 0;
@@ -883,15 +1100,40 @@ class DifyChatService
}
curl_setopt_array($ch, $curlOptions);
curl_exec($ch);
$errno = curl_errno($ch);
$httpCode = (int) curl_getinfo($ch, CURLINFO_HTTP_CODE);
curl_close($ch);
if (!$state['client_aborted'] && !$state['callback_error']) {
self::consumeStreamBytes($protocol, $buffer, '', $state, $onDelta, true);
$errno = -3;
$httpCode = $responseCode;
try {
if ($shouldAbort !== null) {
$errno = self::executeCancellableStream($ch, $abort);
} else {
curl_exec($ch);
$errno = curl_errno($ch);
}
$httpCode = (int) curl_getinfo($ch, CURLINFO_HTTP_CODE);
} catch (\Throwable $ignored) {
// Observer/storage errors must fail closed and still stop a known task.
$state['callback_error'] = true;
} finally {
// A callback failure may escape libcurl before the regular cancel
// poll. Make the coordinated stop attempt before closing this handle.
try { $abort(); } catch (\Throwable $ignored) {}
curl_close($ch);
}
if (!$state['callback_error']) {
try { self::consumeStreamBytes($protocol, $buffer, '', $state, $forward, true); }
catch (\Throwable $ignored) { $state['callback_error'] = true; }
}
$cancelled = $state['client_aborted'] || $state['callback_error'];
try { $cancelled = $cancelled || ($shouldAbort !== null && $shouldAbort()); }
catch (\Throwable $ignored) { $cancelled = true; }
if ($cancelled) {
$state['client_aborted'] = true;
if (is_callable($options['on_stop'] ?? null)) ($options['on_stop'])($stopAccepted, $state['task_id']);
}
if (is_callable($options['on_transfer_end'] ?? null)) ($options['on_transfer_end'])($state['task_id']);
if (!$cancelled && ($errno !== 0 || $httpCode >= 400)) $notify('failed');
return [
'errno' => $errno,
'http_code' => $httpCode,
@@ -903,9 +1145,39 @@ class DifyChatService
'client_aborted' => $state['client_aborted'],
'callback_error' => $state['callback_error'],
'finished' => $state['finished'],
'upstream_terminal_confirmed' => $state['upstream_terminal_confirmed'],
'upstream_terminal_status' => $state['upstream_terminal_status'],
];
}
/** Poll cancellation at most 50ms apart, including a silent upstream. */
private static function executeCancellableStream($ch, callable $abort): int
{
$multi = curl_multi_init();
$added = false;
$errno = 0;
try {
if (curl_multi_add_handle($multi, $ch) !== CURLM_OK) return -3;
$added = true;
do {
if ($abort()) return CURLE_ABORTED_BY_CALLBACK;
do { $result = curl_multi_exec($multi, $running); }
while ($result === CURLM_CALL_MULTI_PERFORM);
while (($info = curl_multi_info_read($multi)) !== false) {
if (($info['handle'] ?? null) === $ch) $errno = (int) $info['result'];
}
if ($result !== CURLM_OK) return -3;
if (!$running) break;
if ($abort()) return CURLE_ABORTED_BY_CALLBACK;
if (curl_multi_select($multi, 0.05) === -1) usleep(1000);
} while (true);
return $errno;
} finally {
if ($added) curl_multi_remove_handle($multi, $ch);
curl_multi_close($multi);
}
}
/**
* @return array{
* content:string,message_id:string,emitted:bool,upstream_error:bool,
@@ -923,6 +1195,10 @@ class DifyChatService
'client_aborted' => false,
'callback_error' => false,
'finished' => false,
'task_id' => '',
'discard_deltas' => false,
'upstream_terminal_confirmed' => false,
'upstream_terminal_status' => '',
];
}
@@ -980,7 +1256,7 @@ class DifyChatService
array &$state,
callable $onDelta
): void {
if ($state['finished'] || $state['upstream_error'] || $state['callback_error']) {
if ($state['callback_error']) {
return;
}
@@ -1009,7 +1285,31 @@ class DifyChatService
$delta = '';
if ($protocol === 'dify') {
$taskId = $decoded['task_id'] ?? '';
if (is_string($taskId) && preg_match('/^[a-zA-Z0-9_-]{1,128}$/', $taskId) === 1) {
// One HTTP generation owns one task. Ignore foreign late metadata.
if ($state['task_id'] !== '' && $taskId !== $state['task_id']) return;
if ($state['task_id'] === '') {
$state['task_id'] = $taskId;
if (is_callable($state['on_task'] ?? null)) ($state['on_task'])($taskId);
}
}
$event = strtolower((string) ($decoded['event'] ?? ''));
if ($event === 'workflow_finished') {
$rawStatus = $decoded['data']['status'] ?? '';
$status = is_string($rawStatus) ? strtolower($rawStatus) : '';
if ($status === 'cancelled') $status = 'stopped';
if (!$state['upstream_terminal_confirmed'] && $state['task_id'] !== ''
&& in_array($status, ['stopped', 'succeeded', 'failed'], true)) {
$state['upstream_terminal_confirmed'] = true;
$state['upstream_terminal_status'] = $status;
if (is_callable($state['on_terminal'] ?? null)) ($state['on_terminal'])($state['task_id'], $status);
}
return;
}
// Keep observing workflow_finished even after message_end; the latter
// alone is stream completion, never proof that a workflow stopped.
if ($state['finished'] || $state['upstream_error']) return;
if ($event === 'message_end') {
$state['message_id'] = (string) ($decoded['message_id'] ?? $state['message_id']);
$state['finished'] = true;
@@ -1027,6 +1327,7 @@ class DifyChatService
$delta = is_string($decoded['answer'] ?? null) ? $decoded['answer'] : '';
$state['message_id'] = (string) ($decoded['message_id'] ?? $state['message_id']);
} else {
if ($state['finished'] || $state['upstream_error']) return;
$delta = self::extractStreamDelta($decoded);
$state['message_id'] = (string) ($decoded['id'] ?? $state['message_id']);
}
@@ -1044,6 +1345,7 @@ class DifyChatService
$state['callback_error'] = true;
return;
}
if ($state['discard_deltas']) return;
$state['content'] .= $delta;
$state['emitted'] = true;
}