104 lines
7.0 KiB
PHP
104 lines
7.0 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\common\service {
|
|
require __DIR__ . '/fixtures/CancellableCurlMultiStub.php';
|
|
function config(string $name): array { return $GLOBALS['timeoutCallbackConfig']; }
|
|
function curl_init(): \stdClass { return (object) ['errno' => 0]; }
|
|
function curl_setopt_array(\stdClass $handle, array $options): bool {
|
|
$handle->options = $options; $handle->url = $options[CURLOPT_URL]; return true;
|
|
}
|
|
function curl_exec(\stdClass $handle): string|bool {
|
|
$GLOBALS['timeoutCallbackRequests'][] = $handle->url;
|
|
if (str_ends_with($handle->url, '/stop')) {
|
|
\callbackExpect(!$GLOBALS['timeoutGenerationClosed'], 'stop executes before releasing the generation handle');
|
|
\callbackExpect(json_decode($handle->options[CURLOPT_POSTFIELDS], true) === ['user' => 'timeout-callback-user'],
|
|
'stop reuses the original user');
|
|
return '{"result":"success"}';
|
|
}
|
|
$header = $handle->options[CURLOPT_HEADERFUNCTION];
|
|
$write = $handle->options[CURLOPT_WRITEFUNCTION];
|
|
$header($handle, "HTTP/1.1 200 OK\r\n");
|
|
$frame = "data: {\"event\":\"workflow_started\",\"task_id\":\"timeout-callback-task\"}";
|
|
$write($handle, $frame . ($GLOBALS['timeoutCallbackMode'] === 'late_frame' ? '' : "\n\n"));
|
|
if ($GLOBALS['timeoutCallbackMode'] !== 'task_observer') $handle->errno = CURLE_OPERATION_TIMEDOUT;
|
|
return false;
|
|
}
|
|
function curl_errno(\stdClass $handle): int { return $handle->errno; }
|
|
function curl_getinfo(\stdClass $handle, int $option): int { return 200; }
|
|
function curl_close(\stdClass $handle): void {
|
|
if (!str_ends_with($handle->url, '/stop')) $GLOBALS['timeoutGenerationClosed'] = true;
|
|
}
|
|
}
|
|
|
|
namespace {
|
|
require dirname(__DIR__) . '/vendor/autoload.php';
|
|
use app\adminapi\service\AssistantRequestControl;
|
|
use app\common\service\DifyChatService;
|
|
|
|
function callbackExpect(bool $ok, string $message): void {
|
|
if (!$ok) throw new RuntimeException($message);
|
|
}
|
|
$timeoutCallbackConfig = ['enable' => true, 'base_url' => 'https://synthetic.invalid/v1/chat-messages',
|
|
'timeout' => 5, 'models' => ['qwen' => ['name' => 'synthetic', 'api_key' => 'fixture-only']]];
|
|
$directory = sys_get_temp_dir() . '/zyt-timeout-callback-' . bin2hex(random_bytes(8));
|
|
try {
|
|
foreach (['hard_timeout', 'late_frame', 'timeout_observer', 'task_observer', 'stop_observer',
|
|
'end_observer', 'coordination_observer', 'abort_observer'] as $index => $timeoutCallbackMode) {
|
|
$timeoutCallbackRequests = []; $timeoutGenerationClosed = false; $deltas = [];
|
|
$id = sprintf('ffabcdef-1234-4234-8234-%012d', $index);
|
|
$control = new AssistantRequestControl(7, 11, $id, $directory);
|
|
callbackExpect($control->begin(), 'fresh synthetic control begins');
|
|
$result = DifyChatService::streamChat('qwen', [], '合成超时回归', 'timeout-callback-user',
|
|
static function (string $text) use (&$deltas): bool { $deltas[] = $text; return true; },
|
|
$timeoutCallbackMode === 'abort_observer' ? static function () use ($control): bool {
|
|
if ($control->publicState()['timeout_reached']) throw new RuntimeException('synthetic abort observer');
|
|
return false;
|
|
} : null, [], ['knowledge_mode' => 'evidence',
|
|
'on_task' => static function (string $taskId) use ($control, $timeoutCallbackMode): void {
|
|
$control->bindTask('qwen', 'timeout-callback-user', $taskId);
|
|
if ($timeoutCallbackMode === 'task_observer') throw new RuntimeException('synthetic task observer');
|
|
},
|
|
'on_timeout' => static function () use ($control, $timeoutCallbackMode): void {
|
|
$control->signalTimeout();
|
|
if ($timeoutCallbackMode === 'timeout_observer') throw new RuntimeException('synthetic timeout observer');
|
|
},
|
|
'on_cancel_task' => static function (string $taskId, callable $stop, string $reason) use ($control, $timeoutCallbackMode): array {
|
|
if ($reason === 'timeout') $control->signalTimeout(); else $control->signalCancel();
|
|
$outcome = $control->stopBoundTask($stop, $taskId);
|
|
if ($timeoutCallbackMode === 'coordination_observer') throw new RuntimeException('synthetic lease observer');
|
|
return $outcome;
|
|
},
|
|
'on_stop' => static function (bool $accepted, string $taskId) use ($control, $timeoutCallbackMode): void {
|
|
$control->markStop($accepted, $taskId);
|
|
if ($timeoutCallbackMode === 'stop_observer') throw new RuntimeException('synthetic stop observer');
|
|
},
|
|
'on_transfer_end' => static function (string $taskId) use ($control, $timeoutCallbackMode): void {
|
|
$control->clearTask($taskId !== '' ? $taskId : null);
|
|
if ($timeoutCallbackMode === 'end_observer') throw new RuntimeException('synthetic end observer');
|
|
},
|
|
]);
|
|
$control->finish('error');
|
|
$state = $control->publicState();
|
|
callbackExpect(count($timeoutCallbackRequests) === 2 && str_ends_with($timeoutCallbackRequests[1], '/timeout-callback-task/stop'),
|
|
$timeoutCallbackMode . ': known task stops once despite observer errors');
|
|
callbackExpect($deltas === [] && $timeoutGenerationClosed && $state['upstream_stop_confirmed']
|
|
&& !$state['upstream_terminal_confirmed'], 'transport closes without inventing workflow termination');
|
|
callbackExpect(($result['error_code'] ?? '') === ($timeoutCallbackMode === 'task_observer' ? 'AI_ASSISTANT_CANCELLED' : 'UPSTREAM_TIMEOUT'),
|
|
$timeoutCallbackMode . ': timeout remains timeout even when cleanup callbacks throw');
|
|
$private = json_decode((string) file_get_contents($directory . '/' . hash('sha256', '7:11:' . $id) . '.json'), true);
|
|
callbackExpect(($private['upstream']['task_id'] ?? '') === 'timeout-callback-task', 'unconfirmed task survives all cleanup paths');
|
|
$duplicate = 0;
|
|
$control->stopBoundTask(static function () use (&$duplicate): bool { ++$duplicate; return true; });
|
|
callbackExpect($duplicate === 0, 'accepted shared lease prevents a duplicate stop after observer failure');
|
|
if ($timeoutCallbackMode !== 'task_observer') callbackExpect(!$control->shouldAbort() && $state['status'] === 'error', 'ordinary timeout does not create cancellation intent');
|
|
}
|
|
echo "Diagnosis AI hard timeout / final task frame / observer exceptions / stop idempotence: OK\n";
|
|
} finally {
|
|
foreach (glob($directory . '/*') ?: [] as $path) unlink($path);
|
|
if (is_file($directory . '/.gc.lock')) unlink($directory . '/.gc.lock');
|
|
if (is_dir($directory)) rmdir($directory);
|
|
}
|
|
}
|