← All changes | app/Services/Integrations/FluentBot/FluentBotAPI.php +143 -58 2.3.2 → 2.4.5 View file →

		@@ -54,9 +54,8 @@
54	54	$tokenUsage = $responseBody['token_usage'] ?? [];
55	55	$totalTokens = ($tokenUsage['input_tokens'] ?? 0) + ($tokenUsage['output_tokens'] ?? 0);
56	56	do_action('fluent_support/ai_response_success', $ticketId, $prompt, $totalTokens, "FluentBot");
57	57
58		- // Return both content and chat_id if available
59	58	return [
60	59	'content' => $content,
61	60	'chat_id' => $responseBody['chat_id'] ?? null
62	61	];
		@@ -65,10 +64,9 @@
65	64	public function makeStreamRequest(int $ticketId, $prompt, array $args = [])
66	65	{
67	66	$timeout = apply_filters('fs_ai_request_timeout', 120);
68	67
69		- // Use cURL for streaming
70		- // Note: Using cURL here because WordPress HTTP API doesn't support streaming SSE responses
	68	+ // Raw cURL because the WP HTTP API can't stream SSE.
71	69	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init
72	70	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt
73	71	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec
74	72	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close
		@@ -89,64 +87,16 @@
89	87	$conversationId = null;
90	88	$streamTokens = 0;
91	89
92	90	curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer, &$conversationId, &$streamTokens) {
93		- $buffer .= $data;
94		-
95		- // Process complete SSE events from the AI API
96		- $events = explode("\n\n", $buffer);
97		- $buffer = array_pop($events); // Keep incomplete event in buffer
98		-
99		- foreach ($events as $event) {
100		- if (trim($event)) {
101		- $lines = explode("\n", $event);
102		- $eventType = '';
103		- $eventId = '';
104		- $eventDataLines = [];
105		-
106		- foreach ($lines as $line) {
107		- if (strpos($line, 'event: ') === 0) {
108		- $eventType = trim((string)substr($line, 7));
109		- } elseif (strpos($line, 'id: ') === 0) {
110		- $eventId = trim((string)substr($line, 4));
111		- } elseif (strpos($line, 'data: ') === 0) {
112		- $eventDataLines[] = (string)substr($line, 6);
113		- }
	91	+ foreach ($this->drainSseChunk($data, $buffer) as $parsed) {
	92	+ if ($parsed['event'] === 'chat_id' && !empty($parsed['data'])) {
	93	+ $conversationId = $parsed['data'][0];
	94	+ } elseif ($parsed['event'] === 'token_usage' && !empty($parsed['data'])) {
	95	+ $usage = json_decode($parsed['data'][0], true);
	96	+ if (is_array($usage)) {
	97	+ $streamTokens = ($usage['input_tokens'] ?? 0) + ($usage['output_tokens'] ?? 0);
114	98	}
115		-
116		- // Forward the event to the browser with proper formatting
117		- if ($eventType) {
118		- echo "event: ".esc_html($eventType)."\n";
119		-
120		- // Include ID if present
121		- if ($eventId !== '') {
122		- echo "id: ".esc_html($eventId)."\n";
123		- }
124		-
125		- // Handle multiple data lines properly
126		- if (!empty($eventDataLines)) {
127		- foreach ($eventDataLines as $dataLine) {
128		- // phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- raw SSE payload from trusted bot endpoint; rendered output sanitized client-side
129		- echo "data: ".$dataLine."\n";
130		- }
131		- } else {
132		- echo "data: \n";
133		- }
134		-
135		- echo "\n";
136		-
137		- // Store chat_id for later use
138		- if ($eventType === 'chat_id' && !empty($eventDataLines)) {
139		- $conversationId = $eventDataLines[0];
140		- } elseif ($eventType === 'token_usage' && !empty($eventDataLines)) {
141		- $usage = json_decode($eventDataLines[0], true);
142		- if (is_array($usage)) {
143		- $streamTokens = ($usage['input_tokens'] ?? 0) + ($usage['output_tokens'] ?? 0);
144		- }
145		- }
146		-
147		- flush();
148		- }
149	99	}
150	100	}
151	101
152	102	return strlen($data);
		@@ -176,8 +126,143 @@
176	126
177	127	if ($httpCode === 200) {
178	128	do_action('fluent_support/ai_response_success', $ticketId, $prompt, $streamTokens, "FluentBot");
179	129	}
	130	+ }
	131	+
	132	+ /**
	133	+ * Forward every complete SSE frame in $data to the browser, return the parsed frames.
	134	+ * $buffer holds the partial trailing frame across cURL writes, so pass it by reference.
	135	+ *
	136	+ * @param string $data raw bytes from cURL
	137	+ * @param string $buffer incomplete frame carried over from the previous write
	138	+ * @return array<int, array{event: string, data: array<int, string>}>
	139	+ */
	140	+ private function drainSseChunk(string $data, string &$buffer): array
	141	+ {
	142	+ $buffer .= $data;
	143	+
	144	+ $events = explode("\n\n", $buffer);
	145	+ $buffer = array_pop($events); // Keep incomplete event in buffer
	146	+
	147	+ $parsed = [];
	148	+
	149	+ foreach ($events as $event) {
	150	+ if (!trim($event)) {
	151	+ continue;
	152	+ }
	153	+
	154	+ // Forward keepalive comments as-is so proxies don't hit an idle timeout.
	155	+ if (strpos(ltrim($event), ':') === 0) {
	156	+ echo $event . "\n\n";
	157	+ flush();
	158	+ continue;
	159	+ }
	160	+
	161	+ $lines = explode("\n", $event);
	162	+ $eventType = '';
	163	+ $eventId = '';
	164	+ $eventDataLines = [];
	165	+
	166	+ foreach ($lines as $line) {
	167	+ if (strpos($line, 'event: ') === 0) {
	168	+ $eventType = trim((string)substr($line, 7));
	169	+ } elseif (strpos($line, 'id: ') === 0) {
	170	+ $eventId = trim((string)substr($line, 4));
	171	+ } elseif (strpos($line, 'data: ') === 0) {
	172	+ $eventDataLines[] = (string)substr($line, 6);
	173	+ }
	174	+ }
	175	+
	176	+ if (!$eventType) {
	177	+ continue;
	178	+ }
	179	+
	180	+ echo "event: ".esc_html($eventType)."\n";
	181	+
	182	+ if ($eventId !== '') {
	183	+ echo "id: ".esc_html($eventId)."\n";
	184	+ }
	185	+
	186	+ if (!empty($eventDataLines)) {
	187	+ foreach ($eventDataLines as $dataLine) {
	188	+ // phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- raw SSE payload from trusted bot endpoint; rendered output sanitized client-side
	189	+ echo "data: ".$dataLine."\n";
	190	+ }
	191	+ } else {
	192	+ echo "data: \n";
	193	+ }
	194	+
	195	+ echo "\n";
	196	+ flush();
	197	+
	198	+ $parsed[] = ['event' => $eventType, 'data' => $eventDataLines];
	199	+ }
	200	+
	201	+ return $parsed;
	202	+ }
	203	+
	204	+ /**
	205	+ * Reconnect to an in-flight turn: replay its buffered SSE and tail it live.
	206	+ * Same wire format as makeStreamRequest. Upstream ends on `__done__` or an `idle` frame.
	207	+ */
	208	+ public function makeResumeStreamRequest()
	209	+ {
	210	+ // Raw cURL because the WP HTTP API can't stream SSE.
	211	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init
	212	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt
	213	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec
	214	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close
	215	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_error
	216	+ // PluginCheck:ignoreFile
	217	+ $ch = curl_init();
	218	+ curl_setopt($ch, CURLOPT_URL, $this->apiUrl);
	219	+ curl_setopt($ch, CURLOPT_HTTPGET, true);
	220	+
	221	+ $streamHeaders = ['Accept: text/event-stream'];
	222	+ if ($this->apiKey !== '') {
	223	+ $streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey;
	224	+ }
	225	+ curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders);
	226	+
	227	+ $buffer = '';
	228	+
	229	+ curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer) {
	230	+ // Stop when the agent leaves. The turn itself is persisted upstream.
	231	+ if (connection_aborted()) {
	232	+ return 0;
	233	+ }
	234	+
	235	+ $this->drainSseChunk($data, $buffer);
	236	+
	237	+ return strlen($data);
	238	+ });
	239	+
	240	+ // connection_aborted() can lag behind proxies and buffering, so cap the tail
	241	+ // or a refresh mid-turn can pin a worker. The client refetches a cut tail.
	242	+ curl_setopt($ch, CURLOPT_TIMEOUT, apply_filters('fs_ai_resume_timeout', 120));
	243	+ curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 15);
	244	+ // Upstream sends keepalives during quiet gaps, so a stall here means it's dead.
	245	+ curl_setopt($ch, CURLOPT_LOW_SPEED_LIMIT, 1);
	246	+ curl_setopt($ch, CURLOPT_LOW_SPEED_TIME, apply_filters('fs_ai_resume_stall_timeout', 60));
	247	+ curl_setopt($ch, CURLOPT_TCP_KEEPALIVE, 1);
	248	+ curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
	249	+ curl_setopt($ch, CURLOPT_BUFFERSIZE, 128); // Smaller buffer for faster streaming
	250	+
	251	+ curl_exec($ch);
	252	+
	253	+ if (curl_error($ch)) {
	254	+ echo "event: error\n";
	255	+ echo "data: " . wp_json_encode(['error' => curl_error($ch)]) . "\n\n";
	256	+ flush();
	257	+ }
	258	+
	259	+ curl_close($ch);
	260	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_init
	261	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_setopt
	262	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_exec
	263	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_close
	264	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_error
180	265	}
181	266
182	267	protected function sendRequest(array $payload)
183	268	{