<?php
/**
 * Agent Stream Reader — SSE streaming for the OpenAI Responses API.
 *
 * Uses the Codex Responses endpoint for OAuth/Codex auth and the standard
 * OpenAI Responses endpoint for API key mode.
 *
 * @package Niwa_AI
 * @since   0.12.0
 */

if ( ! defined( 'ABSPATH' ) ) {
    exit;
}

class Niwa_AI_Agent_Stream_Reader {

    /** @var string OpenAI Responses API endpoint (API key mode). */
    const RESPONSES_API_URL = 'https://api.openai.com/v1/responses';

    /** @var string Codex Responses API endpoint (OAuth mode). */
    const CODEX_RESPONSES_URL = 'https://chatgpt.com/backend-api/codex/responses';

    /** @var string Codex originator header value (mirrors Niwa_AI_OpenAI_Client). */
    const CODEX_ORIGINATOR = 'codex_cli_rs';

    /** @var string Default Codex instructions when none provided. */
    const DEFAULT_CODEX_INSTRUCTIONS = 'You are Codex, a coding assistant.';

    /** @var int Minimum seconds between transport heartbeat callbacks. */
    const CURL_TRANSPORT_HEARTBEAT_INTERVAL = 15;

    /** @var string Model identifier. */
    private $model;

    /** @var array Messages array [{role, content}, ...]. */
    private $messages;

    /** @var array|null Tool definitions or null if no tools. */
    private $tools;

    /** @var string Auth mode: 'api_key', 'oauth', or 'codex'. */
    private $auth_mode;

    /** @var array Credentials array (api_key, access_token, codex_token, account_id, etc.). */
    private $creds;

    /** @var array Options: timeout, temperature, tool_choice, api_base_url. */
    private $options;

    /**
     * Constructor.
     *
     * @param string     $model     Model identifier (e.g. 'gpt-4o', 'gpt-5.3-codex-spark').
     * @param array      $messages  Messages array [{role, content}, ...].
     * @param array|null $tools     Tool definitions in Chat Completions-compatible format, or null.
     * @param string     $auth_mode Auth mode: 'api_key', 'oauth', or 'codex'.
     * @param array      $creds     Credentials array.
     * @param array      $options   Optional overrides: timeout, temperature, tool_choice, api_base_url.
     */
    public function __construct( $model, $messages, $tools, $auth_mode, $creds, $options = array() ) {
        $this->model     = (string) $model;
        $this->messages  = is_array( $messages ) ? $messages : array();
        $this->tools     = is_array( $tools ) && ! empty( $tools ) ? $tools : null;
        $this->auth_mode = (string) $auth_mode;
        $this->creds     = is_array( $creds ) ? $creds : array();
        $this->options   = is_array( $options ) ? $options : array();
    }

    /**
     * Execute a streaming request and read SSE events.
     *
     * @param callable|null $event_emitter Optional callback invoked on each content delta.
     *                                     Signature: function( string $event, array $data ).
     *                                     Events: 'message_update', 'tool_call_update', 'done'.
     * @return array {
     *     @type bool   $success           Whether the stream completed successfully.
     *     @type string $role              Always 'assistant'.
     *     @type string $content           Accumulated text content.
     *     @type array  $tool_calls        Assembled tool calls [{id, name, arguments}, ...].
     *     @type string $stop_reason       'stop', 'tool_calls', or 'length'.
     *     @type array  $usage             Token usage {prompt_tokens, completion_tokens, total_tokens}.
     *     @type array  $assistant_message Full assistant message suitable for appending to messages array.
     *     @type string $error             Error message (empty on success).
     *     @type string $error_type        Error category: 'http_error', 'stream_error', 'timeout', or ''.
     * }
     */
    public function stream( $event_emitter = null ) {
        $url     = $this->resolve_api_url();
        $timeout = isset( $this->options['timeout'] ) ? max( 10, intval( $this->options['timeout'] ) ) : 120;

        $headers_array = $this->build_http_headers();
        $request_body  = $this->build_request_body();
        $body_json     = wp_json_encode( $request_body, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE );
        $audit_id      = class_exists( 'Niwa_AI_LLM_Audit_Logger' ) ? Niwa_AI_LLM_Audit_Logger::new_event_id() : '';
        $audit_started = microtime( true );
        $timing        = array(
            'request_payload_bytes' => is_string( $body_json ) ? strlen( $body_json ) : 0,
            'transport'             => 'stream_wrapper',
        );

        if ( ! is_string( $body_json ) || '' === $body_json ) {
            $result = $this->error_result( 'Failed to encode request body.', 'stream_error' );
            $result = $this->add_transport_meta( $result, $audit_started, $timeout, 0 );
            $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
            return $result;
        }

        if ( $this->should_use_curl_stream_transport() ) {
            return $this->stream_with_curl_transport(
                $url,
                $headers_array,
                $body_json,
                $timeout,
                $event_emitter,
                $audit_id,
                $request_body,
                $audit_started
            );
        }

        // Build raw header string for stream_context.
        $header_lines = array();
        foreach ( $headers_array as $key => $value ) {
            $header_lines[] = $key . ': ' . $value;
        }

        $context = stream_context_create( array(
            'http' => array(
                'method'        => 'POST',
                'header'        => implode( "\r\n", $header_lines ),
                'content'       => $body_json,
                'timeout'       => $timeout,
                'ignore_errors' => true,
            ),
            'ssl'  => array(
                'verify_peer'      => true,
                'verify_peer_name' => true,
            ),
        ) );

        // Suppress warnings from fopen — we handle errors explicitly.
        $stream = @fopen( $url, 'r', false, $context );

        if ( ! $stream ) {
            $result = $this->error_result(
                'Failed to open stream connection to ' . $url,
                'http_error'
            );
            $result = $this->add_transport_meta( $result, $audit_started, $timeout, 0 );
            $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
            return $result;
        }

        // Check HTTP status from response headers.
        $meta = stream_get_meta_data( $stream );
        $http_status = $this->extract_http_status( $meta );

        if ( $http_status >= 400 ) {
            $error_body = stream_get_contents( $stream );
            fclose( $stream );
            $error_detail = $this->parse_api_error( $error_body, $http_status );
            $result = $this->error_result( $error_detail, 'http_error' );
            $result['http_status'] = $http_status;
            $result['raw_error_body'] = $error_body;
            $result = $this->add_transport_meta( $result, $audit_started, $timeout, $http_status );
            $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
            return $result;
        }

        $result = $this->read_responses_api_stream( $stream, $timeout, $event_emitter, $audit_started, $timing );
        $result = $this->add_transport_meta( $result, $audit_started, $timeout, $http_status );
        $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
        return $result;
    }

    /**
     * Whether to use cURL for streaming transport.
     *
     * PHP stream wrappers can remain blocked inside fopen()/fgets() on some
     * shared-hosting/LiteSpeed runtimes even when a logical timeout is passed.
     * cURL gives us a real transport-level wall timeout plus low-speed stall
     * detection, which is the safe default for autonomous workers.
     *
     * @return bool
     */
    private function should_use_curl_stream_transport() {
        if ( ! empty( $this->options['disable_curl_transport'] ) ) {
            return false;
        }

        return function_exists( 'curl_init' );
    }

    /**
     * Build hard transport timeout settings for cURL streaming.
     *
     * @param int $timeout Total timeout in seconds.
     * @return array{timeout:int,connect_timeout:int,low_speed_limit:int,low_speed_time:int}
     */
	private function build_curl_timeout_options( $timeout ) {
		$timeout = max( 10, intval( $timeout ) );
		if ( $timeout < 35 ) {
			$low_speed_time = max( 3, $timeout - 7 );
		} elseif ( $timeout >= 145 ) {
			$low_speed_time = min( 90, max( 60, $timeout - 105 ) );
		} else {
			$low_speed_time = min( 120, max( 40, $timeout - 15 ) );
		}

		return array(
			'timeout'         => $timeout,
			'connect_timeout' => min( 10, max( 3, $timeout - 2 ) ),
			'low_speed_limit' => 1,
			'low_speed_time'  => $low_speed_time,
		);
	}

    /**
     * Execute the Responses API streaming request via cURL with hard timeouts.
     *
     * cURL is used as the transport guard. The SSE body is parsed by the same
     * Responses API reader after the transfer returns so the event semantics
     * remain identical to the stream-wrapper path while blocked reads cannot
     * strand the worker indefinitely.
     *
     * @param string        $url           Endpoint URL.
     * @param array         $headers_array HTTP headers as key => value.
     * @param string        $body_json     JSON body.
     * @param int           $timeout       Total request timeout.
     * @param callable|null $event_emitter Optional stream event callback.
     * @param string        $audit_id      Audit event id.
     * @param array         $request_body  Original request body.
     * @param float         $audit_started Request start timestamp.
     * @return array Stream result.
     */
    private function stream_with_curl_transport( $url, $headers_array, $body_json, $timeout, $event_emitter, $audit_id, $request_body, $audit_started ) {
        $timeout_options = $this->build_curl_timeout_options( $timeout );
        $header_lines    = array();
        foreach ( $headers_array as $key => $value ) {
            $header_lines[] = $key . ': ' . $value;
        }

        $response_body    = '';
        $response_headers = array();
        $heartbeat_count  = 0;
        $last_heartbeat_at = 0.0;
        $timing           = array(
            'request_payload_bytes' => strlen( (string) $body_json ),
            'transport'             => 'curl',
        );
        $curl_started_at  = microtime( true );
        $ch               = curl_init( $url );

        if ( false === $ch ) {
            $result = $this->error_result( 'Failed to initialize cURL stream transport.', 'stream_error' );
            $result['transport'] = 'curl';
            $result = $this->add_model_request_timing_meta( $result, $timing );
            $result = $this->add_transport_meta( $result, $audit_started, $timeout, 0 );
            $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
            return $result;
        }

        curl_setopt( $ch, CURLOPT_POST, true );
        curl_setopt( $ch, CURLOPT_POSTFIELDS, $body_json );
        curl_setopt( $ch, CURLOPT_HTTPHEADER, $header_lines );
        curl_setopt( $ch, CURLOPT_RETURNTRANSFER, false );
        curl_setopt( $ch, CURLOPT_HEADER, false );
        curl_setopt( $ch, CURLOPT_FOLLOWLOCATION, false );
        curl_setopt( $ch, CURLOPT_SSL_VERIFYPEER, true );
        curl_setopt( $ch, CURLOPT_SSL_VERIFYHOST, 2 );
        curl_setopt( $ch, CURLOPT_TIMEOUT, $timeout_options['timeout'] );
        curl_setopt( $ch, CURLOPT_CONNECTTIMEOUT, $timeout_options['connect_timeout'] );
        curl_setopt( $ch, CURLOPT_LOW_SPEED_LIMIT, $timeout_options['low_speed_limit'] );
        curl_setopt( $ch, CURLOPT_LOW_SPEED_TIME, $timeout_options['low_speed_time'] );
        curl_setopt( $ch, CURLOPT_NOSIGNAL, true );
        curl_setopt( $ch, CURLOPT_HEADERFUNCTION, function ( $curl, $header_line ) use ( &$response_headers ) {
            $response_headers[] = trim( (string) $header_line );
            return strlen( (string) $header_line );
        } );
        $emit_transport_heartbeat = function ( $stage, $downloaded = 0, $uploaded = 0 ) use ( $event_emitter, $audit_started, $timeout_options, &$heartbeat_count, &$last_heartbeat_at ) {
            $now = microtime( true );
            if ( $last_heartbeat_at > 0 && ( $now - $last_heartbeat_at ) < self::CURL_TRANSPORT_HEARTBEAT_INTERVAL ) {
                return;
            }

            $last_heartbeat_at = $now;
            $heartbeat_count++;

            if ( function_exists( 'set_time_limit' ) ) {
                @set_time_limit( max( 60, self::CURL_TRANSPORT_HEARTBEAT_INTERVAL * 4 ) );
            }

            $this->emit_stream_event(
                $event_emitter,
                'transport_heartbeat',
                array(
                    'stage'           => (string) $stage,
                    'elapsed_ms'      => (int) round( ( $now - (float) $audit_started ) * 1000 ),
                    'timeout_seconds' => intval( $timeout_options['timeout'] ),
                    'downloaded'      => max( 0, intval( $downloaded ) ),
                    'uploaded'        => max( 0, intval( $uploaded ) ),
                    'heartbeat_count' => intval( $heartbeat_count ),
                ),
                true
            );
        };

        $emit_transport_heartbeat( 'curl_start', 0, 0 );

        if ( defined( 'CURLOPT_NOPROGRESS' ) ) {
            curl_setopt( $ch, CURLOPT_NOPROGRESS, false );
        }
        if ( defined( 'CURLOPT_XFERINFOFUNCTION' ) ) {
            curl_setopt( $ch, CURLOPT_XFERINFOFUNCTION, function ( $curl, $download_total, $downloaded, $upload_total, $uploaded ) use ( $emit_transport_heartbeat ) {
                $emit_transport_heartbeat( 'curl_progress', $downloaded, $uploaded );
                return 0;
            } );
        }

        curl_setopt( $ch, CURLOPT_WRITEFUNCTION, function ( $curl, $chunk ) use ( &$response_body, &$timing, $audit_started, $emit_transport_heartbeat ) {
            $response_body .= (string) $chunk;
            if ( '' !== (string) $chunk ) {
                $this->mark_model_request_timing_once( $timing, 'first_body_chunk_ms', $audit_started );
                if ( false !== strpos( $response_body, 'event:' ) || false !== strpos( $response_body, 'data:' ) ) {
                    $this->mark_model_request_timing_once( $timing, 'first_sse_event_ms', $audit_started );
                }
                if ( false !== strpos( $response_body, '"type":"function_call"' ) || false !== strpos( $response_body, '"type": "function_call"' ) ) {
                    $this->mark_model_request_timing_once( $timing, 'first_tool_call_event_ms', $audit_started );
                }
                if ( false !== strpos( $response_body, 'response.completed' ) ) {
                    $this->mark_model_request_timing_once( $timing, 'response_completed_ms', $audit_started );
                }
            }
            $emit_transport_heartbeat( 'body_chunk', strlen( $response_body ), 0 );
            return strlen( (string) $chunk );
        } );

        $ok          = curl_exec( $ch );
        $timing['curl_exec_ms'] = (int) round( ( microtime( true ) - $curl_started_at ) * 1000 );
        $curl_errno  = curl_errno( $ch );
        $curl_error  = curl_error( $ch );
        $http_status = intval( curl_getinfo( $ch, CURLINFO_RESPONSE_CODE ) );
        // PHP 8+ cURL handles are objects cleaned up by GC; curl_close() is
        // deprecated on PHP 8.5 and only produces noisy worker warnings.
        unset( $ch );

        if ( false === $ok ) {
            $is_timeout = defined( 'CURLE_OPERATION_TIMEDOUT' ) && CURLE_OPERATION_TIMEDOUT === $curl_errno;
            $result = $this->error_result(
                $is_timeout
                    ? 'cURL stream transport timed out after ' . intval( $timeout ) . ' seconds: ' . $curl_error
                    : 'cURL stream transport failed: ' . $curl_error,
                $is_timeout ? 'timeout' : 'stream_error'
            );
            $result['transport']      = 'curl';
            $result['curl_errno']     = intval( $curl_errno );
            $result['curl_error']     = (string) $curl_error;
            $result['http_status']    = $http_status;
            $result['raw_error_body'] = $response_body;
            $result['transport_heartbeats'] = intval( $heartbeat_count );
            $timing['response_body_bytes'] = strlen( $response_body );
            $result = $this->add_model_request_timing_meta( $result, $timing );
            $result = $this->add_transport_meta( $result, $audit_started, $timeout, $http_status );
            $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
            return $result;
        }

        if ( $http_status >= 400 ) {
            $result = $this->error_result( $this->parse_api_error( $response_body, $http_status ), 'http_error' );
            $result['transport']      = 'curl';
            $result['http_status']    = $http_status;
            $result['raw_error_body'] = $response_body;
            $result['transport_heartbeats'] = intval( $heartbeat_count );
            $timing['response_body_bytes'] = strlen( $response_body );
            $result = $this->add_model_request_timing_meta( $result, $timing );
            $result = $this->add_transport_meta( $result, $audit_started, $timeout, $http_status );
            $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
            return $result;
        }

        $memory_stream = fopen( 'php://temp', 'r+' );
        if ( ! $memory_stream ) {
            $result = $this->error_result( 'Failed to allocate memory stream for cURL SSE response.', 'stream_error' );
            $result['transport'] = 'curl';
            $result['transport_heartbeats'] = intval( $heartbeat_count );
            $timing['response_body_bytes'] = strlen( $response_body );
            $result = $this->add_model_request_timing_meta( $result, $timing );
            $result = $this->add_transport_meta( $result, $audit_started, $timeout, $http_status );
            $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
            return $result;
        }

        fwrite( $memory_stream, $response_body );
        rewind( $memory_stream );

        $parse_started_at = microtime( true );
        $result = $this->read_responses_api_stream( $memory_stream, $timeout, $event_emitter, $audit_started, $timing );
        if ( isset( $result['model_request_timing'] ) && is_array( $result['model_request_timing'] ) ) {
            $timing = array_merge( $timing, $result['model_request_timing'] );
        }
        $timing['parse_ms'] = (int) round( ( microtime( true ) - $parse_started_at ) * 1000 );
        $timing['response_body_bytes'] = strlen( $response_body );
        $result = $this->add_model_request_timing_meta( $result, $timing );
        $result['transport']       = 'curl';
        $result['curl_errno']      = 0;
        $result['response_headers'] = $response_headers;
        $result['transport_heartbeats'] = intval( $heartbeat_count );
        $result = $this->add_transport_meta( $result, $audit_started, $timeout, $http_status );
        $this->record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $audit_started, $result );
        return $result;
    }

    /* ================================================================
     * Responses API SSE reader
     * ============================================================= */

    /**
     * Read SSE events in Responses API format.
     *
     * Processes event-typed SSE lines:
     *   - `event: response.output_text.delta` + `data: {"delta":"..."}`
     *   - `event: response.output_item.done` + `data: {"item":{...}}`
     *   - `event: response.completed` + `data: {"response":{...}}`
     *   - `event: response.failed` + `data: {...}`
     *
     * @param resource      $stream        Open stream resource.
     * @param int           $timeout       Read timeout in seconds.
     * @param callable|null $event_emitter Event callback.
     * @return array Stream result.
     */
    private function read_responses_api_stream( $stream, $timeout, $event_emitter = null, $started_at = null, $timing = array() ) {
        $content        = '';
        $output_items   = array(); // Completed output items from response.output_item.done.
        $stop_reason    = '';
        $usage          = array();
        $current_event  = ''; // The event: line value for the current SSE block.
        $response_data  = array(); // Final response from response.completed.
        $stream_failed  = false;
        $error_message  = '';
        $events_seen    = 0;
        $data_lines     = 0;
        $bytes_read     = 0;
        $last_event_type = '';
        $started_at     = null === $started_at ? microtime( true ) : (float) $started_at;
        $timing         = is_array( $timing ) ? $timing : array();

        // Set a read timeout on the stream to detect stalls.
        stream_set_timeout( $stream, $timeout );

        while ( ! feof( $stream ) ) {
            $line = fgets( $stream, 65536 );
            if ( false !== $line ) {
                $bytes_read += strlen( (string) $line );
                if ( '' !== (string) $line ) {
                    $this->mark_model_request_timing_once( $timing, 'first_body_chunk_ms', $started_at );
                }
            }

            // Check for timeout.
            $stream_meta = stream_get_meta_data( $stream );
            if ( ! empty( $stream_meta['timed_out'] ) ) {
                fclose( $stream );
                $result = $this->error_result(
                    'Stream read timed out after ' . $timeout . ' seconds.',
                    'timeout'
                );
                $result = $this->add_model_request_timing_meta( $result, $timing );
                return $this->add_stream_observation_meta( $result, $events_seen, $data_lines, $bytes_read, $last_event_type );
            }

            if ( false === $line ) {
                break;
            }

            $line = trim( $line );

            // SSE blank lines are event separators — reset current event type.
            if ( '' === $line ) {
                $current_event = '';
                continue;
            }

            // Capture event: lines.
            if ( 0 === strpos( $line, 'event:' ) ) {
                $current_event = trim( substr( $line, 6 ) );
                continue;
            }

            // Only process data: lines.
            if ( 0 !== strpos( $line, 'data:' ) ) {
                continue;
            }

            $data_raw = trim( substr( $line, 5 ) );

            if ( '' === $data_raw || '[DONE]' === $data_raw ) {
                break;
            }

            $data = json_decode( $data_raw, true );
            if ( ! is_array( $data ) ) {
                continue;
            }

            // Resolve event type: prefer the event: line, fall back to data.type.
            $event_type = '' !== $current_event
                ? $current_event
                : ( isset( $data['type'] ) ? (string) $data['type'] : '' );
            $data_lines++;
            $events_seen++;
            $last_event_type = $event_type;
            $this->mark_model_request_timing_once( $timing, 'first_sse_event_ms', $started_at );

            // --- Handle event types ---

            // Text delta.
            if ( 'response.output_text.delta' === $event_type ) {
                $delta_text = isset( $data['delta'] ) && is_string( $data['delta'] ) ? $data['delta'] : '';
                if ( '' !== $delta_text ) {
                    $content .= $delta_text;

                    $this->emit_stream_event(
                        $event_emitter,
                        'message_update',
                        array(
                            'delta'   => $delta_text,
                            'content' => $content,
                        )
                    );
                }
                continue;
            }

            // Completed text (only use if we have no delta-accumulated content).
            if ( 'response.output_text.done' === $event_type ) {
                if ( '' === $content && isset( $data['text'] ) && is_string( $data['text'] ) ) {
                    $content = $data['text'];
                }
                continue;
            }

            // Completed output item (may be function_call, message, reasoning, etc.).
            if ( 'response.output_item.done' === $event_type ) {
                if ( ! empty( $data['item'] ) && is_array( $data['item'] ) ) {
                    $output_items[] = $data['item'];

                    // Emit tool call update for function_call items.
                    $item_type = isset( $data['item']['type'] ) ? (string) $data['item']['type'] : '';
                    if ( 'function_call' === $item_type ) {
                        $this->mark_model_request_timing_once( $timing, 'first_tool_call_event_ms', $started_at );
                        $tc_name = isset( $data['item']['name'] ) ? (string) $data['item']['name'] : '';
                        $tc_id   = isset( $data['item']['call_id'] ) ? (string) $data['item']['call_id'] : '';
                        $this->emit_stream_event(
                            $event_emitter,
                            'tool_call_update',
                            array(
                                'index' => count( $output_items ) - 1,
                                'id'    => $tc_id,
                                'name'  => $tc_name,
                            )
                        );
                    }
                }
                continue;
            }

            // Response completed — extract final response data.
            if ( 'response.completed' === $event_type ) {
                $this->mark_model_request_timing_once( $timing, 'response_completed_ms', $started_at );
                if ( ! empty( $data['response'] ) && is_array( $data['response'] ) ) {
                    $response_data = $data['response'];
                }
                break;
            }

            // Response failed.
            if ( 'response.failed' === $event_type ) {
                $stream_failed = true;
                $error_message = 'Responses API stream failed';
                if ( ! empty( $data['response']['status_details']['error']['message'] ) ) {
                    $error_message .= ': ' . (string) $data['response']['status_details']['error']['message'];
                } elseif ( ! empty( $data['error']['message'] ) ) {
                    $error_message .= ': ' . (string) $data['error']['message'];
                } elseif ( ! empty( $data['response']['status_details']['error'] ) && is_array( $data['response']['status_details']['error'] ) ) {
                    $details      = $data['response']['status_details']['error'];
                    $detail_parts = array();
                    foreach ( array( 'code', 'type', 'param' ) as $detail_key ) {
                        if ( isset( $details[ $detail_key ] ) && '' !== trim( (string) $details[ $detail_key ] ) ) {
                            $detail_parts[] = $detail_key . '=' . (string) $details[ $detail_key ];
                        }
                    }
                    if ( ! empty( $detail_parts ) ) {
                        $error_message .= ': ' . implode( ', ', $detail_parts );
                    }
                }
                break;
            }

            // Response incomplete.
            if ( 'response.incomplete' === $event_type ) {
                if ( ! empty( $data['response'] ) && is_array( $data['response'] ) ) {
                    $response_data = $data['response'];
                }
                $stop_reason = 'length';
                break;
            }
        }

        fclose( $stream );

        if ( $stream_failed ) {
            $result = $this->error_result( $error_message, 'stream_error' );
            $result = $this->add_model_request_timing_meta( $result, $timing );
            return $this->add_stream_observation_meta( $result, $events_seen, $data_lines, $bytes_read, $last_event_type );
        }

        // --- Extract data from the final response or accumulated items ---

        // If we got a completed response, prefer its output and usage.
        if ( ! empty( $response_data ) ) {
            // Extract output items from the final response if we didn't collect enough from events.
            if ( ! empty( $response_data['output'] ) && is_array( $response_data['output'] ) ) {
                // The completed response has the authoritative list of output items.
                $output_items = $response_data['output'];
            }

            // Extract usage.
            if ( ! empty( $response_data['usage'] ) && is_array( $response_data['usage'] ) ) {
                $usage = $this->extract_responses_usage( $response_data['usage'] );
            }

            // Extract stop reason from status.
            if ( empty( $stop_reason ) && ! empty( $response_data['status'] ) ) {
                $status = (string) $response_data['status'];
                if ( 'completed' === $status ) {
                    // Will be refined below based on tool calls.
                    $stop_reason = 'stop';
                } elseif ( 'incomplete' === $status ) {
                    $stop_reason = 'length';
                }
            }
        }

        // Extract text content from output items if not already accumulated from deltas.
        if ( '' === $content ) {
            $content = $this->extract_text_from_output_items( $output_items );
        }

        // Extract tool calls from output items.
        $tool_calls = $this->extract_tool_calls_from_output_items( $output_items );

        // Extract response_items for replay (function_call items, message items).
        $response_items = $this->extract_response_items( $output_items );

        // Normalize stop reason.
        if ( '' === $stop_reason ) {
            $stop_reason = ! empty( $tool_calls ) ? 'tool_calls' : 'stop';
        } elseif ( 'stop' === $stop_reason && ! empty( $tool_calls ) ) {
            $stop_reason = 'tool_calls';
        }

        // Emit done event.
        $this->emit_stream_event(
            $event_emitter,
            'done',
            array(
                'stop_reason' => $stop_reason,
            )
        );

        $result = $this->build_success_result( $content, $tool_calls, $stop_reason, $usage, $response_items );
        if ( ! empty( $response_data['id'] ) ) {
            $result['response_id'] = (string) $response_data['id'];
        }
        $result = $this->add_model_request_timing_meta( $result, $timing );
        return $this->add_stream_observation_meta( $result, $events_seen, $data_lines, $bytes_read, $last_event_type );
    }

    private function emit_stream_event( $event_emitter, $event, $payload = array(), $swallow_exceptions = false ) {
        if ( ! is_callable( $event_emitter ) ) {
            return;
        }

        if ( ! $swallow_exceptions ) {
            call_user_func( $event_emitter, (string) $event, is_array( $payload ) ? $payload : array() );
            return;
        }

        try {
            call_user_func( $event_emitter, (string) $event, is_array( $payload ) ? $payload : array() );
        } catch ( Throwable $e ) {
            // Transport watchdog must never make cURL fail.
        }
    }

    /* ================================================================
     * Responses API — output extraction helpers
     * ============================================================= */

    /**
     * Extract text content from Responses API output items.
     *
     * Looks for message items with output_text content parts.
     *
     * @param array $output_items Output items from the response.
     * @return string Concatenated text content.
     */
    private function extract_text_from_output_items( $output_items ) {
        if ( ! is_array( $output_items ) || empty( $output_items ) ) {
            return '';
        }

        $parts = array();

        foreach ( $output_items as $item ) {
            if ( ! is_array( $item ) ) {
                continue;
            }

            $type = isset( $item['type'] ) ? (string) $item['type'] : '';

            // Direct output_text item.
            if ( 'output_text' === $type && isset( $item['text'] ) && is_string( $item['text'] ) ) {
                $parts[] = $item['text'];
                continue;
            }

            // Message item with content array.
            if ( 'message' === $type && ! empty( $item['content'] ) && is_array( $item['content'] ) ) {
                foreach ( $item['content'] as $c ) {
                    if ( ! is_array( $c ) ) {
                        continue;
                    }
                    $c_type = isset( $c['type'] ) ? (string) $c['type'] : '';
                    if ( in_array( $c_type, array( 'output_text', 'text', 'input_text' ), true ) ) {
                        if ( isset( $c['text'] ) && is_string( $c['text'] ) && '' !== $c['text'] ) {
                            $parts[] = $c['text'];
                        }
                    }
                }
                continue;
            }

            // Message item with string content.
            if ( 'message' === $type && ! empty( $item['content'] ) && is_string( $item['content'] ) ) {
                $parts[] = $item['content'];
            }
        }

        return implode( "\n", $parts );
    }

    /**
     * Extract tool calls from Responses API output items.
     *
     * Looks for function_call items and normalizes them into the standard
     * [{id, name, arguments}, ...] format.
     *
     * @param array $output_items Output items from the response.
     * @return array Tool calls [{id, name, arguments}, ...].
     */
    private function extract_tool_calls_from_output_items( $output_items ) {
        if ( ! is_array( $output_items ) || empty( $output_items ) ) {
            return array();
        }

        $tool_calls = array();

        foreach ( $output_items as $item ) {
            if ( ! is_array( $item ) ) {
                continue;
            }

            $type = isset( $item['type'] ) ? (string) $item['type'] : '';
            if ( 'function_call' !== $type ) {
                continue;
            }

            $id = isset( $item['call_id'] ) ? trim( (string) $item['call_id'] ) : '';
            if ( '' === $id && ! empty( $item['id'] ) ) {
                $id = trim( (string) $item['id'] );
            }
            if ( '' === $id ) {
                $id = 'call_' . $this->generate_uuid();
            }

            $name = isset( $item['name'] ) ? trim( (string) $item['name'] ) : '';
            if ( '' === $name ) {
                continue;
            }

            $arguments = '';
            if ( isset( $item['arguments'] ) ) {
                if ( is_string( $item['arguments'] ) ) {
                    $arguments = $item['arguments'];
                } elseif ( is_array( $item['arguments'] ) ) {
                    $arguments = wp_json_encode( $item['arguments'] );
                }
            }
            if ( '' === $arguments ) {
                $arguments = '{}';
            }

            $tool_calls[] = array(
                'id'        => $id,
                'name'      => $name,
                'arguments' => $arguments,
            );
        }

        return $tool_calls;
    }

    /**
     * Extract response_items for replay from Responses API output items.
     *
     * These items are stored on the assistant message so that future requests
     * can replay them using the Responses API input format.
     *
     * @param array $output_items Output items from the response.
     * @return array Normalized response items.
     */
    private function extract_response_items( $output_items ) {
        if ( ! is_array( $output_items ) || empty( $output_items ) ) {
            return array();
        }

        $items = array();

        foreach ( $output_items as $item ) {
            if ( ! is_array( $item ) || empty( $item['type'] ) ) {
                continue;
            }

            $type = (string) $item['type'];

            if ( 'message' === $type ) {
                $role = isset( $item['role'] ) ? strtolower( trim( (string) $item['role'] ) ) : 'assistant';
                if ( ! in_array( $role, array( 'assistant', 'user' ), true ) ) {
                    continue;
                }
                $entry = array(
                    'type' => 'message',
                    'role' => $role,
                );
                if ( isset( $item['content'] ) ) {
                    $entry['content'] = $item['content'];
                }
                $items[] = $entry;
                continue;
            }

            if ( 'function_call' === $type ) {
                $call_id = isset( $item['call_id'] ) ? trim( (string) $item['call_id'] ) : '';
                $name    = isset( $item['name'] ) ? trim( (string) $item['name'] ) : '';
                if ( '' === $call_id || '' === $name ) {
                    continue;
                }
                $entry = array(
                    'type'      => 'function_call',
                    'call_id'   => $call_id,
                    'name'      => $name,
                    'arguments' => isset( $item['arguments'] ) && is_string( $item['arguments'] ) ? $item['arguments'] : '{}',
                );
                if ( ! empty( $item['id'] ) ) {
                    $entry['id'] = (string) $item['id'];
                }
                $items[] = $entry;
                continue;
            }

            if ( 'reasoning' === $type ) {
                if ( empty( $item['id'] ) ) {
                    continue;
                }
                // Preserve the FULL reasoning item including encrypted_content.
                // With store:false + include:["reasoning.encrypted_content"],
                // the API returns encrypted reasoning data that can be sent back
                // for multi-turn continuity — exactly how Pi/OpenClaw works.
                $entry = array(
                    'type' => 'reasoning',
                    'id'   => (string) $item['id'],
                );
                // encrypted_content is the key to replay without store:true.
                if ( isset( $item['encrypted_content'] ) && is_string( $item['encrypted_content'] ) ) {
                    $entry['encrypted_content'] = $item['encrypted_content'];
                }
                // Summary — required field for reasoning items in input.
                if ( isset( $item['summary'] ) && is_array( $item['summary'] ) ) {
                    $entry['summary'] = $item['summary'];
                } elseif ( isset( $item['summary'] ) && is_string( $item['summary'] ) && '' !== $item['summary'] ) {
                    $entry['summary'] = array( array( 'type' => 'summary_text', 'text' => $item['summary'] ) );
                } else {
                    $entry['summary'] = array();
                }
                $items[] = $entry;
                continue;
            }

            if ( 'web_search_call' === $type ) {
                $items[] = $item;
            }
        }

        return $items;
    }

    /**
     * Extract usage from Responses API usage object.
     *
     * The Responses API uses input_tokens/output_tokens instead of
     * prompt_tokens/completion_tokens.
     *
     * @param array $usage Raw usage from response.
     * @return array Normalized usage {prompt_tokens, completion_tokens, total_tokens, cached_tokens, cache_hit_rate}.
     */
    private function extract_responses_usage( $usage ) {
        if ( ! is_array( $usage ) ) {
            return array();
        }

        $prompt     = isset( $usage['input_tokens'] ) ? intval( $usage['input_tokens'] ) : 0;
        $completion = isset( $usage['output_tokens'] ) ? intval( $usage['output_tokens'] ) : 0;
        $cached     = 0;

        // Fall back to standard keys if present.
        if ( 0 === $prompt && isset( $usage['prompt_tokens'] ) ) {
            $prompt = intval( $usage['prompt_tokens'] );
        }
        if ( 0 === $completion && isset( $usage['completion_tokens'] ) ) {
            $completion = intval( $usage['completion_tokens'] );
        }

        $total = isset( $usage['total_tokens'] ) ? intval( $usage['total_tokens'] ) : ( $prompt + $completion );

        if ( isset( $usage['input_tokens_details'] ) && is_array( $usage['input_tokens_details'] ) && isset( $usage['input_tokens_details']['cached_tokens'] ) ) {
            $cached = intval( $usage['input_tokens_details']['cached_tokens'] );
        } elseif ( isset( $usage['prompt_tokens_details'] ) && is_array( $usage['prompt_tokens_details'] ) && isset( $usage['prompt_tokens_details']['cached_tokens'] ) ) {
            $cached = intval( $usage['prompt_tokens_details']['cached_tokens'] );
        } elseif ( isset( $usage['cached_tokens'] ) ) {
            $cached = intval( $usage['cached_tokens'] );
        }

        $cached = max( 0, min( $cached, $prompt ) );

        return array(
            'prompt_tokens'     => $prompt,
            'completion_tokens' => $completion,
            'total_tokens'      => $total,
            'cached_tokens'     => $cached,
            'cache_hit_rate'    => $prompt > 0 ? round( $cached / $prompt, 4 ) : 0.0,
        );
    }

    /* ================================================================
     * Request building
     * ============================================================= */

    /**
     * Build the JSON request body for the streaming API call.
     *
     * @return array Associative array ready for JSON encoding.
     */
    private function build_request_body() {
        return $this->build_responses_request_body();
    }

    /**
     * Build the Responses API format request body.
     *
     * Converts Chat Completions messages to Responses API input format:
     *   - system messages -> instructions
     *   - user/assistant messages -> input items (type: message)
     *   - tool messages -> function_call_output items
     *   - assistant tool_calls -> function_call items
     *
     * Mirrors the logic in Niwa_AI_OpenAI_Client::convert_messages_to_responses_payload().
     *
     * @return array Associative array for Responses API.
     */
    private function build_responses_request_body() {
        $converted = $this->convert_messages_to_responses_format( $this->messages );

        $body = array(
            'model'  => $this->normalize_request_model( $this->model ),
            'input'  => isset( $converted['input'] ) ? $converted['input'] : array(),
            'stream' => true,
            'store'  => false,
            // Request encrypted reasoning content so it can be replayed in
            // future turns without server-side persistence (store:false).
            // This is how Pi/OpenClaw maintain multi-turn reasoning continuity.
            'include' => array( 'reasoning.encrypted_content' ),
        );

        // Set instructions.
        $instructions = isset( $converted['instructions'] ) ? trim( (string) $converted['instructions'] ) : '';
        $mode = $this->get_normalized_auth_mode();
        if ( 'oauth' === $mode || 'codex' === $mode ) {
            $body['instructions'] = '' !== $instructions ? $instructions : self::DEFAULT_CODEX_INSTRUCTIONS;
        } elseif ( '' !== $instructions ) {
            $body['instructions'] = $instructions;
        }

        // Convert and add tools.
        if ( null !== $this->tools ) {
            $response_tools = $this->convert_tools_to_responses_format( $this->tools );
            if ( ! empty( $response_tools ) ) {
                $body['tools'] = $response_tools;
            }

            $tool_choice = isset( $this->options['tool_choice'] ) ? $this->options['tool_choice'] : 'auto';
            if ( 'none' !== $tool_choice ) {
                $body['tool_choice'] = $tool_choice;
            }
            $body['parallel_tool_calls'] = true;
        }

        // Request auto-generated reasoning summaries — like Pi SDK (summary: "auto").
        // This lets reasoning items include a summary that can be sent back
        // without server-side storage.
        $body['reasoning'] = $this->reasoning_options_for_model(
            $this->model,
            array(
                'summary' => 'auto',
            )
        );

        return $body;
    }

    /**
     * Convert Chat Completions messages array to Responses API input format.
     *
     * Mirrors Niwa_AI_OpenAI_Client::convert_messages_to_responses_payload().
     *
     * @param array $messages Messages array [{role, content, tool_calls?, tool_call_id?}, ...].
     * @return array {instructions: string, input: array}.
     */
    private function convert_messages_to_responses_format( $messages ) {
        $instructions = '';
        $input = array();

        if ( ! is_array( $messages ) ) {
            return array( 'instructions' => '', 'input' => array() );
        }

        foreach ( $messages as $message ) {
            if ( ! is_array( $message ) ) {
                continue;
            }

            $role    = isset( $message['role'] ) ? strtolower( trim( (string) $message['role'] ) ) : 'user';
            $content = $this->stringify_content( isset( $message['content'] ) ? $message['content'] : '' );

            // System messages -> instructions.
            if ( 'system' === $role ) {
                if ( '' !== $content ) {
                    $instructions = '' === $instructions ? $content : ( $instructions . "\n\n" . $content );
                }
                continue;
            }

            // Assistant messages -> replay response_items when available,
            // otherwise rebuild from tool_calls.
            //
            // With store:false + include:["reasoning.encrypted_content"] +
            // reasoning.summary:"auto", the API returns reasoning items with
            // encrypted_content and summary.  These CAN be replayed without
            // server-side persistence — this is exactly how Pi/OpenClaw works.
            if ( 'assistant' === $role ) {
                $function_call_items     = array();
                $has_message_item        = false;
                $has_replayable_response = false;

                if ( ! empty( $message['response_items'] ) && is_array( $message['response_items'] ) ) {
                    foreach ( $message['response_items'] as $ri ) {
                        if ( ! is_array( $ri ) || empty( $ri['type'] ) ) {
                            continue;
                        }
                        $ri_type = (string) $ri['type'];
                        if ( 'function_call' === $ri_type ) {
                            $function_call_items[] = $ri;
                            continue;
                        }
                        if ( 'message' === $ri_type ) {
                            $has_message_item = true;
                        }
                        // Ensure reasoning items have summary (required by API).
                        if ( 'reasoning' === $ri_type && ! isset( $ri['summary'] ) ) {
                            $ri['summary'] = array();
                        }
                        $input[] = $ri;
                        $has_replayable_response = true;
                    }
                }

                if ( ! empty( $message['tool_calls'] ) && is_array( $message['tool_calls'] ) ) {
                    if ( ! $has_message_item && '' !== $content ) {
                        $msg_item = $this->build_responses_message_item( 'assistant', $message['content'] );
                        if ( ! empty( $msg_item ) ) {
                            $input[] = $msg_item;
                        }
                    }

                    // Use function_call items from response_items if available
                    // (they have the correct server-assigned call_id), otherwise
                    // rebuild from portable tool_calls.
                    if ( empty( $function_call_items ) ) {
                        foreach ( $message['tool_calls'] as $call ) {
                            $resolved = $this->normalize_tool_call( $call );
                            if ( empty( $resolved ) ) {
                                continue;
                            }
                            $function_call_items[] = array(
                                'type'      => 'function_call',
                                'call_id'   => $resolved['id'],
                                'name'      => $resolved['name'],
                                'arguments' => $resolved['arguments'],
                            );
                        }
                    }

                    foreach ( $function_call_items as $fc_item ) {
                        $input[] = $fc_item;
                    }
                    continue;
                }

                // Replayable response without tool_calls (e.g. reasoning + text).
                if ( $has_replayable_response ) {
                    if ( ! $has_message_item && '' !== $content ) {
                        $msg_item = $this->build_responses_message_item( 'assistant', $message['content'] );
                        if ( ! empty( $msg_item ) ) {
                            $input[] = $msg_item;
                        }
                    }
                    continue;
                }

                // Plain assistant message — no response_items, no tool_calls.
                if ( '' !== $content ) {
                    $msg_item = $this->build_responses_message_item( 'assistant', $message['content'] );
                    if ( ! empty( $msg_item ) ) {
                        $input[] = $msg_item;
                    }
                    continue;
                }
            }

            // Tool result messages -> function_call_output items.
            if ( 'tool' === $role || ( 'user' === $role && ! empty( $message['tool_call_id'] ) ) ) {
                $call_id = isset( $message['tool_call_id'] ) ? trim( (string) $message['tool_call_id'] ) : '';
                if ( '' === $call_id ) {
                    continue;
                }
                $input[] = array(
                    'type'    => 'function_call_output',
                    'call_id' => $call_id,
                    'output'  => $content,
                );
                continue;
            }

            // Regular user/assistant messages -> message items.
            if ( in_array( $role, array( 'user', 'assistant' ), true ) ) {
                $msg_item = $this->build_responses_message_item( $role, isset( $message['content'] ) ? $message['content'] : $content );
                if ( ! empty( $msg_item ) ) {
                    $input[] = $msg_item;
                }
            }
        }

        // Prune orphan function_call items (those without matching function_call_output).
        $input = $this->prune_orphan_function_calls( $input );

        return array(
            'instructions' => $instructions,
            'input'        => $input,
        );
    }

    /**
     * Build a Responses API message item.
     *
     * @param string $role    'user' or 'assistant'.
     * @param mixed  $content Content (string, array of parts, etc.).
     * @return array Message item or empty array.
     */
    private function build_responses_message_item( $role, $content ) {
        $parts = $this->normalize_content_for_responses( $content, $role );
        if ( empty( $parts ) ) {
            return array();
        }

        return array(
            'type'    => 'message',
            'role'    => $role,
            'content' => $parts,
        );
    }

    /**
     * Normalize content into Responses API content parts format.
     *
     * The Responses API requires different content types per role:
     *   - user messages      → input_text
     *   - assistant messages  → output_text
     *
     * @param mixed  $content String or array content.
     * @param string $role    Message role ('user' or 'assistant'). Defaults to 'user'.
     * @return array Array of content parts.
     */
    private function normalize_content_for_responses( $content, $role = 'user' ) {
        $text_type = 'assistant' === $role ? 'output_text' : 'input_text';

        if ( is_string( $content ) || is_scalar( $content ) ) {
            $text = (string) $content;
            if ( '' === $text ) {
                return array();
            }
            return array( array( 'type' => $text_type, 'text' => $text ) );
        }

        if ( ! is_array( $content ) ) {
            return array();
        }

        $parts = array();

        foreach ( $content as $item ) {
            if ( is_string( $item ) ) {
                if ( '' !== $item ) {
                    $parts[] = array( 'type' => $text_type, 'text' => $item );
                }
                continue;
            }

            if ( ! is_array( $item ) ) {
                continue;
            }

            $type = isset( $item['type'] ) ? strtolower( trim( (string) $item['type'] ) ) : '';

            if ( in_array( $type, array( 'text', 'input_text', 'output_text' ), true ) ) {
                $text = isset( $item['text'] ) ? (string) $item['text'] : '';
                if ( '' !== $text ) {
                    $parts[] = array( 'type' => $text_type, 'text' => $text );
                }
                continue;
            }

            // Image parts — only valid in user messages.
            if ( ! empty( $item['image_url'] ) ) {
                $url = '';
                if ( is_string( $item['image_url'] ) ) {
                    $url = $item['image_url'];
                } elseif ( is_array( $item['image_url'] ) && ! empty( $item['image_url']['url'] ) ) {
                    $url = (string) $item['image_url']['url'];
                }
                if ( '' !== $url ) {
                    $img = array( 'type' => 'input_image', 'image_url' => $url );
                    if ( ! empty( $item['detail'] ) ) {
                        $img['detail'] = (string) $item['detail'];
                    } elseif ( is_array( $item['image_url'] ) && ! empty( $item['image_url']['detail'] ) ) {
                        $img['detail'] = (string) $item['image_url']['detail'];
                    }
                    $parts[] = $img;
                }
                continue;
            }

            // Fallback: extract text if present.
            if ( ! empty( $item['text'] ) && is_string( $item['text'] ) ) {
                $parts[] = array( 'type' => $text_type, 'text' => $item['text'] );
            }
        }

        return $parts;
    }

    /**
     * Convert Chat Completions tools array to Responses API format.
     *
     * Chat Completions: [{type: "function", function: {name, description, parameters}}]
     * Responses API:    [{type: "function", name, description, parameters, strict: false}]
     *
     * Mirrors Niwa_AI_OpenAI_Client::convert_tools_to_responses().
     *
     * @param array $tools Tools in Chat Completions format.
     * @return array Tools in Responses API format.
     */
    private function convert_tools_to_responses_format( $tools ) {
        if ( ! is_array( $tools ) ) {
            return array();
        }

        $converted = array();

        foreach ( $tools as $tool ) {
            if ( ! is_array( $tool ) || empty( $tool['type'] ) ) {
                continue;
            }

            $type = (string) $tool['type'];

            // Pass through web_search tools as-is.
            if ( in_array( $type, array( 'web_search', 'web_search_preview' ), true ) ) {
                $entry = array( 'type' => 'web_search' );
                if ( ! empty( $tool['search_context_size'] ) ) {
                    $entry['search_context_size'] = (string) $tool['search_context_size'];
                }
                $converted[] = $entry;
                continue;
            }

            if ( 'function' !== $type ) {
                continue;
            }

            if ( empty( $tool['function'] ) || ! is_array( $tool['function'] ) ) {
                continue;
            }

            $name = isset( $tool['function']['name'] ) ? trim( (string) $tool['function']['name'] ) : '';
            if ( '' === $name ) {
                continue;
            }

            $params = $this->normalize_json_schema_for_api(
                isset( $tool['function']['parameters'] ) ? $tool['function']['parameters'] : (object) array()
            );

            $entry = array(
                'type'       => 'function',
                'name'       => $name,
                'parameters' => $params,
                'strict'     => false,
            );

            if ( ! empty( $tool['function']['description'] ) ) {
                $entry['description'] = (string) $tool['function']['description'];
            }

            $converted[] = $entry;
        }

        return $converted;
    }

    private function normalize_json_schema_for_api( $schema ) {
        if ( ! is_array( $schema ) ) {
            return $schema;
        }

        $object_map_keys = array( 'properties', 'patternProperties', 'definitions', '$defs', 'dependentSchemas' );

        foreach ( $schema as $key => $value ) {
            if ( is_array( $value ) ) {
                $schema[ $key ] = $this->normalize_json_schema_for_api( $value );
                if ( in_array( $key, $object_map_keys, true ) && empty( $schema[ $key ] ) ) {
                    $schema[ $key ] = new \stdClass();
                }
            }
        }

        return $schema;
    }

    /**
     * Normalize a tool call from Chat Completions format.
     *
     * @param array $call Tool call from assistant message.
     * @return array {id, name, arguments} or empty array.
     */
    private function normalize_tool_call( $call ) {
        if ( ! is_array( $call ) ) {
            return array();
        }

        $id        = isset( $call['id'] ) ? trim( (string) $call['id'] ) : '';
        $name      = '';
        $arguments = '';

        if ( ! empty( $call['function'] ) && is_array( $call['function'] ) ) {
            $name      = isset( $call['function']['name'] ) ? trim( (string) $call['function']['name'] ) : '';
            $arguments = isset( $call['function']['arguments'] ) ? (string) $call['function']['arguments'] : '';
        }

        if ( '' === $name && ! empty( $call['name'] ) ) {
            $name = trim( (string) $call['name'] );
        }
        if ( '' === $arguments && isset( $call['arguments'] ) ) {
            if ( is_string( $call['arguments'] ) ) {
                $arguments = $call['arguments'];
            } elseif ( is_array( $call['arguments'] ) ) {
                $arguments = wp_json_encode( $call['arguments'] );
            }
        }

        if ( '' === $id ) {
            $id = 'call_' . $this->generate_uuid();
        }
        if ( '' === $name ) {
            return array();
        }

        return array(
            'id'        => $id,
            'name'      => $name,
            'arguments' => '' !== $arguments ? $arguments : '{}',
        );
    }

    /**
     * Prune orphan function_call items from Responses API input.
     *
     * A function_call without a matching function_call_output can cause
     * API errors. This removes function_calls that lack a corresponding output,
     * and also removes function_calls entirely if there are no outputs at all.
     *
     * Mirrors Niwa_AI_OpenAI_Client::prune_orphan_function_calls_from_input().
     *
     * @param array $input Responses API input items.
     * @return array Pruned input items.
     */
    private function prune_orphan_function_calls( $input ) {
        if ( ! is_array( $input ) || empty( $input ) ) {
            return array();
        }

        // Collect all function_call_output call_ids.
        $output_ids = array();
        foreach ( $input as $item ) {
            if ( ! is_array( $item ) ) {
                continue;
            }
            if ( 'function_call_output' !== ( isset( $item['type'] ) ? (string) $item['type'] : '' ) ) {
                continue;
            }
            $call_id = isset( $item['call_id'] ) ? trim( (string) $item['call_id'] ) : '';
            if ( '' !== $call_id ) {
                $output_ids[ $call_id ] = true;
            }
        }

        // If no outputs exist, remove all function_calls.
        if ( empty( $output_ids ) ) {
            return array_values(
                array_filter(
                    $input,
                    function ( $item ) {
                        return ! is_array( $item ) || 'function_call' !== ( isset( $item['type'] ) ? (string) $item['type'] : '' );
                    }
                )
            );
        }

        // Remove function_calls without matching output.
        $pruned = array();
        foreach ( $input as $item ) {
            if ( ! is_array( $item ) ) {
                continue;
            }
            if ( 'function_call' === ( isset( $item['type'] ) ? (string) $item['type'] : '' ) ) {
                $call_id = isset( $item['call_id'] ) ? trim( (string) $item['call_id'] ) : '';
                if ( '' === $call_id || empty( $output_ids[ $call_id ] ) ) {
                    continue;
                }
            }
            $pruned[] = $item;
        }

        return $pruned;
    }

    /* ================================================================
     * HTTP / headers / URL resolution
     * ============================================================= */

    /**
     * Build HTTP headers for the streaming request.
     *
     * @return array Associative array of header name => value.
     */
    private function build_http_headers() {
        $token   = $this->resolve_auth_token();
        $headers = array(
            'Content-Type'  => 'application/json',
            'Authorization' => 'Bearer ' . $token,
            'Accept'        => 'text/event-stream',
        );

        $mode = $this->get_normalized_auth_mode();

        if ( 'oauth' === $mode || 'codex' === $mode ) {
            $headers['originator']  = self::CODEX_ORIGINATOR;
            $headers['OpenAI-Beta'] = 'responses=experimental';
            $headers['User-Agent']  = 'CodexBar';

            $account_id = isset( $this->creds['account_id'] ) ? trim( (string) $this->creds['account_id'] ) : '';
            if ( '' !== $account_id ) {
                $headers['ChatGPT-Account-Id'] = $account_id;
            }
        }

        return $headers;
    }

    /**
     * Resolve the auth token from credentials based on auth mode.
     *
     * @return string Bearer token string.
     */
    private function resolve_auth_token() {
        $mode = $this->get_normalized_auth_mode();

        if ( 'oauth' === $mode ) {
            return isset( $this->creds['access_token'] ) ? trim( (string) $this->creds['access_token'] ) : '';
        }

        if ( 'codex' === $mode ) {
            return isset( $this->creds['codex_token'] ) ? trim( (string) $this->creds['codex_token'] ) : '';
        }

        // Default: api_key mode.
        return isset( $this->creds['api_key'] ) ? trim( (string) $this->creds['api_key'] ) : '';
    }

    /**
     * Get the normalized auth mode string.
     *
     * @return string 'api_key', 'oauth', or 'codex'.
     */
    private function get_normalized_auth_mode() {
        return strtolower( trim( (string) $this->auth_mode ) );
    }

    private function normalize_request_model( $model ) {
        if ( class_exists( 'Niwa_AI_Model_Catalog' ) && method_exists( 'Niwa_AI_Model_Catalog', 'normalize_request_model' ) ) {
            return (string) Niwa_AI_Model_Catalog::normalize_request_model( $model );
        }

        return (string) $model;
    }

    private function reasoning_options_for_model( $model, $reasoning ) {
        if ( class_exists( 'Niwa_AI_Model_Catalog' ) && method_exists( 'Niwa_AI_Model_Catalog', 'with_reasoning_effort' ) ) {
            return Niwa_AI_Model_Catalog::with_reasoning_effort( $model, $reasoning );
        }

        return is_array( $reasoning ) ? $reasoning : array();
    }

    /**
     * Resolve the API endpoint URL.
     *
     * Routes to the correct Responses endpoint based on auth mode:
     *   - OAuth -> Codex Responses endpoint (chatgpt.com)
     *   - Codex -> Codex Responses endpoint (chatgpt.com)
     *   - API key -> OpenAI Responses endpoint
     *
     * Custom api_base_url in options overrides all auto-detection.
     *
     * @return string Full API URL.
     */
    private function resolve_api_url() {
        // Custom URL takes precedence.
        if ( isset( $this->options['api_base_url'] ) && '' !== trim( (string) $this->options['api_base_url'] ) ) {
            $base = rtrim( trim( (string) $this->options['api_base_url'] ), '/' );

            // If the custom URL already ends with /responses, use as-is.
            if ( preg_match( '/\/responses\/?$/', $base ) ) {
                return $base;
            }

            return $base . '/responses';
        }

        $mode = $this->get_normalized_auth_mode();

        // OAuth and codex modes -> Codex endpoint.
        if ( 'oauth' === $mode || 'codex' === $mode ) {
            return self::CODEX_RESPONSES_URL;
        }

        return self::RESPONSES_API_URL;
    }

    /* ================================================================
     * Tool call assembly & result building
     * ============================================================= */

    /**
     * Build a standardized success result array.
     *
     * @param string $content        Accumulated text content.
     * @param array  $tool_calls     Tool calls [{id, name, arguments}, ...].
     * @param string $stop_reason    Stop reason string.
     * @param array  $usage          Token usage.
     * @param array  $response_items Optional Responses API replay items.
     * @return array Success result.
     */
    private function build_success_result( $content, $tool_calls, $stop_reason, $usage, $response_items = array() ) {
        // Build the assistant message suitable for the messages array.
        $assistant_message = array(
            'role'    => 'assistant',
            'content' => $content,
        );

        if ( ! empty( $tool_calls ) ) {
            $api_tool_calls = array();
            foreach ( $tool_calls as $tc ) {
                $api_tool_calls[] = array(
                    'id'       => $tc['id'],
                    'type'     => 'function',
                    'function' => array(
                        'name'      => $tc['name'],
                        'arguments' => $tc['arguments'],
                    ),
                );
            }
            $assistant_message['tool_calls'] = $api_tool_calls;
        }

        // Include response_items for Responses API replay.
        if ( ! empty( $response_items ) ) {
            $assistant_message['response_items'] = $response_items;
        }

        return array(
            'success'           => true,
            'role'              => 'assistant',
            'content'           => $content,
            'tool_calls'        => $tool_calls,
            'stop_reason'       => $stop_reason,
            'usage'             => $usage,
            'assistant_message' => $assistant_message,
            'error'             => '',
            'error_type'        => '',
        );
    }

    private function record_stream_audit_event( $audit_id, $url, $timeout, $request_body, $started_at, $result ) {
        if ( ! class_exists( 'Niwa_AI_LLM_Audit_Logger' ) || ! Niwa_AI_LLM_Audit_Logger::is_enabled() ) {
            return;
        }

        Niwa_AI_LLM_Audit_Logger::record_event(
            'llm_stream_request',
            array(
                'audit_id' => $audit_id,
                'endpoint' => $this->is_codex_url( $url ) ? 'codex_responses_stream' : 'openai_responses_stream',
                'endpoint_url' => (string) $url,
                'timeout_seconds' => intval( $timeout ),
                'duration_ms' => round( ( microtime( true ) - floatval( $started_at ) ) * 1000, 2 ),
                'request' => array(
                    'model' => isset( $request_body['model'] ) ? (string) $request_body['model'] : $this->model,
                    'has_tools' => ! empty( $request_body['tools'] ),
                    'tool_choice' => isset( $request_body['tool_choice'] ) ? $request_body['tool_choice'] : '',
                    'payload' => $request_body,
                ),
                'response' => array(
                    'success' => ! empty( $result['success'] ),
                    'status' => isset( $result['http_status'] ) ? intval( $result['http_status'] ) : 0,
                    'error_type' => isset( $result['error_type'] ) ? (string) $result['error_type'] : '',
                    'error' => isset( $result['error'] ) ? (string) $result['error'] : '',
                    'output_text' => isset( $result['content'] ) ? (string) $result['content'] : '',
                    'tool_calls' => isset( $result['tool_calls'] ) ? $result['tool_calls'] : array(),
                    'stop_reason' => isset( $result['stop_reason'] ) ? (string) $result['stop_reason'] : '',
                    'usage' => isset( $result['usage'] ) ? $result['usage'] : array(),
                    'assistant_message' => isset( $result['assistant_message'] ) ? $result['assistant_message'] : array(),
                    'raw_error_body' => isset( $result['raw_error_body'] ) ? (string) $result['raw_error_body'] : '',
                ),
            )
        );
    }

    private function is_codex_url( $url ) {
        return false !== strpos( strtolower( (string) $url ), '/backend-api/codex/responses' );
    }

    /* ================================================================
     * General helpers
     * ============================================================= */

    /**
     * Extract HTTP status code from stream metadata.
     *
     * @param array $meta Stream metadata from stream_get_meta_data().
     * @return int HTTP status code, or 0 if not found.
     */
    private function extract_http_status( $meta ) {
        if ( ! isset( $meta['wrapper_data'] ) || ! is_array( $meta['wrapper_data'] ) ) {
            return 0;
        }

        foreach ( $meta['wrapper_data'] as $header_line ) {
            if ( preg_match( '/^HTTP\/\d+\.?\d*\s+(\d{3})/', (string) $header_line, $matches ) ) {
                return intval( $matches[1] );
            }
        }

        return 0;
    }

    /**
     * Parse an API error response body into a human-readable message.
     *
     * @param string $body        Raw response body.
     * @param int    $http_status HTTP status code.
     * @return string Error description.
     */
    private function parse_api_error( $body, $http_status ) {
        $prefix = 'HTTP ' . $http_status . ' error';

        if ( empty( $body ) ) {
            return $prefix . '.';
        }

        $decoded = json_decode( (string) $body, true );

        if ( is_array( $decoded ) && isset( $decoded['error']['message'] ) ) {
            return $prefix . ': ' . (string) $decoded['error']['message'];
        }

        if ( is_array( $decoded ) && isset( $decoded['error'] ) && is_string( $decoded['error'] ) ) {
            return $prefix . ': ' . $decoded['error'];
        }

        // Truncate raw body for safety.
        $truncated = substr( (string) $body, 0, 500 );
        return $prefix . ': ' . $truncated;
    }

    private function add_transport_meta( $result, $started_at, $timeout, $http_status = 0 ) {
        $result = is_array( $result ) ? $result : array();
        if ( ! isset( $result['elapsed_ms'] ) ) {
            $result['elapsed_ms'] = (int) round( ( microtime( true ) - (float) $started_at ) * 1000 );
        }
        $result['timeout_seconds'] = intval( $timeout );
        if ( $http_status > 0 && ! isset( $result['http_status'] ) ) {
            $result['http_status'] = intval( $http_status );
        }
        return $result;
    }

    private function add_stream_observation_meta( $result, $events_seen, $data_lines, $bytes_read, $last_event_type ) {
        $result = is_array( $result ) ? $result : array();
        $result['events_seen']     = intval( $events_seen );
        $result['data_lines']      = intval( $data_lines );
        $result['bytes_read']      = intval( $bytes_read );
        $result['last_event_type'] = (string) $last_event_type;
        return $result;
    }

    private function mark_model_request_timing_once( &$timing, $key, $started_at ) {
        if ( ! is_array( $timing ) || '' === (string) $key || ! empty( $timing[ $key ] ) ) {
            return;
        }

        $timing[ $key ] = max( 0, (int) round( ( microtime( true ) - (float) $started_at ) * 1000 ) );
    }

    private function add_model_request_timing_meta( $result, $timing ) {
        $result = is_array( $result ) ? $result : array();
        if ( ! is_array( $timing ) || empty( $timing ) ) {
            return $result;
        }

        $clean = array();
        foreach ( $timing as $key => $value ) {
            $key = sanitize_key( (string) $key );
            if ( '' === $key ) {
                continue;
            }
            $clean[ $key ] = is_numeric( $value ) ? max( 0, intval( $value ) ) : sanitize_text_field( (string) $value );
        }
        if ( ! empty( $clean ) ) {
            $result['model_request_timing'] = $clean;
        }

        return $result;
    }

    /**
     * Build a standardized error result array.
     *
     * @param string $error      Error message.
     * @param string $error_type Error category.
     * @return array Error result in the same format as a successful stream() return.
     */
    private function error_result( $error, $error_type ) {
        return array(
            'success'           => false,
            'role'              => 'assistant',
            'content'           => '',
            'tool_calls'        => array(),
            'stop_reason'       => '',
            'usage'             => array(),
            'assistant_message' => array(),
            'error'             => (string) $error,
            'error_type'        => (string) $error_type,
        );
    }

    /**
     * Stringify message content into a single string.
     *
     * Handles string, scalar, and array content formats.
     *
     * @param mixed $content Content to stringify.
     * @return string String representation.
     */
    private function stringify_content( $content ) {
        if ( is_string( $content ) ) {
            return $content;
        }
        if ( is_scalar( $content ) ) {
            return (string) $content;
        }
        if ( is_array( $content ) ) {
            $parts = array();
            foreach ( $content as $item ) {
                if ( is_string( $item ) ) {
                    $parts[] = $item;
                    continue;
                }
                if ( is_array( $item ) ) {
                    if ( ! empty( $item['text'] ) && is_string( $item['text'] ) ) {
                        $parts[] = $item['text'];
                        continue;
                    }
                    if ( ! empty( $item['content'] ) && is_string( $item['content'] ) ) {
                        $parts[] = $item['content'];
                    }
                }
            }
            if ( ! empty( $parts ) ) {
                return implode( "\n", $parts );
            }
            $encoded = wp_json_encode( $content );
            return is_string( $encoded ) ? $encoded : '';
        }
        return '';
    }

    /**
     * Generate a UUID v4 string.
     *
     * Uses wp_generate_uuid4() if available, otherwise generates one manually.
     *
     * @return string UUID string.
     */
    private function generate_uuid() {
        if ( function_exists( 'wp_generate_uuid4' ) ) {
            return wp_generate_uuid4();
        }

        // Manual UUID v4 generation.
        $data = random_bytes( 16 );
        $data[6] = chr( ( ord( $data[6] ) & 0x0f ) | 0x40 );
        $data[8] = chr( ( ord( $data[8] ) & 0x3f ) | 0x80 );
        return vsprintf( '%s%s-%s-%s-%s-%s%s%s', str_split( bin2hex( $data ), 4 ) );
    }
}
