getMethod($method)->invokeArgs(null, $args); } function filterFrame(array $data): string { return 'data: ' . json_encode($data, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR) . "\r\n\r\n"; } function filterDecode(string $protocol, array $chunks): array { return filterPrivate('decodeStreamChunks', [$protocol, $chunks]); } $cases = [ ['PRIVATE_PLAN临床正文。', '临床正文。'], ['PRIVATE_PLAN正文', '正文'], ['outerinnerouter正文', '正文'], ['前文PRIVATE_PLAN后文', '前文后文'], ['onetwo正文', '正文'], [" \nPRIVATE_PLAN\n正文", " \n\n正文"], [" \nPRIVATE_PLAN\n\t", ''], ['PRIVATE_PLAN', ''], ['PRIVATE_PLANPRIVATE_PLAN正文', '正文'], ['普通 thinking / planning；x < y；文字；<think>', '普通 thinking / planning；x < y；文字；<think>'], ['正文末尾 <', '正文末尾 <'], ['正文末尾 内容', '正文 内容'], ]; foreach ($cases as [$raw, $expected]) { filterCheck(DifyReasoningTextFilter::text($raw) === $expected, 'Whole text preserves exactly the visible body.'); for ($cut = 0; $cut <= strlen($raw); ++$cut) { $filter = new DifyReasoningTextFilter(); $actual = $filter->push(substr($raw, 0, $cut)) . $filter->push(substr($raw, $cut)) . $filter->finish(); filterCheck($actual === $expected, 'Any byte split matches whole-text filtering.'); filterCheck($filter->finish() === '' && $filter->push('late') === '', 'Terminal filtering is idempotent.'); foreach ($filter->counts() as $count) filterCheck(is_int($count), 'Filter diagnostics contain only integers.'); } $filter = new DifyReasoningTextFilter(); $visible = ''; foreach (str_split($raw) as $byte) $visible .= $filter->push($byte); $visible .= $filter->finish(); filterCheck($visible === $expected, 'One-byte filter chunks preserve UTF-8 bytes and tag boundaries.'); } foreach (['dify', 'openai'] as $protocol) { $wire = ''; foreach (['<', 'th', 'ink>', 'PRIVATE_PLAN', '', '临床', '正文。'] as $part) { $wire .= filterFrame($protocol === 'dify' ? ['event' => 'message', 'answer' => $part] : ['choices' => [['delta' => ['content' => $part]]]]); } $wire .= $protocol === 'dify' ? filterFrame(['event' => 'message_end']) : "data: [DONE]\r\n\r\n"; for ($cut = 0; $cut <= strlen($wire); ++$cut) { $out = filterDecode($protocol, [substr($wire, 0, $cut), substr($wire, $cut)]); filterCheck($out['content'] === '临床正文。' && implode('', $out['deltas']) === $out['content'] && $out['finished'], 'SSE wire splits and token-level tag splits never leak thinking.'); } $out = filterDecode($protocol, str_split($wire)); filterCheck($out['content'] === '临床正文。', 'One-byte SSE network reads retain exact clinical body.'); $blocking = $protocol === 'dify' ? ['answer' => 'PRIVATE_PLAN临床正文。', 'thought' => 'PRIVATE_FIELD'] : ['choices' => [['message' => ['reasoning_content' => 'PRIVATE_FIELD', 'content' => 'PRIVATE_PLAN临床正文。']]]]; $formatted = filterPrivate('formatResponse', [['errno' => 0, 'http_code' => 200, 'body' => json_encode($blocking, JSON_UNESCAPED_UNICODE)], microtime(true)]); filterCheck($formatted['ok'] && $formatted['content'] === $out['content'], 'Blocking and streaming expose the same body.'); foreach (['', 'PRIVATE_PLAN', 'PRIVATE_PLAN 'message', 'answer' => $reasoningOnly, 'thought' => 'PRIVATE_FIELD'] : ['choices' => [['delta' => ['content' => $reasoningOnly, 'reasoning_content' => 'PRIVATE_FIELD']]]]); $wire .= $protocol === 'dify' ? filterFrame(['event' => 'message_end']) : "data: [DONE]\n\n"; $out = filterDecode($protocol, str_split($wire)); filterCheck($out['content'] === '' && $out['deltas'] === [], 'Reasoning-only and unclosed blocks emit no clinical delta.'); $result = filterPrivate('formatStreamResponse', [['errno' => 0, 'http_code' => 200, 'content' => $out['content'], 'finished' => $out['finished']], microtime(true)]); filterCheck(!$result['ok'] && $result['error_code'] === 'EMPTY_RESPONSE', 'Reasoning-only stream is never report success.'); $payload = $protocol === 'dify' ? ['answer' => $reasoningOnly, 'thought' => 'PRIVATE_FIELD'] : ['choices' => [['message' => ['content' => $reasoningOnly, 'reasoning_content' => 'PRIVATE_FIELD']]]]; $result = filterPrivate('formatResponse', [['errno' => 0, 'http_code' => 200, 'body' => json_encode($payload)], microtime(true)]); filterCheck(!$result['ok'] && $result['error_code'] === 'EMPTY_RESPONSE', 'Reasoning-only blocking reply also fails empty.'); } } $parts = ['choices' => [['delta' => ['reasoning_content' => 'PRIVATE_FIELD', 'content' => [ ['type' => 'reasoning', 'text' => 'PRIVATE_PART'], ['type' => 'text', 'text' => 'PRIVATE_PLAN'], ['type' => 'text', 'text' => '真正正文'], ]]]]]; $out = filterDecode('openai', [filterFrame($parts), "data: [DONE]\n\n"]); filterCheck($out['content'] === '真正正文', 'Dedicated reasoning fields/parts stay private alongside genuine text parts.'); $parts['choices'][0]['message'] = $parts['choices'][0]['delta']; unset($parts['choices'][0]['delta']); filterCheck(filterPrivate('extractContent', [$parts]) === '真正正文', 'Blocking text-part arrays use the same filter.'); // Assert the real upstream clock never sees an inline plan as first answer. $now = 0.0; $timing = new DifyStreamTiming(static function () use (&$now): float { return $now; }); $state = filterPrivate('newStreamState', []); $state['timing'] = $timing; $tasks = []; $terminals = []; $received = []; $state['on_task'] = static function (string $id) use (&$tasks): void { $tasks[] = $id; }; $state['on_terminal'] = static function (string $id, string $status) use (&$terminals): void { $terminals[] = [$id, $status]; }; $consume = (new ReflectionClass(DifyChatService::class))->getMethod('consumeStreamFrame'); $receive = static function (string $delta) use (&$received): bool { $received[] = $delta; return true; }; foreach ([[10, ['event' => 'workflow_started', 'task_id' => 'filter-task']], [20, ['event' => 'message', 'task_id' => 'filter-task', 'answer' => 'PRIVATE_PLAN']], [30, ['event' => 'message', 'task_id' => 'filter-task', 'answer' => "\n"]]] as [$now, $event]) { $consume->invokeArgs(null, ['dify', filterFrame($event), &$state, $receive]); filterCheck(!isset($timing->snapshot()['first_answer_ms']) && $received === [], 'No first answer/delivery from thinking or whitespace.'); } $now = 40; $consume->invokeArgs(null, ['dify', filterFrame(['event' => 'message', 'answer' => '正文']), &$state, $receive]); $now = 50; $consume->invokeArgs(null, ['dify', filterFrame(['event' => 'message_end']), &$state, $receive]); $consume->invokeArgs(null, ['dify', filterFrame(['event' => 'workflow_finished', 'task_id' => 'filter-task', 'data' => ['status' => 'succeeded']]), &$state, $receive]); filterCheck($timing->snapshot()['first_answer_ms'] === 40 && $timing->snapshot()['first_delivered_ms'] === 40, 'True clinical text establishes first answer and delivery.'); filterCheck($tasks === ['filter-task'] && $terminals === [['filter-task', 'succeeded']], 'Task and terminal observers survive filtering.'); filterCheck($state['finished'] && $state['content'] === "\n正文" && implode('', $received) === $state['content'], 'Exact visible bytes and terminal completion stay consistent.'); filterCheck(!str_contains(json_encode($timing->snapshot()), 'PRIVATE_'), 'Telemetry never retains reasoning contents.'); $state = filterPrivate('newStreamState', []); $state['discard_deltas'] = true; $received = []; $consume->invokeArgs(null, ['dify', filterFrame(['event' => 'message', 'answer' => 'PRIVATE_PLANlate']), &$state, $receive]); $consume->invokeArgs(null, ['dify', filterFrame(['event' => 'message_end']), &$state, $receive]); filterCheck($received === [] && $state['content'] === '', 'Cancelled/timeout discard state never releases filtered or buffered text.'); echo "Dify incremental reasoning filter: {$assertions} assertions passed (pure offline fixtures).\n";