36b69d300d
Codex 审计发现 124 个问题，修复 59 个 P0-P2 级别问题： P0 (CRITICAL): - TOCTOU 竞态: TransientRateStore/SseConcurrencyGuard 改用 wp_cache_add 原子锁 - REST 路由: __return_true → check_bearer_present 轻量鉴权 - Schema 验证: messages/tools/tool_choice 添加完整 validate_callback - SQL 列名不匹配: upsert 查询对齐 SchemaManager 定义 - SQL 表名插值: 统一使用 %i 占位符 P1 (HIGH): - IP hash 加盐 (HMAC) + IP 解析统一 (context 共享) - Header 注入防护 (X-Request-Id/响应头 \r\n 过滤) - 数值参数范围校验 (temperature/top_p/penalties/max_tokens) - insert_key 错误处理 + revoke_key 审计日志 - Pipeline finally 保护 + RateLimiter rollback 双 store - wp_unslash 补全 + strtotime 返回值检查 - RedisRateStore $rid 冒号验证 - UpstreamStreamClient catch \Throwable P2 (MEDIUM): - AuditLogRepository DRY (提取 build_where_clause) - Token 估算 CJK 优化 (mb_strlen) - ResponseTransformMiddleware null 检查 - 嵌入响应 usage 数据修复 - SchemaManager autoload 优化 25 files changed, ~530 insertions, ~190 deletions Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
89 lines
2.5 KiB
PHP
89 lines
2.5 KiB
PHP
<?php
|
|
/**
|
|
* Upstream Stream Client
|
|
*
|
|
* Wraps the existing PublicAPI::stream() method for SSE proxying,
|
|
* adding cancellation support and token tracking.
|
|
*
|
|
* @package WPMind\Modules\ApiGateway\Stream
|
|
* @since 1.0.0
|
|
*/
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace WPMind\Modules\ApiGateway\Stream;
|
|
|
|
/**
|
|
* Class UpstreamStreamClient
|
|
*
|
|
* Bridges the gateway SSE layer with WPMind's internal
|
|
* streaming API, translating chunks into gateway callbacks
|
|
* while respecting cancellation tokens.
|
|
*/
|
|
final class UpstreamStreamClient {
|
|
|
|
/**
|
|
* Stream a chat completion from the upstream provider.
|
|
*
|
|
* Calls PublicAPI::stream() and forwards each chunk to the
|
|
* provided callback. Checks the cancellation token between
|
|
* chunks and aborts early if cancelled.
|
|
*
|
|
* @param array $messages Chat messages array.
|
|
* @param array $options Provider/model options.
|
|
* @param callable $on_chunk Callback receiving (string $text, array $raw_chunk).
|
|
* @param CancellationToken $token Cancellation token.
|
|
* @return StreamResult Completion result with token count and finish reason.
|
|
*/
|
|
public function stream_chat(
|
|
array $messages,
|
|
array $options,
|
|
callable $on_chunk,
|
|
CancellationToken $token
|
|
): StreamResult {
|
|
$tokens_used = 0;
|
|
$finish_reason = 'stop';
|
|
$error_msg = null;
|
|
$cancelled = false;
|
|
|
|
/**
|
|
* Internal callback passed to PublicAPI::stream().
|
|
*
|
|
* @param string $delta Text delta from the provider.
|
|
* @param array $raw_json Raw JSON chunk from the provider.
|
|
*/
|
|
$callback = function ( string $delta, array $raw_json ) use ( $on_chunk, $token, &$tokens_used, &$cancelled ): void {
|
|
if ( $token->is_cancelled() ) {
|
|
$cancelled = true;
|
|
throw new \RuntimeException( 'Stream cancelled: ' . $token->get_reason() );
|
|
}
|
|
|
|
// Estimate tokens from chunk text (rough: 1 token per 3 chars for mixed CJK/Latin).
|
|
$tokens_used += max( 1, (int) ceil( mb_strlen( $delta, 'UTF-8' ) / 3 ) );
|
|
|
|
$on_chunk( $delta, $raw_json );
|
|
};
|
|
|
|
$api = \WPMind\API\PublicAPI::instance();
|
|
|
|
try {
|
|
/** @var bool|\WP_Error $result */
|
|
$result = $api->stream( $messages, $callback, $options );
|
|
|
|
if ( is_wp_error( $result ) ) {
|
|
$finish_reason = 'error';
|
|
$error_msg = $result->get_error_message();
|
|
}
|
|
} catch ( \Throwable $e ) {
|
|
if ( $cancelled ) {
|
|
$finish_reason = 'cancelled';
|
|
$error_msg = $token->get_reason();
|
|
} else {
|
|
$finish_reason = 'error';
|
|
$error_msg = $e->getMessage();
|
|
}
|
|
}
|
|
|
|
return new StreamResult( $tokens_used, $finish_reason, $error_msg );
|
|
}
|
|
}
|
