Files
zyt/server/tests/DifyWorkflowRunIdentityTest.php
T
2026-10-27 14:03:35 +08:00

305 lines
24 KiB
PHP

<?php
declare(strict_types=1);
namespace think\facade {
final class Log {
public static array $lines = [];
public static function write(string $message, string $level): void { self::$lines[] = $message; }
public static function warning(string $message): void { self::$lines[] = $message; }
}
}
namespace app\common\service {
require __DIR__ . '/fixtures/CancellableCurlMultiStub.php';
// These namespaced cURL functions and clock are deterministic in-process
// fixtures. No socket, provider, runtime control or database is contacted.
function hrtime(bool $asNumber = false): int|array {
$nanoseconds = (int) (($GLOBALS['runIdentityNow'] ?? 0) * 1000000);
return $asNumber ? $nanoseconds : [intdiv($nanoseconds, 1000000000), $nanoseconds % 1000000000];
}
function curl_init(): \stdClass { return (object) ['errno' => 0]; }
function curl_setopt_array(\stdClass $handle, array $options): bool { $handle->options = $options; return true; }
function curl_exec(\stdClass $handle): string|bool {
$options = $handle->options;
if (str_ends_with($options[CURLOPT_URL], '/stop')) {
\runIdentityExpect(!$GLOBALS['runIdentityClosed'], 'stop precedes generation transport close');
\runIdentityExpect(str_ends_with($options[CURLOPT_URL], '/private-task/stop')
&& json_decode($options[CURLOPT_POSTFIELDS], true) === ['user' => 'private-user'],
'stop retains the original task and user');
++$GLOBALS['runIdentityStopIo'];
$GLOBALS['runIdentityNow'] += 7;
return '{"result":"success"}';
}
$header = $options[CURLOPT_HEADERFUNCTION];
$write = $options[CURLOPT_WRITEFUNCTION];
$progressOption = defined('CURLOPT_XFERINFOFUNCTION') ? CURLOPT_XFERINFOFUNCTION : CURLOPT_PROGRESSFUNCTION;
$progress = $options[$progressOption];
$header($handle, "HTTP/1.1 200 OK\r\n");
$mode = $GLOBALS['runIdentityMode'];
if (!in_array($mode, ['late_unknown', 'initial_observer_failure'], true)) {
$write($handle, \runIdentityFrame(['event' => 'node_started', 'task_id' => 'private-task']));
}
$GLOBALS['runIdentityCancelled'] = !in_array($mode, ['observer_failure', 'initial_observer_failure'], true);
$GLOBALS['runIdentityNow'] += 10;
$progress($handle, 0, 0, 1, 1);
$GLOBALS['runIdentityNow'] += $mode === 'absolute_bound' ? 2400 : 100;
$start = \runIdentityFrame(['event' => 'workflow_started', 'task_id' => 'private-task',
'workflow_run_id' => 'private-run', 'data' => ['id' => 'private-run']]);
if (in_array($mode, ['observer_failure', 'initial_observer_failure'], true)) {
// A whole terminal in the same received chunk still cannot bypass
// failure to persist the preceding identity enrichment.
$start .= \runIdentityFrame(['event' => 'workflow_finished', 'task_id' => 'private-task',
'workflow_run_id' => 'private-run', 'data' => ['id' => 'private-run', 'status' => 'succeeded']]);
}
if ($write($handle, $start) === 0) { $handle->errno = CURLE_WRITE_ERROR; return false; }
if ($mode === 'absolute_bound') {
$GLOBALS['runIdentityNow'] = $GLOBALS['runIdentityInterruptedAt'] + 2501;
\runIdentityExpect($progress($handle, 0, 0, 1, 1) !== 0, 'late enrichment cannot restart the absolute drain bound');
$handle->errno = CURLE_ABORTED_BY_CALLBACK;
return false;
}
$write($handle, \runIdentityFrame(['event' => 'message', 'task_id' => 'private-task', 'answer' => 'private-late-text']));
$write($handle, \runIdentityFrame(['event' => 'message_end']) . "data: [DONE]\r\n\r\n");
if ($mode !== 'accepted_only') {
$terminal = \runIdentityFrame(['event' => 'workflow_finished', 'task_id' => 'private-task',
'workflow_run_id' => 'private-run', 'data' => ['id' => 'private-run', 'status' => 'cancelled',
'error' => 'private-provider-error']]);
// Natural EOF flushes a final unterminated terminal frame.
$write($handle, rtrim($terminal, "\r\n"));
}
return true;
}
function curl_errno(\stdClass $handle): int { return $handle->errno; }
function curl_getinfo(\stdClass $handle, int $option): int|float { return $option === CURLINFO_HTTP_CODE ? 200 : 0.001; }
function curl_close(\stdClass $handle): void {
if (!str_ends_with($handle->options[CURLOPT_URL], '/stop')) $GLOBALS['runIdentityClosed'] = true;
}
}
namespace {
require dirname(__DIR__) . '/vendor/autoload.php';
use app\common\service\DifyChatService;
use app\common\service\DifyStreamTiming;
use think\facade\Log;
$runIdentityAssertions = 0;
function runIdentityExpect(bool $condition, string $message): void {
++$GLOBALS['runIdentityAssertions'];
if (!$condition) throw new RuntimeException($message);
}
function runIdentityPrivate(string $method, array $arguments = []): mixed {
return (new ReflectionClass(DifyChatService::class))->getMethod($method)->invokeArgs(null, $arguments);
}
function runIdentityFrame(array $event, string $separator = "\r\n\r\n"): string {
return 'data: ' . json_encode($event, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) . $separator;
}
function runIdentityParse(array $events, ?array $chunks = null, array $overrides = []): array {
$state = runIdentityPrivate('newStreamState');
$timing = new DifyStreamTiming();
$tasks = []; $terminals = []; $deltas = [];
$state['timing'] = $timing;
$state['on_task'] = static function (string $task, string $run = '') use (&$tasks): void { $tasks[] = [$task, $run]; };
$state['on_terminal'] = static function (string $task, string $status, string $run = '') use (&$terminals): void {
$terminals[] = [$task, $status, $run];
};
$state = array_replace($state, $overrides);
$buffer = '';
$receive = static function (string $delta) use (&$deltas): bool { $deltas[] = $delta; return true; };
$chunks ??= array_map('runIdentityFrame', $events);
foreach ($chunks as $chunk) runIdentityPrivate('consumeStreamBytes', ['dify', &$buffer, $chunk, &$state, $receive]);
runIdentityPrivate('consumeStreamBytes', ['dify', &$buffer, '', &$state, $receive, true]);
return compact('state', 'tasks', 'terminals', 'deltas', 'timing');
}
$start = ['event' => 'workflow_started', 'task_id' => 'private-task', 'workflow_run_id' => 'private-run',
'data' => ['id' => 'private-run', 'workflow_id' => 'private-definition']];
$terminal = ['event' => 'workflow_finished', 'task_id' => 'private-task', 'workflow_run_id' => 'private-run',
'data' => ['id' => 'private-run', 'status' => 'cancelled']];
$wire = ": ping\r\n\r\n" . runIdentityFrame(['event' => 'node_started', 'task_id' => 'private-task'])
. runIdentityFrame($start) . runIdentityFrame($start)
. runIdentityFrame(['event' => 'message', 'answer' => '<think>private-plan</think>诊断正文'])
. runIdentityFrame(['event' => 'message_end', 'message_id' => 'public-message'])
. "data: [DONE]\r\n\r\n" . runIdentityFrame($terminal, '');
for ($split = 0; $split <= strlen($wire); ++$split) {
$parsed = runIdentityParse([], [substr($wire, 0, $split), substr($wire, $split)]);
runIdentityExpect($parsed['tasks'] === [['private-task', ''], ['private-task', 'private-run']]
&& $parsed['terminals'] === [['private-task', 'stopped', 'private-run']],
'every byte split retains one original task announcement and one immutable run enrichment');
runIdentityExpect($parsed['state']['content'] === '诊断正文' && $parsed['state']['finished']
&& $parsed['state']['upstream_terminal_confirmed'], 'fragmented CRLF, UTF-8, reasoning and final terminal keep exact delivery');
}
$parsed = runIdentityParse([], str_split($wire, 1));
runIdentityExpect(count($parsed['tasks']) === 2 && count($parsed['terminals']) === 1, 'single-byte fragmentation is idempotent');
foreach (['top', 'nested', 'both'] as $startShape) {
foreach (['top', 'nested', 'both'] as $endShape) {
$shapedStart = $start; $shapedEnd = $terminal;
if ($startShape === 'top') unset($shapedStart['data']['id']);
if ($startShape === 'nested') unset($shapedStart['workflow_run_id']);
if ($endShape === 'top') unset($shapedEnd['data']['id']);
if ($endShape === 'nested') unset($shapedEnd['workflow_run_id']);
$parsed = runIdentityParse([$shapedStart, $shapedEnd, $shapedEnd]);
runIdentityExpect($parsed['state']['workflow_run_id'] === 'private-run'
&& $parsed['terminals'] === [['private-task', 'stopped', 'private-run']], 'start/finish run aliases agree and duplicate terminals are idempotent');
}
}
$invalidIds = [null, 1, false, [], new stdClass(), '', str_repeat('a', 129), 'bad/path', 'https://invalid.test/run',
"private-run\n", "private-run\r", ' private-run', 'private-run ', 'private.run', '运行'];
foreach ($invalidIds as $invalid) {
foreach (['top', 'nested', 'good_top_bad_nested', 'bad_top_good_nested'] as $shape) {
$badEnd = $terminal;
if ($shape === 'top') { $badEnd['workflow_run_id'] = $invalid; unset($badEnd['data']['id']); }
elseif ($shape === 'nested') { $badEnd['data']['id'] = $invalid; unset($badEnd['workflow_run_id']); }
elseif ($shape === 'good_top_bad_nested') $badEnd['data']['id'] = $invalid;
else $badEnd['workflow_run_id'] = $invalid;
$parsed = runIdentityParse([$start, $badEnd]);
runIdentityExpect(!$parsed['state']['upstream_terminal_confirmed'] && $parsed['terminals'] === [], 'malformed terminal aliases cannot confirm a run-bound task');
$badStart = $badEnd; $badStart['event'] = 'workflow_started';
$parsed = runIdentityParse([$badStart, ['event' => 'workflow_finished', 'task_id' => 'private-task', 'data' => ['status' => 'succeeded']]]);
runIdentityExpect($parsed['state']['workflow_run_id'] === '' && !$parsed['state']['upstream_terminal_confirmed'], 'malformed start aliases cannot downgrade into task-only terminal proof');
}
$badTaskEnd = $terminal; $badTaskEnd['task_id'] = $invalid;
$parsed = runIdentityParse([$start, $badTaskEnd]);
runIdentityExpect($parsed['terminals'] === [] && !$parsed['state']['upstream_terminal_confirmed'], 'terminal requires an explicit valid matching task');
$badTaskStart = $start; $badTaskStart['task_id'] = $invalid;
$parsed = runIdentityParse([$badTaskStart, $terminal]);
runIdentityExpect($parsed['tasks'] === [] && $parsed['state']['workflow_run_id'] === ''
&& !$parsed['state']['upstream_terminal_confirmed'], 'invalid start task never binds run and terminal never supplies missing task binding');
}
foreach (['task_id', 'workflow_run_id', 'all_run_fields'] as $missing) {
$missingEnd = $terminal;
if ($missing === 'all_run_fields') { unset($missingEnd['workflow_run_id'], $missingEnd['data']['id']); }
elseif ($missing === 'workflow_run_id') { unset($missingEnd['workflow_run_id'], $missingEnd['data']['id']); }
else unset($missingEnd['task_id']);
runIdentityExpect(!runIdentityParse([$start, $missingEnd])['state']['upstream_terminal_confirmed'], 'missing terminal identity cannot reuse the stored identity');
}
$noTaskStart = $start; unset($noTaskStart['task_id']);
$parsed = runIdentityParse([$noTaskStart, ['event' => 'message', 'task_id' => 'private-task'], $terminal]);
runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['terminals'] === [], 'start with absent task cannot bind a run indirectly through a later task frame');
$parsed = runIdentityParse([$terminal]);
runIdentityExpect($parsed['state']['task_id'] === '' && $parsed['tasks'] === [] && $parsed['terminals'] === [], 'terminal-only stream establishes no original identity');
$foreignStart = $start; $foreignStart['task_id'] = 'foreign-task'; $foreignStart['workflow_run_id'] = 'foreign-run'; $foreignStart['data']['id'] = 'foreign-run';
$conflictStart = $start; $conflictStart['workflow_run_id'] = 'foreign-run'; $conflictStart['data']['id'] = 'foreign-run';
$conflictEnd = $terminal; $conflictEnd['workflow_run_id'] = 'foreign-run'; $conflictEnd['data']['id'] = 'foreign-run';
$aliasConflict = $terminal; $aliasConflict['data']['id'] = 'foreign-run';
$foreignEnd = $terminal; $foreignEnd['task_id'] = 'foreign-task';
$parsed = runIdentityParse([$start, $foreignStart, $conflictStart, $foreignEnd, $conflictEnd, $aliasConflict,
['event' => 'message', 'task_id' => 'foreign-task', 'answer' => 'foreign-text']]);
runIdentityExpect($parsed['state']['task_id'] === 'private-task' && $parsed['state']['workflow_run_id'] === 'private-run'
&& count($parsed['tasks']) === 2 && $parsed['terminals'] === [] && $parsed['deltas'] === [], 'foreign/conflicting start, terminal and text cannot replace or confirm the original pair');
$parsed = runIdentityParse([$start, $conflictStart, $conflictEnd, $terminal]);
runIdentityExpect($parsed['terminals'] === [['private-task', 'stopped', 'private-run']], 'later exact original terminal can still prove the immutable original pair');
$mismatchedStart = $start; $mismatchedStart['data']['id'] = 'foreign-run';
runIdentityExpect(runIdentityParse([$mismatchedStart])['state']['workflow_run_id'] === '', 'start aliases must agree');
$notArrayStart = $start; $notArrayStart['data'] = 'private-malformed';
runIdentityExpect(runIdentityParse([$notArrayStart, $terminal])['terminals'] === [], 'malformed start data does not hide behind a valid top-level run');
$notArrayEnd = $terminal; $notArrayEnd['data'] = 'private-malformed';
runIdentityExpect(runIdentityParse([$start, $notArrayEnd])['terminals'] === [], 'malformed terminal data cannot confirm');
$legacyStart = ['event' => 'workflow_started', 'task_id' => 'private-task', 'data' => ['workflow_id' => 'private-definition']];
$legacyEnd = ['event' => 'workflow_finished', 'task_id' => 'private-task', 'data' => ['status' => 'succeeded']];
$parsed = runIdentityParse([$legacyStart, $legacyEnd]);
runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['tasks'] === [['private-task', '']]
&& $parsed['terminals'] === [['private-task', 'succeeded', '']], 'definition workflow_id never becomes run identity; explicit task-only legacy stream remains compatible');
$parsed = runIdentityParse([$legacyStart, $legacyEnd, $start, $terminal]);
runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['tasks'] === [['private-task', '']]
&& $parsed['terminals'] === [['private-task', 'succeeded', '']], 'start after a proved legacy terminal cannot reopen or enrich a released task');
foreach ($invalidIds as $invalid) {
$badLegacy = $legacyEnd; $badLegacy['task_id'] = $invalid;
runIdentityExpect(runIdentityParse([$legacyStart, $badLegacy])['terminals'] === [], 'legacy terminal never infers an invalid task');
}
$badLegacy = $legacyEnd; unset($badLegacy['task_id']);
runIdentityExpect(runIdentityParse([$legacyStart, $badLegacy])['terminals'] === [], 'legacy terminal never infers an absent task');
$parsed = runIdentityParse([
['event' => 'message', 'task_id' => 'private-task', 'workflow_run_id' => 'private-untrusted', 'answer' => '正文'],
['event' => 'node_started', 'task_id' => 'private-task', 'data' => ['id' => 'private-node']], $terminal]);
runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['terminals'] === [], 'message/node IDs and terminal run cannot invent the original workflow_started pair');
foreach (['succeeded', 'failed', 'stopped', 'cancelled', 'SUCCEEDED', 'running', 'unknown', '', null, ['succeeded']] as $status) {
$statusEnd = $terminal; $statusEnd['data']['status'] = $status;
$parsed = runIdentityParse([$start, $statusEnd]);
$expected = is_string($status) && in_array(strtolower($status), ['succeeded', 'failed', 'stopped', 'cancelled'], true);
runIdentityExpect($parsed['state']['upstream_terminal_confirmed'] === $expected, 'only terminal status whitelist confirms');
if ($status === 'cancelled') runIdentityExpect($parsed['state']['upstream_terminal_status'] === 'stopped', 'cancelled normalizes to stopped');
}
$parsed = runIdentityParse([$start, ['event' => 'message_end']], null, ['discard_deltas' => true, 'client_aborted' => true]);
runIdentityExpect($parsed['state']['finished'] && !$parsed['state']['upstream_terminal_confirmed'], 'message_end plus EOF is separate from real workflow terminal');
$parsed = runIdentityParse([], [runIdentityFrame($legacyStart), "data: [DONE]\r\n\r\n"]);
runIdentityExpect($parsed['state']['finished'] && !$parsed['state']['upstream_terminal_confirmed'], 'DONE plus EOF is separate from real workflow terminal');
$parsed = runIdentityParse([['event' => 'message_end'], $start,
['event' => 'message', 'answer' => 'private-late-text'], $terminal], null,
['discard_deltas' => true, 'client_aborted' => true]);
runIdentityExpect($parsed['state']['client_aborted'] && $parsed['deltas'] === [] && $parsed['state']['content'] === ''
&& $parsed['terminals'] === [['private-task', 'stopped', 'private-run']], 'late start and exact terminal survive completion/cancellation drain while text stays discarded');
$maxStart = $start; $maxStart['task_id'] = str_repeat('t', 128); $maxStart['workflow_run_id'] = str_repeat('r', 128); $maxStart['data']['id'] = str_repeat('r', 128);
$maxEnd = $maxStart; $maxEnd['event'] = 'workflow_finished'; $maxEnd['data']['status'] = 'succeeded';
runIdentityExpect(runIdentityParse([$maxStart, $maxEnd])['state']['upstream_terminal_confirmed'], 'bounded IDs accept exactly the maximum length');
foreach (['initial', 'enrichment', 'terminal'] as $failure) {
$observedTasks = []; $observedTerminals = [];
$parsed = runIdentityParse([$start, $start, $conflictStart, $foreignStart, $terminal, $terminal], null, [
'on_task' => static function (string $task, string $run = '') use ($failure, &$observedTasks): void {
$observedTasks[] = [$task, $run];
if (($failure === 'initial' && $run === '') || ($failure === 'enrichment' && $run !== '')) throw new RuntimeException('private-observer-error');
},
'on_terminal' => static function () use ($failure, &$observedTerminals): void {
$observedTerminals[] = true;
if ($failure === 'terminal') throw new RuntimeException('private-terminal-error');
},
]);
runIdentityExpect($observedTasks === [['private-task', ''], ['private-task', 'private-run']],
'initial observer failure still receives one original run enrichment; repeated/foreign/conflicting starts never rebind or retry');
runIdentityExpect(count($observedTerminals) === ($failure === 'terminal' ? 1 : 0),
'failed identity observer never reaches terminal callback and failed terminal observer is not retried');
runIdentityExpect($parsed['state']['callback_error'] && !$parsed['state']['upstream_terminal_confirmed']
&& $parsed['state']['task_id'] === 'private-task' && $parsed['state']['workflow_run_id'] === 'private-run'
&& !isset($parsed['timing']->snapshot()['workflow_terminal_ms']), 'observer exception fails closed, retains known pair and invents no terminal timing');
}
foreach (['late_known', 'late_unknown', 'accepted_only', 'absolute_bound', 'observer_failure', 'initial_observer_failure'] as $mode) {
$GLOBALS['runIdentityMode'] = $mode; $GLOBALS['runIdentityNow'] = 1000;
$GLOBALS['runIdentityCancelled'] = false; $GLOBALS['runIdentityClosed'] = false;
$GLOBALS['runIdentityStopIo'] = 0; $GLOBALS['runIdentityInterruptedAt'] = null;
$tasks = []; $terminals = []; $deltas = []; $safeTiming = []; $accepted = false; $stopNotification = null;
$result = runIdentityPrivate('sendStreamRequest', ['dify', 'https://private.invalid/chat-messages',
['user' => 'private-user', 'query' => 'private-query'], 'private-api-key', 5,
static function (string $delta) use (&$deltas): bool { $deltas[] = $delta; return true; },
static fn (): bool => $GLOBALS['runIdentityCancelled'], [
'on_task' => static function (string $task, string $run = '') use (&$tasks, &$accepted, $mode): void {
$tasks[] = [$task, $run];
if ($run !== '' && $mode === 'late_known') runIdentityExpect($accepted, 'same-task run enrichment preserves prior accepted stop');
if ($run !== '' && $mode === 'observer_failure') throw new RuntimeException('private-observer-error');
if ($mode === 'initial_observer_failure') throw new RuntimeException('UPSTREAM_TASK_HANDOFF_BLOCKED');
},
'on_cancel_task' => static function (string $task, callable $stop) use (&$accepted): array {
$GLOBALS['runIdentityInterruptedAt'] ??= $GLOBALS['runIdentityNow'];
if (!$accepted) $accepted = $stop();
return ['accepted' => $accepted];
},
'on_terminal' => static function (string $task, string $status, string $run = '') use (&$terminals): void { $terminals[] = [$task, $status, $run]; },
'on_stop' => static function (bool $accepted, string $task) use (&$stopNotification): void { $stopNotification = $accepted; },
'on_upstream_timing' => static function (DifyStreamTiming $timing) use (&$safeTiming): void { $safeTiming = $timing->snapshot(); },
]]);
runIdentityExpect($GLOBALS['runIdentityClosed'] && $GLOBALS['runIdentityStopIo'] === 1
&& $stopNotification === true && $deltas === [] && $result['content'] === '', 'stubbed transport stops exact task once, closes, and discards cancelled text');
runIdentityExpect($tasks === [['private-task', ''], ['private-task', 'private-run']], 'transport captures late original pair and enriches exactly once');
$shouldConfirm = in_array($mode, ['late_known', 'late_unknown'], true);
runIdentityExpect($result['upstream_terminal_confirmed'] === $shouldConfirm
&& count($terminals) === ($shouldConfirm ? 1 : 0), 'accepted stop, EOF, local abort or observer failure cannot impersonate exact terminal');
if ($mode === 'absolute_bound') runIdentityExpect($safeTiming['duration_ms'] <= 2511
&& $safeTiming['transport_exit'] === 'local_abort' && !isset($safeTiming['natural_eof_ms']), 'run enrichment adds no new cleanup window');
if (in_array($mode, ['observer_failure', 'initial_observer_failure'], true)) {
runIdentityExpect($result['callback_error'], 'initial/enrichment observer failure stops known task and keeps terminal unconfirmed');
}
$formatted = runIdentityPrivate('formatStreamResponse', [$result, microtime(true)]);
runIdentityPrivate('logUpstreamFailure', [['protocol' => 'dify', 'url' => 'https://private.invalid/chat-messages'],
$result, 'private-query', [], $formatted]);
runIdentityExpect(!str_contains((string) json_encode([$result, $formatted, $safeTiming, Log::$lines]), 'private-')
&& !array_key_exists('task_id', $result) && !array_key_exists('workflow_run_id', $result), 'private task/run/user/host/key/body/error never enter reply or log/timing allowlists');
}
echo "Dify workflow run identity: {$runIdentityAssertions} assertions passed (pure parser and in-process transport; no network).\n";
}