feat(onebot): Stream API transport plumbing (#163 phase 1)

Foundation for NapCat-compatible Stream API actions (chunked file
upload/download): a single request answered by many response frames.

- streaming.ts: StreamSink + frame wrappers (wrapStreamFrame /
  wrapStreamTerminal) matching NapCat's wire — each frame an OB11 envelope
  with stream:'stream-action', data.type ∈ stream/response/error, and
  message mirrored into wording.
- ApiHandler: registerStreamAction / isStreamAction; handle() threads a
  sink; processStreamRequest drives WS multi-frame dispatch with optional
  liveness-based abort; dead processRequest removed.
- action-kit: defineStreamAction (sink-aware run, registers as a stream
  action) reusing defineAction's parse/describe.
- adapters: HTTP server streams chunked \r\n\r\n-delimited frames (no
  Content-Length, no false application/json), aborting on client
  disconnect; WS server/client use a backpressure-aware safeSendAsync and
  abort a stream once the socket closes.

No stream actions are registered yet — those land in the next phases.
Verified by two adversarial review passes. 480 onebot tests green.

packages/onebot/src/action-kit.ts

@@ -21,6 +21,7 @@
 import { RETCODE, failedResponse } from './types';
 import type { ApiResponse, JsonObject, JsonValue } from './types';
 import type { ApiActionContext, ApiHandler } from './api-handler';
+import { type StreamSink, NOOP_SINK } from './streaming';
 
 // ─────────────────────────── result currency ───────────────────────────
 
@@ -445,6 +446,63 @@ export function defineAction<S extends Spec>(def: ActionDef<S>): ActionSpec<S> {
   };
 }
 
+// ─────────────────────────── defineStreamAction ───────────────────────────
+// A Stream API action (#163): same param contract + doc machinery as
+// `defineAction`, but `run` additionally receives a `StreamSink` to push
+// intermediate frames, and it registers via `registerStreamAction` so the
+// network adapter streams its output. The handler's return value is still the
+// terminal `ApiResponse` (the final frame). Reuses `defineAction` for
+// parse/describe so the two stay in lock-step.
+
+interface StreamActionDef<S extends Spec> {
+  name: string | readonly [string, ...string[]];
+  summary?: string;
+  returns?: string;
+  returnsSchema?: JsonSchema;
+  readOnly?: boolean;
+  params: S;
+  rules?: (r: RuleBuilders<InferParams<S>>) => readonly CrossFieldRule[];
+  run: (p: InferParams<S>, ctx: ApiActionContext, raw: JsonObject, sink: StreamSink) => Promise<ApiResponse> | ApiResponse;
+}
+
+export interface StreamActionSpec<S extends Spec> {
+  readonly names: string[];
+  readonly params: S;
+  parse(raw: JsonObject): CoerceResult<InferParams<S>>;
+  describe(): ActionDoc;
+  register(h: ApiHandler, ctx: ApiActionContext): void;
+}
+
+export function defineStreamAction<S extends Spec>(def: StreamActionDef<S>): StreamActionSpec<S> {
+  // Borrow defineAction's parse + describe (run is never invoked — register
+  // below installs the sink-aware handler instead).
+  const base = defineAction<S>({
+    name: def.name,
+    summary: def.summary,
+    returns: def.returns,
+    returnsSchema: def.returnsSchema,
+    readOnly: def.readOnly,
+    params: def.params,
+    rules: def.rules,
+    run: () => failedResponse(RETCODE.INTERNAL_ERROR, 'stream action dispatched without a sink'),
+  });
+
+  return {
+    names: base.names,
+    params: def.params,
+    parse: base.parse,
+    describe: base.describe,
+    register: (h, ctx) => {
+      const handler = async (params: JsonObject, sink?: StreamSink): Promise<ApiResponse> => {
+        const r = base.parse(params);
+        if (!r.ok) return failedResponse(RETCODE.BAD_REQUEST, wording(r));
+        return def.run(r.value, ctx, params, sink ?? NOOP_SINK);
+      };
+      for (const name of base.names) h.registerStreamAction(name, handler);
+    },
+  };
+}
+
 // ─────────────────────────── presets ───────────────────────────
 // `group_id` appears in 61/142 actions, `user_id` in 27 — these collapse the
 // dominant shape. The injected fields are surfaced in `describe()` so docs

packages/onebot/src/api-handler.ts

@@ -14,6 +14,7 @@ import { register as registerRequest } from './actions/request';
 import type { ForwardPreviewMeta } from './modules/message-actions';
 import type { JsonObject, JsonValue, MessageMeta } from './types';
 import { RETCODE, failedResponse } from './types';
+import { type StreamSink, wrapStreamFrame, wrapStreamTerminal } from './streaming';
 const moduleLog = createLogger('Bridge.Action');
 
 
@@ -80,7 +81,7 @@ export interface ApiActionContext {
   fetchPttText: (messageId: number) => Promise<{ text: string }>;
 }
 
-type ActionHandler = (params: JsonObject) => Promise<import('./types').ApiResponse>;
+type ActionHandler = (params: JsonObject, sink?: StreamSink) => Promise<import('./types').ApiResponse>;
 
 /** A handled-action record handed to debug observers. */
 export interface ActionRecord {
@@ -93,6 +94,9 @@ export type ActionObserver = (rec: ActionRecord) => void;
 
 export class ApiHandler {
   private readonly handlers = new Map<string, ActionHandler>();
+  /** Names that answer with a multi-frame Stream API response (#163). The
+   *  network adapter consults this BEFORE dispatch to pick streaming output. */
+  private readonly streamActions = new Set<string>();
   private readonly log: Logger;
   /** Debug-stream taps — notified after every handled action. Attached
    *  on-demand (ref-counted) by the WebUI debug stream. */
@@ -122,7 +126,19 @@ export class ApiHandler {
     this.handlers.set(action, handler);
   }
 
-  async handle(action: string, params: JsonObject): Promise<import('./types').ApiResponse> {
+  /** Register a Stream API action — dispatched exactly like a normal action,
+   *  but flagged so adapters stream its frames (the handler receives a sink). */
+  registerStreamAction(action: string, handler: ActionHandler): void {
+    this.handlers.set(action, handler);
+    this.streamActions.add(action);
+  }
+
+  /** Whether `action` answers with a multi-frame Stream API response. */
+  isStreamAction(action: string): boolean {
+    return this.streamActions.has(action);
+  }
+
+  async handle(action: string, params: JsonObject, sink?: StreamSink): Promise<import('./types').ApiResponse> {
     const handler = this.handlers.get(action);
     if (!handler) {
       this.log.debug('unknown action %s', action);
@@ -134,15 +150,16 @@ export class ApiHandler {
     // the wrap + id allocation when trace is actually live, so the default
     // path stays allocation-free.
     if (getLogLevel() !== 'trace') {
-      return this.runAction(action, handler, params);
+      return this.runAction(action, handler, params, sink);
     }
-    return runWithRequestId(nextRequestId(), () => this.runAction(action, handler, params));
+    return runWithRequestId(nextRequestId(), () => this.runAction(action, handler, params, sink));
   }
 
   private async runAction(
     action: string,
     handler: ActionHandler,
     params: JsonObject,
+    sink?: StreamSink,
   ): Promise<import('./types').ApiResponse> {
     // Terse breadcrumb to the log file (debug, always persisted): lets the
     // operator grep "what did the bot get asked to do" in post-mortems.
@@ -154,7 +171,7 @@ export class ApiHandler {
     const startedAt = Date.now();
     let response: import('./types').ApiResponse;
     try {
-      response = await handler(params);
+      response = await handler(params, sink);
       this.log.trace(() => [`${action} ⇒ ${response.status} (${Date.now() - startedAt}ms)`]);
     } catch (error) {
       // Action failures are almost always param-shape problems coming
@@ -185,33 +202,52 @@ export class ApiHandler {
     }
   }
 
-  async processRequest(rawRequest: string): Promise<string> {
-    if (!rawRequest.trim()) {
-      return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request'));
-    }
+  /** WS dispatch supporting Stream API multi-frame responses. A normal action
+   *  emits exactly one frame; a stream action emits each intermediate frame
+   *  then the terminal frame — every frame carries the request's echo. `emit`
+   *  writes one JSON string per frame; awaiting it lets the transport apply
+   *  backpressure. `isAlive`, when supplied, is checked before each stream
+   *  frame — returning false aborts the action (e.g. the client disconnected),
+   *  so a dead client can't make a download keep pumping frames into the void. */
+  async processStreamRequest(
+    rawRequest: string,
+    emit: (json: string) => void | Promise<void>,
+    isAlive?: () => boolean,
+  ): Promise<void> {
+    const bad = (): void => { void emit(JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request'))); };
+    if (!rawRequest.trim()) { bad(); return; }
 
+    let action: string;
+    let params: JsonObject;
+    let echo: JsonValue | undefined;
     try {
       const parsed = JSON.parse(rawRequest) as unknown;
-      if (!isJsonObject(parsed)) {
-        return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request'));
-      }
-
-      const action = asString(parsed.action);
-      if (!action) {
-        return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request'));
-      }
-
-      const params = isJsonObject(parsed.params) ? parsed.params : {};
-      const echo = parsed.echo;
-      const response = await this.handle(action, params);
-      if (echo !== undefined) {
-        response.echo = toJsonValue(echo);
-      }
-
-      return JSON.stringify(response);
+      if (!isJsonObject(parsed)) { bad(); return; }
+      const a = asString(parsed.action);
+      if (!a) { bad(); return; }
+      action = a;
+      params = isJsonObject(parsed.params) ? parsed.params : {};
+      echo = parsed.echo !== undefined ? toJsonValue(parsed.echo) : undefined;
     } catch {
-      return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request'));
+      bad();
+      return;
     }
+
+    if (!this.isStreamAction(action)) {
+      const response = await this.handle(action, params);
+      if (echo !== undefined) response.echo = echo;
+      await emit(JSON.stringify(response));
+      return;
+    }
+
+    const sink: StreamSink = {
+      send: async (frame) => {
+        if (isAlive && !isAlive()) throw new Error('stream transport closed');
+        await emit(JSON.stringify(wrapStreamFrame(frame, echo)));
+      },
+    };
+    const response = await this.handle(action, params, sink);
+    await emit(JSON.stringify(wrapStreamTerminal(response, echo)));
   }
 }
 

packages/onebot/src/network/http-server-adapter.ts

@@ -5,8 +5,10 @@ import {
   type Server,
   type ServerResponse,
 } from 'http';
+import type { ApiHandler } from '../api-handler';
 import type { DispatchPayload } from '../event-filter';
-import type { HttpServerNetwork, JsonObject, JsonValue } from '../types';
+import type { ApiResponse, HttpServerNetwork, JsonObject, JsonValue } from '../types';
+import { type StreamSink, wrapStreamFrame, wrapStreamTerminal } from '../streaming';
 import { IOneBotNetworkAdapter, type AdapterStatus, type NetworkAdapterContext } from './adapter';
 import { isAuthorized, normalizePath } from './utils';
 
@@ -181,6 +183,13 @@ export class HttpServerAdapter extends IOneBotNetworkAdapter<HttpServerNetwork>
         return;
       }
 
+      // Stream API (#163): answer with chunked multi-frame output — each frame
+      // a JSON envelope delimited by `\r\n\r\n`, terminated by the final frame.
+      if (this.ctx.api.isStreamAction(action)) {
+        await streamHttpResponse(res, this.ctx.api, action, params as JsonObject, echo as JsonValue | undefined);
+        return;
+      }
+
       const response = await this.ctx.api.handle(action, params as JsonObject);
       if (echo !== undefined) {
         response.echo = echo as JsonValue;
@@ -212,7 +221,41 @@ function readRequestBody(req: IncomingMessage, maxBytes = 2 * 1024 * 1024): Prom
 }
 
 function writeJson(res: ServerResponse, statusCode: number, data: unknown): void {
+  // A streaming response may already have flushed headers (and even ended); a
+  // late error must not double-send and trip ERR_HTTP_HEADERS_SENT.
+  if (res.headersSent || res.writableEnded) return;
   res.statusCode = statusCode;
   res.setHeader('Content-Type', 'application/json; charset=utf-8');
   res.end(JSON.stringify(data));
 }
+
+/** Stream a multi-frame Stream API response (#163): chunked transfer (no
+ *  Content-Length → Node frames it), each frame a JSON envelope delimited by
+ *  `\r\n\r\n`, the action's terminal response written last. Matches NapCat:
+ *  the body is `\r\n\r\n`-joined JSON objects, NOT a single JSON document, so
+ *  no `application/json` Content-Type is claimed. Each `res.write` is awaited
+ *  (flush callback) for backpressure, and the sink aborts the action once the
+ *  client disconnects so a big download stops pumping into a dead socket. */
+async function streamHttpResponse(
+  res: ServerResponse,
+  api: ApiHandler,
+  action: string,
+  params: JsonObject,
+  echo: JsonValue | undefined,
+): Promise<void> {
+  res.statusCode = 200;
+  const writeFrame = (frame: ApiResponse): Promise<void> =>
+    new Promise((resolve) => {
+      if (res.writableEnded || res.destroyed) { resolve(); return; }
+      res.write(JSON.stringify(frame) + '\r\n\r\n', () => resolve());
+    });
+  const sink: StreamSink = {
+    send: async (frame) => {
+      if (res.writableEnded || res.destroyed) throw new Error('stream client disconnected');
+      await writeFrame(wrapStreamFrame(frame, echo));
+    },
+  };
+  const response = await api.handle(action, params, sink);
+  await writeFrame(wrapStreamTerminal(response, echo));
+  if (!res.writableEnded && !res.destroyed) res.end();
+}

packages/onebot/src/network/utils.ts

@@ -31,6 +31,19 @@ export function safeSend(socket: WebSocket, payload: string, onError?: (err: Err
   });
 }
 
+/** Backpressure-aware send: resolves only once `ws` has flushed `payload` to
+ *  the socket (the `send` callback fires post-write), so a streaming producer
+ *  that awaits this won't outrun a slow client and balloon the send buffer.
+ *  Never rejects — a closed socket or send error resolves quietly (the stream
+ *  sink handles liveness/abort separately so a dead client can't crash the
+ *  per-message handler with an unhandled rejection). */
+export function safeSendAsync(socket: WebSocket, payload: string): Promise<void> {
+  return new Promise((resolve) => {
+    if (socket.readyState !== 1 /* WebSocket.OPEN */) { resolve(); return; }
+    socket.send(payload, () => resolve());
+  });
+}
+
 export function safeClose(socket: WebSocket, code = 1000, reason = 'normal'): void {
   if (socket.readyState === 3 /* CLOSED */ || socket.readyState === 2 /* CLOSING */) return;
   socket.close(code, reason);

packages/onebot/src/network/ws-client-adapter.ts

@@ -8,7 +8,7 @@ import {
 } from '../event-filter';
 import type { JsonObject, WsClientNetwork, WsRole } from '../types';
 import { IOneBotNetworkAdapter, type AdapterStatus, type NetworkAdapterContext } from './adapter';
-import { rawDataToString, safeClose, safeSend } from './utils';
+import { rawDataToString, safeClose, safeSend, safeSendAsync } from './utils';
 
 const moduleLog = createLogger('OneBot.WS-Client');
 const DEFAULT_RECONNECT_INTERVAL_MS = 5000;
@@ -146,8 +146,13 @@ export class WsClientAdapter extends IOneBotNetworkAdapter<WsClientNetwork> {
     if (this.role !== 'Api' && this.role !== 'Universal') return;
     const text = rawDataToString(raw);
     if (!text) return;
-    const response = await this.ctx.api.processRequest(text);
-    safeSend(socket, response);
+    // Stream API (#163): one frame for a normal action, N for a streaming one.
+    // Async send = backpressure; liveness check aborts on disconnect.
+    await this.ctx.api.processStreamRequest(
+      text,
+      (frame) => safeSendAsync(socket, frame),
+      () => socket.readyState === 1,
+    );
   }
 
   private sendBootstrapMetaEvents(socket: WebSocket): void {

packages/onebot/src/network/ws-server-adapter.ts

@@ -9,7 +9,7 @@ import {
 } from '../event-filter';
 import type { JsonObject, WsRole, WsServerNetwork } from '../types';
 import { IOneBotNetworkAdapter, type AdapterStatus, type NetworkAdapterContext } from './adapter';
-import { isAuthorized, normalizePath, parseRequestPath, rawDataToString, safeClose, safeSend } from './utils';
+import { isAuthorized, normalizePath, parseRequestPath, rawDataToString, safeClose, safeSend, safeSendAsync } from './utils';
 
 const moduleLog = createLogger('OneBot.WS-Server');
 
@@ -119,7 +119,9 @@ export class WsServerAdapter extends IOneBotNetworkAdapter<WsServerNetwork> {
     this.connections.set(socket, conn);
 
     socket.on('message', (raw: Buffer) => {
-      void this.handleApiMessage(socket, role, raw);
+      void this.handleApiMessage(socket, role, raw).catch((err) => {
+        this.log.warn('[%s] handleApiMessage threw: %s', this.name, err instanceof Error ? (err.stack ?? err.message) : String(err));
+      });
     });
     socket.on('close', () => this.connections.delete(socket));
     socket.on('error', (err: Error) => {
@@ -135,8 +137,15 @@ export class WsServerAdapter extends IOneBotNetworkAdapter<WsServerNetwork> {
     if (role !== 'Api' && role !== 'Universal') return;
     const text = rawDataToString(raw);
     if (!text) return;
-    const response = await this.ctx.api.processRequest(text);
-    safeSend(socket, response);
+    // Stream API (#163) emits multiple frames per request; processStreamRequest
+    // sends one frame for a normal action and N for a streaming one. The async
+    // send applies backpressure (awaits flush); the liveness check aborts a
+    // streaming action once the client goes away.
+    await this.ctx.api.processStreamRequest(
+      text,
+      (frame) => safeSendAsync(socket, frame),
+      () => socket.readyState === 1,
+    );
   }
 
   private sendBootstrapMetaEvents(socket: WebSocket): void {

packages/onebot/src/streaming.ts

@@ -0,0 +1,75 @@
+// streaming — Stream API plumbing (#163, NapCat-compatible).
+//
+// A handful of OneBot actions ("Stream API") answer a single request with
+// MANY response frames instead of one: an upload acks each chunk, a download
+// pushes the file out chunk-by-chunk. SnowLuma's normal dispatch is strictly
+// one-request → one-response, so streaming needs a small seam:
+//
+//   • a `StreamSink` the action calls to emit intermediate frames, and
+//   • per-adapter wiring that writes each frame on the wire (HTTP chunked /
+//     repeated WS messages) instead of closing after the first.
+//
+// Wire format mirrors NapCat exactly so its client scripts work unchanged:
+// every frame is an OB11 envelope carrying `stream:'stream-action'`, and the
+// frame `data` is a StreamPacket whose `type` is one of stream/response/error.
+
+import type { ApiResponse, JsonValue } from './types';
+
+/** Marks an OB11 envelope as one frame of a streaming response. */
+export const STREAM_MARK = 'stream-action';
+
+/** Frame `type` discriminator carried inside each frame's `data`. */
+export enum StreamStatus {
+  /** An intermediate chunk frame (transfer in progress). */
+  Stream = 'stream',
+  /** The terminal frame — the stream finished successfully. */
+  Response = 'response',
+  /** A stream was reset / aborted by the caller. */
+  Reset = 'reset',
+  /** The terminal frame — the stream failed. */
+  Error = 'error',
+}
+
+/** A sink an action uses to push intermediate stream frames. Each `send`
+ *  carries the raw frame DATA (a StreamPacket: `{ type, data_type, ... }`);
+ *  the network adapter wraps it in an OB11 envelope + the stream marker and
+ *  flushes it on the wire. The action's eventual return value is the terminal
+ *  frame, so do NOT `send` the final result through the sink. */
+export interface StreamSink {
+  send(frame: JsonValue): Promise<void>;
+}
+
+/** Fallback sink for when a stream action is invoked over a transport that
+ *  can't stream (or by a plain client): intermediate frames are dropped, the
+ *  terminal response is still returned normally. */
+export const NOOP_SINK: StreamSink = { send: async () => { /* dropped */ } };
+
+/** Wrap a raw stream-frame `data` payload into an OB11 envelope carrying the
+ *  stream marker (mirrors NapCat's `OB11Response.ok(data, echo, true)`). The
+ *  empty `message`/`wording` pair matches NapCat — clients reading either key
+ *  on an intermediate frame get `''`, never `undefined`. */
+export function wrapStreamFrame(data: JsonValue, echo: JsonValue | undefined): ApiResponse {
+  const frame: ApiResponse = { status: 'ok', retcode: 0, data, message: '', wording: '', stream: STREAM_MARK };
+  if (echo !== undefined) frame.echo = echo;
+  return frame;
+}
+
+/** Turn an action's terminal `ApiResponse` into the final streaming frame:
+ *  attach the stream marker + echo, mirror `wording` into `message` (NapCat
+ *  sets both), and — on failure — normalise `data` to an error packet
+ *  (`{ type:'error' }`) so NapCat clients see the error `type` while the
+ *  human-readable reason stays in `message`/`wording`.
+ *
+ *  Contract: on success the action's `data` MUST carry `type:'response'` (the
+ *  client treats that frame as the stream terminator). Stream actions are
+ *  responsible for that, mirroring NapCat's `BaseDownloadStream`. */
+export function wrapStreamTerminal(response: ApiResponse, echo: JsonValue | undefined): ApiResponse {
+  const frame: ApiResponse = { ...response, stream: STREAM_MARK };
+  if (echo !== undefined) frame.echo = echo;
+  if (response.status === 'failed') {
+    frame.data = { type: StreamStatus.Error, data_type: 'error' };
+  }
+  frame.message = frame.wording ?? '';
+  if (frame.wording === undefined) frame.wording = '';
+  return frame;
+}

packages/onebot/src/types.ts

@@ -7,6 +7,13 @@ export interface ApiResponse {
   data: JsonValue;
   echo?: JsonValue;
   wording?: string;
+  /** Mirror of `wording` carried on Stream API frames — NapCat puts the
+   *  message text in BOTH `message` and `wording`, so clients reading either
+   *  key work. Only set on stream frames. */
+  message?: string;
+  /** Only set on Stream API frames — `'stream-action'`. Marks an envelope as
+   *  one frame of a multi-frame streaming response (NapCat Stream API wire). */
+  stream?: string;
 }
 
 export interface OneBotRequest {

packages/onebot/tests/http-server-content-type.test.ts

@@ -18,6 +18,7 @@ function fakeCtx(handleFn: (action: string, params: Record<string, unknown>) =>
         const data = handleFn(action, params);
         return { status: 'ok', retcode: 0, data };
       }),
+      isStreamAction: () => false,
     } as any,
     buildLifecycleEvent: vi.fn(() => ({})),
     buildHeartbeatEvent: vi.fn(() => ({})),

packages/onebot/tests/network-bootstrap-meta.test.ts

@@ -68,7 +68,7 @@ import type { JsonObject, WsClientNetwork, WsServerNetwork } from '../src/types'
 function ctx(): NetworkAdapterContext {
   return {
     uin: '10001',
-    api: { processRequest: async () => '' } as never,
+    api: { processStreamRequest: async () => {} } as never,
     buildLifecycleEvent: (sub) => ({ post_type: 'meta_event', meta_event_type: 'lifecycle', sub_type: sub }),
     buildHeartbeatEvent: () => ({ post_type: 'meta_event', meta_event_type: 'heartbeat', interval: 5000 }),
   };

packages/onebot/tests/network-manager.test.ts

@@ -52,7 +52,7 @@ class FakeAdapter extends IOneBotNetworkAdapter<FakeNetworkConfig> {
 
 const NULL_CTX: NetworkAdapterContext = {
   uin: '10001',
-  api: { handle: async () => ({ status: 'failed', retcode: 0, data: null }), processRequest: async () => '' } as never,
+  api: { handle: async () => ({ status: 'failed', retcode: 0, data: null }), processStreamRequest: async () => {} } as never,
   buildLifecycleEvent: () => ({}),
   buildHeartbeatEvent: () => ({}),
 };

packages/onebot/tests/streaming.test.ts

@@ -0,0 +1,167 @@
+// Phase 1 — Stream API plumbing (#163). Verifies the dispatch seam:
+// defineStreamAction registers a streaming handler, processStreamRequest
+// emits N intermediate frames + a terminal frame (all echo-tagged, marked
+// `stream:'stream-action'`), normal actions stay single-frame, and failures
+// terminate with an error frame. Wire format mirrors NapCat's Stream API.
+import { describe, it, expect } from 'vitest';
+import { ApiHandler, type ApiActionContext } from '../src/api-handler';
+import { defineAction, defineStreamAction, f } from '../src/action-kit';
+import { okResponse, failedResponse } from '../src/types';
+import { wrapStreamFrame, wrapStreamTerminal, STREAM_MARK } from '../src/streaming';
+
+function newHandler(): ApiHandler {
+  // The real register* run lazily (handlers capture ctx, never call it at
+  // registration), so an empty ctx is enough to exercise the dispatch seam.
+  return new ApiHandler({} as ApiActionContext);
+}
+
+const parse = (frames: string[]) => frames.map((f) => JSON.parse(f) as Record<string, unknown>);
+
+describe('Stream API plumbing', () => {
+  it('streams N intermediate frames + a terminal frame, all echo-tagged', async () => {
+    const h = newHandler();
+    defineStreamAction({
+      name: 'demo_stream',
+      params: { n: f.uint() },
+      run: async (p, _ctx, _raw, sink) => {
+        for (let i = 0; i < p.n; i++) await sink.send({ type: 'stream', index: i });
+        return okResponse({ type: 'response', total: p.n });
+      },
+    }).register(h, {} as ApiActionContext);
+
+    const frames: string[] = [];
+    await h.processStreamRequest(
+      JSON.stringify({ action: 'demo_stream', params: { n: 2 }, echo: 'e1' }),
+      (j) => frames.push(j),
+    );
+
+    const got = parse(frames);
+    expect(got).toHaveLength(3);
+    expect(got[0]).toMatchObject({ status: 'ok', retcode: 0, stream: STREAM_MARK, echo: 'e1', data: { type: 'stream', index: 0 } });
+    expect(got[1]).toMatchObject({ stream: STREAM_MARK, echo: 'e1', data: { type: 'stream', index: 1 } });
+    expect(got[2]).toMatchObject({ status: 'ok', stream: STREAM_MARK, echo: 'e1', data: { type: 'response', total: 2 } });
+  });
+
+  it('marks the action as a stream action; normal actions are not', () => {
+    const h = newHandler();
+    defineStreamAction({ name: 'demo_stream', params: {}, run: () => okResponse({ type: 'response' }) }).register(h, {} as ApiActionContext);
+    defineAction({ name: 'demo_normal', params: {}, run: () => okResponse({ ok: true }) }).register(h, {} as ApiActionContext);
+    expect(h.isStreamAction('demo_stream')).toBe(true);
+    expect(h.isStreamAction('demo_normal')).toBe(false);
+    expect(h.isStreamAction('nope')).toBe(false);
+  });
+
+  it('a normal action through processStreamRequest emits exactly one un-marked frame', async () => {
+    const h = newHandler();
+    defineAction({ name: 'demo_normal', params: {}, run: () => okResponse({ ok: true }) }).register(h, {} as ApiActionContext);
+
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'demo_normal', params: {}, echo: 'e2' }), (j) => frames.push(j));
+
+    expect(frames).toHaveLength(1);
+    const r = parse(frames)[0];
+    expect(r).toMatchObject({ status: 'ok', echo: 'e2', data: { ok: true } });
+    expect(r.stream).toBeUndefined();
+  });
+
+  it('a throwing stream action terminates with a single error frame', async () => {
+    const h = newHandler();
+    defineStreamAction({
+      name: 'demo_boom',
+      params: {},
+      run: async () => { throw new Error('boom'); },
+    }).register(h, {} as ApiActionContext);
+
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'demo_boom', params: {}, echo: 'e3' }), (j) => frames.push(j));
+
+    expect(frames).toHaveLength(1);
+    expect(parse(frames)[0]).toMatchObject({
+      status: 'failed', stream: STREAM_MARK, echo: 'e3', wording: 'boom', data: { type: 'error', data_type: 'error' },
+    });
+  });
+
+  it('emits already-sent frames before the terminal error when a stream throws mid-way', async () => {
+    const h = newHandler();
+    defineStreamAction({
+      name: 'demo_partial',
+      params: {},
+      run: async (_p, _ctx, _raw, sink) => {
+        await sink.send({ type: 'stream', index: 0 });
+        throw new Error('mid');
+      },
+    }).register(h, {} as ApiActionContext);
+
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'demo_partial', params: {} }), (j) => frames.push(j));
+
+    const got = parse(frames);
+    expect(got).toHaveLength(2);
+    expect(got[0]).toMatchObject({ stream: STREAM_MARK, data: { type: 'stream', index: 0 } });
+    expect(got[1]).toMatchObject({ status: 'failed', data: { type: 'error' } });
+  });
+
+  it('stream param validation fails terminate with an error frame', async () => {
+    const h = newHandler();
+    defineStreamAction({ name: 'demo_needs', params: { n: f.uint() }, run: () => okResponse({ type: 'response' }) }).register(h, {} as ApiActionContext);
+
+    const frames: string[] = [];
+    await h.processStreamRequest(JSON.stringify({ action: 'demo_needs', params: {} }), (j) => frames.push(j));
+    expect(frames).toHaveLength(1);
+    expect(parse(frames)[0]).toMatchObject({ status: 'failed', stream: STREAM_MARK, data: { type: 'error' } });
+  });
+
+  it('aborts a stream when isAlive turns false (client disconnect)', async () => {
+    const h = newHandler();
+    let sent = 0;
+    defineStreamAction({
+      name: 'demo_abort',
+      params: {},
+      run: async (_p, _ctx, _raw, sink) => {
+        for (let i = 0; i < 5; i++) { await sink.send({ type: 'stream', index: i }); sent++; }
+        return okResponse({ type: 'response' });
+      },
+    }).register(h, {} as ApiActionContext);
+
+    const frames: string[] = [];
+    let alive = true;
+    await h.processStreamRequest(
+      JSON.stringify({ action: 'demo_abort', params: {} }),
+      (j) => { frames.push(j); if (frames.length >= 2) alive = false; },
+      () => alive,
+    );
+
+    expect(sent).toBeLessThan(5); // loop aborted once the client "left"
+    const got = parse(frames);
+    expect(got[got.length - 1]).toMatchObject({ status: 'failed', data: { type: 'error' } });
+  });
+
+  it('wrapStreamFrame / wrapStreamTerminal shape (NapCat message+wording mirror)', () => {
+    expect(wrapStreamFrame({ type: 'stream', x: 1 }, 'e')).toEqual({
+      status: 'ok', retcode: 0, data: { type: 'stream', x: 1 }, message: '', wording: '', stream: STREAM_MARK, echo: 'e',
+    });
+    // no echo → no echo key
+    expect(wrapStreamFrame({ type: 'stream' }, undefined)).toEqual({
+      status: 'ok', retcode: 0, data: { type: 'stream' }, message: '', wording: '', stream: STREAM_MARK,
+    });
+    // failed terminal → data normalised to an error packet; reason in BOTH message + wording
+    expect(wrapStreamTerminal(failedResponse(1404, 'nope'), 'e')).toMatchObject({
+      status: 'failed', retcode: 1404, message: 'nope', wording: 'nope', stream: STREAM_MARK, echo: 'e',
+      data: { type: 'error', data_type: 'error' },
+    });
+    // ok terminal → data passes through, message/wording empty
+    expect(wrapStreamTerminal(okResponse({ type: 'response', a: 1 }), undefined)).toMatchObject({
+      status: 'ok', stream: STREAM_MARK, message: '', data: { type: 'response', a: 1 },
+    });
+  });
+
+  it('malformed stream requests emit one bad-request frame', async () => {
+    const h = newHandler();
+    const frames: string[] = [];
+    await h.processStreamRequest('not json', (j) => frames.push(j));
+    await h.processStreamRequest('', (j) => frames.push(j));
+    await h.processStreamRequest(JSON.stringify({ params: {} }), (j) => frames.push(j)); // no action
+    expect(frames).toHaveLength(3);
+    for (const f of parse(frames)) expect(f).toMatchObject({ status: 'failed' });
+  });
+});

packages/onebot/tests/ws-client-stale-close.test.ts

@@ -70,7 +70,7 @@ import type { WsClientNetwork } from '../src/types';
 function ctx(): NetworkAdapterContext {
   return {
     uin: '10001',
-    api: { processRequest: async () => '' } as never,
+    api: { processStreamRequest: async () => {} } as never,
     buildLifecycleEvent: () => ({}),
     buildHeartbeatEvent: () => ({}),
   };