feat(driver): 支持 SSE 流式响应，并为 worker 崩溃加上可观测性与熔断

src/OneBot/Driver/Event/Http/HttpRequestEvent.php

@@ -67,7 +67,7 @@ public function setErrorHandler(callable $callable)
         $this->error_handler = $callable;
     }
 
-    public function getErrorHandler(): callable
+    public function getErrorHandler(): ?callable
     {
         return $this->error_handler;
     }

src/OneBot/Driver/Interfaces/StreamResponseInterface.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OneBot\Driver\Interfaces;
+
+use Psr\Http\Message\ResponseInterface;
+
+/**
+ * 可分块写出的响应（SSE / 打字机 / 大文件转发）
+ *
+ * 普通响应走 PSR-7 的 body，驱动会等 body 全部就绪后一次性 end()；
+ * 实现了本接口的响应则交给驱动「边算边发」，body 不参与输出。
+ *
+ * 典型实现见 OneBot\Driver\Stream\SseResponse。
+ *
+ * 注意：
+ * - writeTo() 在请求所在的协程/回调里执行，可以在里面做阻塞 IO（读上游 SSE、sleep、轮询），
+ *   Swoole 下这些 IO 会被协程调度，不会阻塞其它请求；
+ * - 一旦开始写数据，HTTP 状态码和响应头就已经发出去了，此时再抛异常无法改成 500，
+ *   驱动层会尽力写一条错误事件再结束流。
+ */
+interface StreamResponseInterface extends ResponseInterface
+{
+    /**
+     * 把响应体逐块写出去
+     *
+     * @param StreamSinkInterface $sink 写出目标；write() 返回 false 时应立即停止生产
+     */
+    public function writeTo(StreamSinkInterface $sink): void;
+}

src/OneBot/Driver/Interfaces/StreamSinkInterface.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OneBot\Driver\Interfaces;
+
+/**
+ * 流式响应的「出水口」抽象
+ *
+ * 驱动层（Swoole / Workerman）各自实现，业务代码只面向本接口写数据，
+ * 因此同一份 SSE 输出逻辑可以跨驱动复用。
+ *
+ * 约定：
+ * - write() 返回 false 表示对端已断开（浏览器关页面、用户按 Ctrl+C），
+ *   生产者应当立刻停止拉取上游数据并退出，否则会白白占用上游连接。
+ * - end() 只能调用一次，且必须在 write() 之后调用，用于结束 chunked 响应。
+ */
+interface StreamSinkInterface
+{
+    /**
+     * 写出一块数据（立刻 flush 到客户端，不做缓冲）
+     *
+     * @param string $chunk 原始字节，SSE 场景通常是一帧或多帧 "data: ...\n\n"
+     *
+     * @return bool 对端是否仍然可写；false 表示连接已关闭，应停止发送
+     */
+    public function write(string $chunk): bool;
+
+    /**
+     * 结束响应（会真正关掉这条 chunked 流）
+     */
+    public function end(): void;
+
+    /**
+     * 连接是否已经关闭（写入失败或已 end 都会置为 true）
+     */
+    public function isClosed(): bool;
+}

src/OneBot/Driver/Stream/SseResponse.php

@@ -0,0 +1,136 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OneBot\Driver\Stream;
+
+use Choir\Http\Response;
+use OneBot\Driver\Interfaces\StreamResponseInterface;
+use OneBot\Driver\Interfaces\StreamSinkInterface;
+
+/**
+ * 开箱即用的 SSE（text/event-stream）响应
+ *
+ * 两种用法：
+ *
+ * 1) 只要标准 SSE 帧，用 send()/comment()：
+ *    return SseResponse::create(function (StreamSinkInterface $sink, SseResponse $res) {
+ *        foreach ($items as $i => $item) {
+ *            if (!$res->send($sink, $item, 'token', (string) $i)) {
+ *                break; // 客户端断开，立刻停止生产
+ *            }
+ *        }
+ *    });
+ *
+ * 2) LLM / 上游 SSE 透传：原样转发上游字节（不要二次解析，省 CPU 也不会改变语义）：
+ *    return SseResponse::create(function (StreamSinkInterface $sink) use ($chunks) {
+ *        while (($chunk = $upstream->read()) !== null) {
+ *            if (!$sink->write($chunk)) {
+ *                break;
+ *            }
+ *        }
+ *    });
+ *
+ * 注意：writer 在请求协程里执行，上游读取可以是阻塞写法（Swoole 会做协程调度）。
+ */
+class SseResponse extends Response implements StreamResponseInterface
+{
+    /** SSE 的默认响应头；同时挡住 nginx 之类的反向代理缓冲（否则前端会「一次性收到全部」） */
+    public const DEFAULT_HEADERS = [
+        'Content-Type' => 'text/event-stream; charset=utf-8',
+        'Cache-Control' => 'no-cache, no-store, must-revalidate',
+        'Connection' => 'keep-alive',
+        'X-Accel-Buffering' => 'no',
+    ];
+
+    /** @var callable(StreamSinkInterface, SseResponse):void */
+    private $writer;
+
+    /**
+     * @param callable(StreamSinkInterface, SseResponse):void $writer
+     * @param array<string, string|string[]>                  $headers 覆盖/追加的响应头
+     */
+    public function __construct(callable $writer, array $headers = [], int $status = 200)
+    {
+        parent::__construct($status, array_merge(self::DEFAULT_HEADERS, $headers));
+        $this->writer = $writer;
+    }
+
+    /**
+     * @param callable(StreamSinkInterface, SseResponse):void $writer
+     * @param array<string, string|string[]>                  $headers
+     */
+    public static function create(callable $writer, array $headers = [], int $status = 200): self
+    {
+        return new self($writer, $headers, $status);
+    }
+
+    public function writeTo(StreamSinkInterface $sink): void
+    {
+        ($this->writer)($sink, $this);
+    }
+
+    /**
+     * 写一帧标准 SSE 事件
+     *
+     * @param mixed       $data  数组/对象会被 json_encode；字符串原样（多行会自动拆成多个 data: 行）
+     * @param null|string $event 事件名（前端 addEventListener($event, ...) 能单独接）
+     * @param null|string $id    事件 ID（断线重连时浏览器会带 Last-Event-ID）
+     * @param null|int    $retry 重连间隔（毫秒）
+     *
+     * @return bool 对端是否仍可写
+     */
+    public function send(StreamSinkInterface $sink, $data, ?string $event = null, ?string $id = null, ?int $retry = null): bool
+    {
+        return $sink->write(self::format($data, $event, $id, $retry));
+    }
+
+    /**
+     * 写一行注释（SSE 规范里的 ": xxx"），常用于心跳保活，避免中间层按空闲超时掐连接
+     */
+    public function comment(StreamSinkInterface $sink, string $text = 'keep-alive'): bool
+    {
+        // 注释行也必须逐行加 ": " 前缀：否则多行文本里的空行会把后面的内容
+        // 变成独立字段（例如 "hb\n\ndata: x" 会被浏览器当成一条 data 事件）。
+        $buffer = '';
+        foreach (preg_split('/\r\n|\r|\n/', $text) as $line) {
+            $buffer .= ': ' . $line . "\n";
+        }
+
+        return $sink->write($buffer . "\n");
+    }
+
+    /**
+     * 把一条数据格式化成 SSE 帧
+     *
+     * @param mixed $data
+     */
+    public static function format($data, ?string $event = null, ?string $id = null, ?int $retry = null): string
+    {
+        if (is_array($data) || is_object($data)) {
+            $data = json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
+        } elseif ($data === null) {
+            $data = '';
+        } else {
+            $data = (string) $data;
+        }
+        $buffer = '';
+        if ($id !== null && $id !== '') {
+            $buffer .= 'id: ' . str_replace(["\r", "\n"], '', $id) . "\n";
+        }
+        if ($event !== null && $event !== '') {
+            $buffer .= 'event: ' . str_replace(["\r", "\n"], '', $event) . "\n";
+        }
+        if ($retry !== null) {
+            $buffer .= 'retry: ' . $retry . "\n";
+        }
+        // 多行数据必须每行一个 data:，否则浏览器只认第一行。
+        // 这里不能 rtrim：SSE 的行为分隔符是 CRLF / LF / CR 三种，且尾部的空行
+        // 代表数据里的换行（format("a\n\n") 应当保留一个换行），trim 掉就是丢数据。
+        foreach (preg_split('/\r\n|\r|\n/', $data) as $line) {
+            $buffer .= 'data: ' . $line . "\n";
+        }
+
+        return $buffer . "\n";
+    }
+}

src/OneBot/Driver/Swoole/SwooleDriver.php

@@ -24,6 +24,7 @@
 use Swoole\Atomic;
 use Swoole\Event;
 use Swoole\Http\Server as SwooleHttpServer;
+use Swoole\Runtime;
 use Swoole\Server;
 use Swoole\Server\Port;
 use Swoole\WebSocket\Server as SwooleWebSocketServer;
@@ -147,6 +148,17 @@ public function run(): void
                 $this->server->addProcess($process);
             }
             $this->server->set($this->server_set);
+
+            // 开启协程 Hook（对应 config: global.swoole_options.coroutine_hook_flags）。
+            // 不开启的话，请求协程里的阻塞 IO（PDO / Redis / stream_socket / usleep 等）
+            // 会真的把整个 worker 进程占住：SSE 这类长连接每挂一个就吃掉一个 worker，
+            // 12 个 worker 被占满后整台服务就只剩心跳。开了之后这些 IO 会自动让出协程。
+            // 只在配置里显式给了 coroutine_hook_flags 时才开，避免改变既有部署的默认行为。
+            $hook_flags = $this->getParam('coroutine_hook_flags', null);
+            if ($hook_flags !== null && class_exists(Runtime::class)) {
+                Runtime::enableCoroutine((int) $hook_flags);
+            }
+
             // 插入一个退出的 exitcode 全局 atomic
             global $_swoole_exit;
             $_swoole_exit = new Atomic();
@@ -235,6 +247,10 @@ private function initServer(): void
         $this->server->on('managerstop', [TopEventListener::getInstance(), 'onManagerStop']);
         $this->server->on('workerstop', [TopEventListener::getInstance(), 'onWorkerStop']);
         $this->server->on('workerexit', [TopEventListener::getInstance(), 'onWorkerExit']);
+        // Worker 异常退出（崩溃）事件：Swoole 中唯一能拿到 exit_code 与 signal 的地方。
+        // 不注册的话，「worker 起来就死、Manager 无限重启」这类问题会完全静默，
+        // 现场只能看到 CPU 被打满，日志里一个字都没有。
+        $this->server->on('workererror', [TopEventListener::getInstance(), 'onWorkerError']);
     }
 
     /**

src/OneBot/Driver/Swoole/SwooleStreamSink.php

@@ -0,0 +1,67 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OneBot\Driver\Swoole;
+
+use OneBot\Driver\Interfaces\StreamSinkInterface;
+use Swoole\Http\Response;
+
+/**
+ * Swoole 驱动的流式出水口
+ *
+ * Swoole 的 Http\Response::write() 会在第一次写入时自动切到
+ * Transfer-Encoding: chunked 并立刻把数据推到 socket，所以这里不需要手动 flush。
+ */
+class SwooleStreamSink implements StreamSinkInterface
+{
+    private Response $response;
+
+    private bool $closed = false;
+
+    public function __construct(Response $response)
+    {
+        $this->response = $response;
+    }
+
+    public function write(string $chunk): bool
+    {
+        if ($this->closed) {
+            return false;
+        }
+        // Swoole 的 write('') 会把空 chunk 发出去，直接跳过
+        if ($chunk === '') {
+            return true;
+        }
+        try {
+            $ok = $this->response->write($chunk);
+        } catch (\Throwable $e) {
+            $ok = false;
+        }
+        if ($ok === false) {
+            // 对端断开（浏览器关页面 / 网络断）时 Swoole 返回 false
+            $this->closed = true;
+            return false;
+        }
+
+        return true;
+    }
+
+    public function end(): void
+    {
+        if ($this->closed) {
+            return;
+        }
+        $this->closed = true;
+        try {
+            $this->response->end();
+        } catch (\Throwable $e) {
+            // 连接已经没了，忽略
+        }
+    }
+
+    public function isClosed(): bool
+    {
+        return $this->closed;
+    }
+}