WIP: PR #8 plus guarded Thinking and reliable stream errors

.env.example

@@ -29,6 +29,28 @@ CNB_REPO_SLUG=your-org/ai-proxy
 # Per-stream idle watchdog (ms)
 #PROXY_IDLE_TIMEOUT_MS=300000
 
+# ---- thinking (optional, off by default) ----
+# Forward the client's thinking toggles (reasoning_effort / reasoning /
+# enable_thinking / thinking, and Anthropic thinking:{type:enabled}) to the
+# upstream thinking variant instead of stripping them.
+#
+# Why this is off by default: the thinking variant bills its reasoning phase
+# against max_tokens. With a small budget the reasoning consumes everything and
+# the answer is never generated (finish_reason=length, empty content).
+#
+# Turn it on only if your clients send a large max_tokens (>= 32k is a safe
+# starting point; 60k was verified). Requests that carry no thinking toggle are
+# unaffected either way.
+# Legacy alias (equivalent to mode=passthrough when mode is not explicitly set):
+#CNB2API_ALLOW_THINKING=1
+#
+# Preferred mode: off (default), guarded, or passthrough.
+# guarded returns HTTP 400 for requests explicitly enabling Thinking unless
+# max_tokens is an integer >= CNB2API_THINKING_MIN_TOKENS. This DOES NOT cap
+# reasoning output or guarantee the upstream will produce a final answer.
+#CNB2API_THINKING_MODE=guarded
+#CNB2API_THINKING_MIN_TOKENS=32768
+
 # ---- self-registration to the fixed-domain relay (optional) ----
 # Full URL of the relay endpoint, e.g. https://ai.example.com/ops/register
 #REGISTER_URL=

.github/workflows/test.yml

@@ -2,8 +2,9 @@ name: test
 
 on:
   push:
-    branches: [main]
+    branches: [main, feat/reliable-thinking-streams]
   pull_request:
+  workflow_dispatch:
 
 jobs:
   test:
@@ -13,6 +14,9 @@ jobs:
       - uses: actions/setup-node@v4
         with:
           node-version: 22
+      - name: JavaScript syntax check
+        run: |
+          for file in src/*.mjs test/*.mjs bin/*.mjs; do node --check "$file"; done
       - run: node --test
       - name: deploy scripts syntax
         run: |

README.md

@@ -176,9 +176,29 @@ npm run quota                 # or: npx cnb2api-quota
 | `PROXY_PORT` | `9001` | Listen port. |
 | `PROXY_UPSTREAM_TIMEOUT_MS` | `15000` | Upstream connect / first-byte timeout (ms). |
 | `PROXY_IDLE_TIMEOUT_MS` | `300000` | Per-stream idle watchdog (ms). |
+| `CNB2API_THINKING_MODE` | `off` | `off`: remove thinking toggles (original default); `guarded`: allow thinking only when the request declares a sufficient total token budget; `passthrough`: forward thinking toggles unchanged. |
+| `CNB2API_THINKING_MIN_TOKENS` | `32768` | Minimum explicit `max_tokens` in guarded mode. This is a warning threshold, **not** an upstream reasoning-token cap. |
+| `CNB2API_ALLOW_THINKING` | off | Legacy alias: `1` selects `passthrough` when `CNB2API_THINKING_MODE` is unset. Set to `1` to forward client thinking toggles (`reasoning_effort` / `reasoning` / `enable_thinking` / `thinking`, and Anthropic `thinking:{type:"enabled"}`) to the upstream thinking variant instead of stripping them. Only use it if your clients send a large `max_tokens` — see the thinking note below. |
 | `REGISTER_URL` | — (optional) | Relay `/ops/register` endpoint for self-registration. |
 | `REG_TOKEN` | — (optional) | Shared secret for self-registration. |
 
+### Thinking is stripped by default
+
+The upstream thinking variant bills its reasoning phase against `max_tokens`. With a
+small budget the reasoning consumes the whole allowance, the stream ends with
+`finish_reason: length` and **no answer is ever generated** (measured: a hard prompt
+at `max_tokens` 2048 and 8192 produced 4.6k / 20.3k characters of reasoning and zero
+content). To keep clients from hitting that, thinking toggles are removed from every
+outbound request.
+
+That default is safe but absolute: it also locks out callers that *do* provide a large
+budget, for whom thinking works correctly (verified at `max_tokens: 60000`, where
+reasoning and answer are emitted in sequence). Set `CNB2API_ALLOW_THINKING=1` to opt in
+and take responsibility for the budget. Requests that carry no thinking toggle behave
+identically in all modes. In `guarded` mode, a client explicitly requesting Thinking with `max_tokens` below 32768 (or with no numeric budget) receives HTTP 400 *before* the upstream request. This is only a preflight check; CNB may still exhaust larger budgets. Upstream SSE streams with no terminal finish reason are reported as errors instead of fabricated successful completions.
+
+**CNB branch caution:** while validating a feature branch, make sure its workspace runs the same branch. The existing `.cnb.yml` keepalive is pinned to `main` and can restart a `main` workspace. Do not merge before tests.
+
 ---
 
 ## FAQ

README.zh-CN.md

@@ -176,9 +176,25 @@ npm run quota                 # 或: npx cnb2api-quota
 | `PROXY_PORT` | `9001` | 服务监听端口。 |
 | `PROXY_UPSTREAM_TIMEOUT_MS` | `15000` | 上游连接首字节超时 (ms)。 |
 | `PROXY_IDLE_TIMEOUT_MS` | `300000` | 流式事件空闲超时看门狗 (ms)。 |
+| `CNB2API_THINKING_MODE` | `off` | `off`：默认剥除思考参数；`guarded`：仅在请求携带足够大 `max_tokens` 时放行；`passthrough`：完全透传思考开关。 |
+| `CNB2API_THINKING_MIN_TOKENS` | `32768` | `guarded` 模式要求的最小 `max_tokens`；这是预检阈值，**不是**上游推理 token 的实际限额。 |
+| `CNB2API_ALLOW_THINKING` | 关闭 | 兼容旧版：当未配置 `CNB2API_THINKING_MODE` 时，设为 `1` 等同于 `passthrough`。设为 `1` 时，把客户端的思考开关（`reasoning_effort` / `reasoning` / `enable_thinking` / `thinking`，以及 Anthropic 的 `thinking:{type:"enabled"}`）透传到上游 thinking 变体，而不是出站剥除。仅在客户端会给出足够大 `max_tokens` 时开启——详见下方说明。 |
 | `REGISTER_URL` | — (可选) | 中继自注册 `/ops/register` 地址。 |
 | `REG_TOKEN` | — (可选) | 中继自注册鉴权密钥。 |
 
+### 默认剥除思考开关
+
+上游 thinking 变体会把推理阶段计入 `max_tokens`。预算偏小时推理吃满全部额度，流以
+`finish_reason: length` 结束且**答案永不产出**（实测：硬 prompt 在 `max_tokens` 2048 与
+8192 下分别产出 4.6k / 20.3k 字符推理、正文为零）。为避免客户端踩到该退化形态，出站请求
+统一剥除思考开关。
+
+该默认值安全但绝对：它同时把**给了足够预算**的调用方一并挡在门外，而这类调用方本来是能
+正常思考的（`max_tokens: 60000` 实测推理与正文依次产出）。设 `CNB2API_ALLOW_THINKING=1`
+即可显式开启，并由调用方自行保证预算。未携带思考开关的请求在两种设置下行为完全一致。`guarded` 模式下，若请求显式开启 Thinking，但 `max_tokens` 小于 32768（或没有合法的数值），会在访问上游前返回 HTTP 400。阈值只是本地预检，不保证 CNB 上游一定能在预算内答完。
+
+**CNB 分支注意：**在开发分支验证时，确保工作区也运行该分支。原 `.cnb.yml` 的保活任务固定指向 `main`，自动恢复时可能启动未改进的旧版本。不要在测试通过前合并到 `main`。
+
 ---
 
 ## 常见问题 (FAQ)

src/anthropic.mjs

@@ -57,7 +57,7 @@ function neutralize(text) {
   return t;
 }
 
-export function fromAnthropicRequest(body) {
+export function fromAnthropicRequest(body, { allowThinking = false } = {}) {
   if (!body || typeof body !== 'object') throw new ProtocolError('request body must be a JSON object');
   if (!Array.isArray(body.messages) || body.messages.length === 0) {
     throw new ProtocolError('messages: must be a non-empty array');
@@ -108,6 +108,17 @@ export function fromAnthropicRequest(body) {
     const tc = convertToolChoice(body.tool_choice);
     if (tc) out.tool_choice = tc;
   }
+
+  // 思考开关（默认丢弃，见下）：Anthropic 用 thinking:{type,budget_tokens} 表达，
+  // 上游 OpenAI 端点认 thinking:{type:enabled} + reasoning_effort。仅当部署显式开启
+  // CNB2API_ALLOW_THINKING=1 时映射透传；默认丢弃，理由与 proxy.mjs 出站剥除同一
+  // 教训——thinking 变体把推理计入 max_tokens，预算不足时 content 永不产出。
+  // budget_tokens 不映射：上游没有对应字段，强度用 reasoning_effort 表达。
+  if (allowThinking && body.thinking && typeof body.thinking === 'object'
+    && body.thinking.type === 'enabled') {
+    out.thinking = { type: 'enabled' };
+    if (typeof body.thinking.effort === 'string') out.reasoning_effort = body.thinking.effort;
+  }
   return out;
 }
 
@@ -349,10 +360,16 @@ export class AnthropicStream {
     return [...head, ...this.out];
   }
 
-  // 上游中途夭折：Anthropic 规范无 abort 事件，尽力收尾（客户端看到的是截断的完整消息）
+  // Upstream stream failure must never be disguised as a successful message_stop.
   abort() {
     if (this.finished) return [];
-    return this.end();
+    this.finished = true;
+    this.out.length = 0;
+    this.out.push(evt('error', {
+      type: 'error',
+      error: { type: 'api_error', message: 'upstream stream interrupted' },
+    }));
+    return [...this.out];
   }
 
   deltaFor(type, deltaEvent, partial) {

src/config.mjs

@@ -1,4 +1,8 @@
+import { THINKING_MODES } from './thinking.mjs';
+
 // Central env loading + fail-fast validation.
+const thinkingMode = process.env.CNB2API_THINKING_MODE
+  || (process.env.CNB2API_ALLOW_THINKING === '1' ? 'passthrough' : 'off');
 // The proxy refuses to start with a missing/weak key instead of running exposed.
 export const config = {
   port: Number(process.env.PROXY_PORT || 9001),
@@ -21,6 +25,11 @@ export const config = {
   idleTimeoutMs: Number(process.env.PROXY_IDLE_TIMEOUT_MS || 300_000),        // per-stream idle cap (reset each chunk)
   authFailWindowMs: 60_000,
   authFailMax: 10,
+  // Off by default. Legacy CNB2API_ALLOW_THINKING=1 retains passthrough behavior.
+  // guarded checks max_tokens before forwarding but cannot cap upstream reasoning.
+  thinkingMode,
+  thinkingMinTokens: Number(process.env.CNB2API_THINKING_MIN_TOKENS || 32768),
+  allowThinking: thinkingMode !== 'off',
   upstreamUrl: '',
 };
 
@@ -43,6 +52,14 @@ if (!config.upstreamUrl) {
   console.error('[config] upstream URL is empty: set CNB_REPO_SLUG (org/repo) or UPSTREAM_OVERRIDE. Refusing to start.');
   process.exit(1);
 }
+if (!THINKING_MODES.includes(config.thinkingMode)) {
+  console.error('[config] CNB2API_THINKING_MODE must be off, guarded, or passthrough. Refusing to start.');
+  process.exit(1);
+}
+if (!Number.isSafeInteger(config.thinkingMinTokens) || config.thinkingMinTokens <= 0) {
+  console.error('[config] CNB2API_THINKING_MIN_TOKENS must be a positive integer. Refusing to start.');
+  process.exit(1);
+}
 // 数字环境变量塞非法值时 Number() 得 NaN，setTimeout(fn, NaN) 按 0ms 立即触发 →
 // 每条上游流瞬间 abort，全线不可用。fail-fast 在启动时拦下，而不是带病运行。
 for (const [key, env] of [['upstreamTimeoutMs', 'PROXY_UPSTREAM_TIMEOUT_MS'], ['idleTimeoutMs', 'PROXY_IDLE_TIMEOUT_MS']]) {

src/proxy.mjs

@@ -10,9 +10,18 @@ import { observe as observeModel } from './models.mjs';
 // content 全空（答案永不生成，真机实测 2048/8192 token 必现）。上游无正确开关可配，
 // 出站统一剥除、锁定非 thinking 变体（与 /v1/messages 请求侧丢弃 thinking 块同一策略）。
 // 表驱动：上游若引入新的思考开关命名，往这张表加一项即可。
+//
+// 逃生口：剥除是"安全默认值"，不是"唯一正确策略"。它把客户端永久锁死在非 thinking
+// 变体上，思考能力整体消失——2026-10-08 同步本提交后，某下游的 CNB 侧 reasoning 从
+// 每天数百万字符直接归零，且试遍参数拼写（reasoningEffort / enableThinking /
+// chat_template_kwargs / extra_body / thinking_budget）都无法再触达 thinking 变体。
+// 会翻车的组合其实只是"思考开关 + 过小的 max_tokens"；max_tokens 给够时推理与正文会
+// 依次产出（2026-09-18/09-20 真实流量在 max_tokens=60000 下完全正常）。
+// 因此：默认仍剥除，工作区显式设 CNB2API_ALLOW_THINKING=1 时放行，由客户端自负预算。
 const REASONING_TRIGGERS = ['reasoning_effort', 'reasoning', 'enable_thinking', 'thinking'];
 
 function stripReasoningTriggers(payload) {
+  if (config.allowThinking) return payload;
   if (!REASONING_TRIGGERS.some((k) => k in payload)) return payload;
   const out = { ...payload };
   for (const k of REASONING_TRIGGERS) delete out[k];
@@ -83,6 +92,7 @@ export async function forward({ reqId, payload, wantStream, res, transform }) {
     let sseTail = '';      // 跨 chunk 的半行缓冲（SSE 行可能被 TCP 分包截断）
     let usage = null;      // 流式路径从尾部 chunk 提取 usage（OpenAI 约定：最后一个 data 带 usage）
     let ok = false;        // 流正常走完（未中途 abort/停滞）
+    let sawFinishReason = false; // A clean EOF without a final choice is still an incomplete response.
     let clientGone = false; // 客户端主动断开（正常行为，与上游流故障区分：不计 errors）
     // 行处理：usage 提取（所有路径）+ 协议转换（挂 transform 时逐行喂流式状态机）
     const handleLine = (line) => {
@@ -92,6 +102,8 @@ export async function forward({ reqId, payload, wantStream, res, transform }) {
       if (!data || data === '[DONE]') return;
       let j;
       try { j = JSON.parse(data); } catch { return; /* 心跳/非 JSON 行忽略 */ }
+      if (j.error) throw new Error('upstream stream returned an error event');
+      if (j.choices?.some((choice) => choice?.finish_reason)) sawFinishReason = true;
       if (j.usage) usage = j.usage;
       if (j.model) observeModel(j.model); // 嗅探上游真实模型名（无 list 接口，只能从响应读）
       if (transform) writeEvents(transform.stream.feed(j));
@@ -118,7 +130,12 @@ export async function forward({ reqId, payload, wantStream, res, transform }) {
     });
     try {
       for (;;) {
-        if (res.destroyed || ac.signal.aborted) { clientGone = true; await reader.cancel().catch(() => {}); break; }
+        if (res.destroyed || ac.signal.aborted) {
+          clientGone = true;
+          ac.abort();
+          void reader.cancel().catch(() => {});
+          break;
+        }
         const { done, value } = await readChunk();
         if (done) break;
         if (transform) {
@@ -134,18 +151,24 @@ export async function forward({ reqId, payload, wantStream, res, transform }) {
         }
       }
       // 收尾 flush：末尾 usage chunk 若不以换行结尾（上游直接 FIN），残余在 sseTail 里，此处补解析
-      decodeAndParse(null);
-      if (transform) writeEvents(transform.stream.end());
-      if (!res.destroyed) res.end();
-      ok = true;
+      if (!clientGone) {
+        decodeAndParse(null);
+        if (!sawFinishReason) throw new Error('upstream stream ended without finish_reason');
+        if (transform) writeEvents(transform.stream.end());
+        if (!res.destroyed) res.end();
+        ok = true;
+      }
     } catch (e) {
       clearTimeout(idle);
-      await reader.cancel().catch(() => {});
-      ac.abort();
+      ac.abort(); // Abort pending read before cancellation; a stuck reader.cancel() must not hang error delivery.
+      void reader.cancel().catch(() => {});
       // 中途断流归因：客户端先断（res 已死）→ 取消上游属正常止损；res 仍活 → 上游停滞/读失败（watchdog 或传输错误）
       if (res.destroyed || res.writableEnded) clientGone = true;
-      if (transform) writeEvents(transform.stream.abort());
-      if (!res.destroyed) res.end();
+      if (!res.destroyed && !res.writableEnded) {
+        if (transform) writeEvents(transform.stream.abort());
+        else res.write('data: ' + JSON.stringify({ error: { type: 'api_error', message: 'upstream stream interrupted' } }) + '\n\n');
+        res.end();
+      }
       log.warn(reqId, 'stream aborted', { err: String(e).slice(0, 120) });
     } finally {
       clearTimeout(idle);
@@ -173,6 +196,11 @@ export async function forward({ reqId, payload, wantStream, res, transform }) {
   }
   clearTimeout(idleTimer);
   const body = aggregate(text);
+  if (!body.choices?.[0]?.finish_reason) {
+    recordUsage({ errors: 1 });
+    log.warn(reqId, 'upstream incomplete response', { ms: Date.now() - started });
+    return { handled: false, status: 502, body: { error: { message: 'upstream stream ended without finish_reason' } } };
+  }
   if (body.model) observeModel(body.model); // 嗅探上游真实模型名（非流式聚合结果里的 model）
   recordUsage({ prompt: body.usage.prompt_tokens || 0, completion: body.usage.completion_tokens || 0 });
   log.info(reqId, 'chat done', { ms: Date.now() - started, out: body.usage.completion_tokens });

src/server.mjs

@@ -3,6 +3,7 @@ import http from 'node:http';
 import { config } from './config.mjs';
 import { checkAuth } from './auth.mjs';
 import { forward } from './proxy.mjs';
+import { validateThinkingBudget } from './thinking.mjs';
 import { log, newReqId } from './log.mjs';
 import { snapshot as usageSnapshot } from './usage.mjs';
 import { list as listModels } from './models.mjs';
@@ -95,6 +96,8 @@ async function handleChatCompletions(req, res, reqId) {
   const parsed = await readJsonBody(req, res, reqId);
   if (parsed === null) return;
 
+  const budgetError = validateThinkingBudget(parsed, { mode: config.thinkingMode, minTokens: config.thinkingMinTokens });
+  if (budgetError) return send(res, 400, { error: { type: 'invalid_request_error', message: budgetError } });
   const wantStream = parsed.stream === true;
   const out = await forward({ reqId, payload: parsed, wantStream, res });
   if (out.handled) return; // 流式：proxy.mjs 已写完响应
@@ -159,13 +162,15 @@ const server = http.createServer(async (req, res) => {
 
       let openaiPayload;
       try {
-        openaiPayload = fromAnthropicRequest(anthropicBody);
+        openaiPayload = fromAnthropicRequest(anthropicBody, { allowThinking: config.allowThinking });
       } catch (e) {
         if (e instanceof ProtocolError) {
           return send(res, e.status, { type: 'error', error: { type: 'invalid_request_error', message: e.message } });
         }
         throw e;
       }
+      const budgetError = validateThinkingBudget(openaiPayload, { mode: config.thinkingMode, minTokens: config.thinkingMinTokens });
+      if (budgetError) return send(res, 400, anthropicError(400, { error: { message: budgetError } }));
       const clientModel = anthropicBody.model || '';
       const wantStream = openaiPayload.stream === true;
 

src/sse.mjs

@@ -53,7 +53,7 @@ export function aggregate(text) {
 
   return {
     id, object: 'chat.completion', created, model,
-    choices: [{ index: 0, message, finish_reason: finishReason || 'stop' }],
+    choices: [{ index: 0, message, finish_reason: finishReason }],
     usage: usage || { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0, usage_estimated: true },
   };
 }