← All changes | app/Services/Integrations/FluentBot/FluentBotAPI.php +153 -55 2.3.2 → 2.4.0 View file →

		@@ -89,64 +89,17 @@
89	89	$conversationId = null;
90	90	$streamTokens = 0;
91	91
92	92	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		- }
	93	+ foreach ($this->drainSseChunk($data, $buffer) as $parsed) {
	94	+ // Store chat_id for later use
	95	+ if ($parsed['event'] === 'chat_id' && !empty($parsed['data'])) {
	96	+ $conversationId = $parsed['data'][0];
	97	+ } elseif ($parsed['event'] === 'token_usage' && !empty($parsed['data'])) {
	98	+ $usage = json_decode($parsed['data'][0], true);
	99	+ if (is_array($usage)) {
	100	+ $streamTokens = ($usage['input_tokens'] ?? 0) + ($usage['output_tokens'] ?? 0);
114	101	}
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	102	}
150	103	}
151	104
152	105	return strlen($data);
		@@ -176,8 +129,153 @@
176	129
177	130	if ($httpCode === 200) {
178	131	do_action('fluent_support/ai_response_success', $ticketId, $prompt, $streamTokens, "FluentBot");
179	132	}
	133	+ }
	134	+
	135	+ /**
	136	+ * Forward every complete SSE frame in $data to the browser, return the parsed frames.
	137	+ * $buffer holds the partial trailing frame across cURL writes, so pass it by reference.
	138	+ *
	139	+ * @param string $data raw bytes from cURL
	140	+ * @param string $buffer incomplete frame carried over from the previous write
	141	+ * @return array<int, array{event: string, data: array<int, string>}>
	142	+ */
	143	+ private function drainSseChunk(string $data, string &$buffer): array
	144	+ {
	145	+ $buffer .= $data;
	146	+
	147	+ // Process complete SSE events from the AI API
	148	+ $events = explode("\n\n", $buffer);
	149	+ $buffer = array_pop($events); // Keep incomplete event in buffer
	150	+
	151	+ $parsed = [];
	152	+
	153	+ foreach ($events as $event) {
	154	+ if (!trim($event)) {
	155	+ continue;
	156	+ }
	157	+
	158	+ // SSE comment/keepalive (starts ":"): forward raw so proxies keep seeing bytes, no idle-timeout.
	159	+ if (strpos(ltrim($event), ':') === 0) {
	160	+ echo $event . "\n\n";
	161	+ flush();
	162	+ continue;
	163	+ }
	164	+
	165	+ $lines = explode("\n", $event);
	166	+ $eventType = '';
	167	+ $eventId = '';
	168	+ $eventDataLines = [];
	169	+
	170	+ foreach ($lines as $line) {
	171	+ if (strpos($line, 'event: ') === 0) {
	172	+ $eventType = trim((string)substr($line, 7));
	173	+ } elseif (strpos($line, 'id: ') === 0) {
	174	+ $eventId = trim((string)substr($line, 4));
	175	+ } elseif (strpos($line, 'data: ') === 0) {
	176	+ $eventDataLines[] = (string)substr($line, 6);
	177	+ }
	178	+ }
	179	+
	180	+ // Forward the event to the browser with proper formatting
	181	+ if (!$eventType) {
	182	+ continue;
	183	+ }
	184	+
	185	+ echo "event: ".esc_html($eventType)."\n";
	186	+
	187	+ // Include ID if present
	188	+ if ($eventId !== '') {
	189	+ echo "id: ".esc_html($eventId)."\n";
	190	+ }
	191	+
	192	+ // Handle multiple data lines properly
	193	+ if (!empty($eventDataLines)) {
	194	+ foreach ($eventDataLines as $dataLine) {
	195	+ // phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- raw SSE payload from trusted bot endpoint; rendered output sanitized client-side
	196	+ echo "data: ".$dataLine."\n";
	197	+ }
	198	+ } else {
	199	+ echo "data: \n";
	200	+ }
	201	+
	202	+ echo "\n";
	203	+ flush();
	204	+
	205	+ $parsed[] = ['event' => $eventType, 'data' => $eventDataLines];
	206	+ }
	207	+
	208	+ return $parsed;
	209	+ }
	210	+
	211	+ /**
	212	+ * Reconnect to an in-flight turn: replay its buffered SSE and tail it live. Same wire
	213	+ * format as makeStreamRequest. Upstream ends on `__done__` or an `idle` frame; a total
	214	+ * cap still applies — see the timeout block below.
	215	+ */
	216	+ public function makeResumeStreamRequest()
	217	+ {
	218	+ // Use cURL for streaming
	219	+ // Note: Using cURL here because WordPress HTTP API doesn't support streaming SSE responses
	220	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init
	221	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt
	222	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec
	223	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close
	224	+ // phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_error
	225	+ // PluginCheck:ignoreFile
	226	+ $ch = curl_init();
	227	+ curl_setopt($ch, CURLOPT_URL, $this->apiUrl);
	228	+ curl_setopt($ch, CURLOPT_HTTPGET, true);
	229	+
	230	+ $streamHeaders = ['Accept: text/event-stream'];
	231	+ if ($this->apiKey !== '') {
	232	+ $streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey;
	233	+ }
	234	+ curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders);
	235	+
	236	+ $buffer = '';
	237	+
	238	+ curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer) {
	239	+ // Stop as soon as the agent navigates away — nothing here needs to
	240	+ // outlive the client, the turn itself is persisted upstream.
	241	+ if (connection_aborted()) {
	242	+ return 0;
	243	+ }
	244	+
	245	+ $this->drainSseChunk($data, $buffer);
	246	+
	247	+ return strlen($data);
	248	+ });
	249	+
	250	+ // Bounded on purpose: connection_aborted() can lag (abort travels browser ->
	251	+ // proxy -> fluent-bot, Apache buffers writes), so a refresh mid-turn can leave
	252	+ // the old tail pinning a worker. The cap makes that self-limiting — the client
	253	+ // treats a cut tail as a body ended without __done__ and refetches.
	254	+ curl_setopt($ch, CURLOPT_TIMEOUT, apply_filters('fs_ai_resume_timeout', 120));
	255	+ curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 15);
	256	+ // Abort sooner when genuinely stalled: the upstream sends ': keepalive'
	257	+ // during quiet gaps, so a healthy tail always beats this window.
	258	+ curl_setopt($ch, CURLOPT_LOW_SPEED_LIMIT, 1);
	259	+ curl_setopt($ch, CURLOPT_LOW_SPEED_TIME, apply_filters('fs_ai_resume_stall_timeout', 60));
	260	+ curl_setopt($ch, CURLOPT_TCP_KEEPALIVE, 1);
	261	+ curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
	262	+ curl_setopt($ch, CURLOPT_BUFFERSIZE, 128); // Smaller buffer for faster streaming
	263	+
	264	+ curl_exec($ch);
	265	+
	266	+ if (curl_error($ch)) {
	267	+ echo "event: error\n";
	268	+ echo "data: " . wp_json_encode(['error' => curl_error($ch)]) . "\n\n";
	269	+ flush();
	270	+ }
	271	+
	272	+ curl_close($ch);
	273	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_init
	274	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_setopt
	275	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_exec
	276	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_close
	277	+ // phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_error
180	278	}
181	279
182	280	protected function sendRequest(array $payload)
183	281	{