/**
 * DEVIN_CONNECT → OpenAI ChatCompletion adapter.
 *
 * Bridges the structured stream from devin-connect.js (content / reasoning /
 * finish / usage events) into the two OpenAI-compatible response shapes the rest
 * of the proxy already emits:
 *
 *   - non-stream: a single `chat.completion` object
 *   - stream:     a sequence of `chat.completion.chunk` SSE frames + [DONE]
 *
 * The output shape is matched field-for-field to the Cascade path in
 * handlers/chat.js (role-priming chunk, reasoning_content before content, a
 * finish_reason:'stop' chunk, then a usage-only chunk) so a client can't tell
 * which backend served the request. Reasoning maps to `reasoning_content`,
 * which OpenAI-style clients hide by default — see handlers/chat.js:737.
 *
 * Pure translation: no network of its own, it only consumes streamChat().
 */

import { randomUUID } from 'crypto';
import { filterToolCallsByAllowlist } from './handlers/chat.js';
import { streamChat as realStreamChat, isRetryable, messageText, applyOutputCap } from './devin-connect.js';
import { ToolCallStreamParser, parseToolCallsFromText, isWeakEmulationModel } from './handlers/tool-emulation.js';
import { log } from './config.js';
import { safeLogValue } from './log-safety.js';
import { leakTraceEnabled, thinkMarkersIn, leakSample } from './leak-trace.js';
import { systemFingerprint } from './system-fingerprint.js';
import { applyStop, StopSequenceGate } from './stop-sequences.js';
import { normalizeToolCallArgs, recordArgRepair } from './handlers/cline-compat.js';
import { ThinkTextClassifier } from './response-classifier.js';
import { budgetExceededError } from './request-budget.js';
import { bumpConnect } from './devin-connect-metrics.js';
import { toClientToolCallId, restoreUpstreamToolIds } from './tool-ids.js';

// Gate for the leading think-tag reroute (Item 1, loop break). The classifier
// lives HERE, in the connect layer, so the DEVIN_CONNECT_ prefix is accurate:
// only the connect backend's streams are reclassified. Default OFF — the
// reroute is a behavior change and must be opted into per deployment.
function thinkTextRerouteEnabled() {
  return String(process.env.DEVIN_CONNECT_THINKTEXT_REROUTE || '') === '1';
}

// Apply the Cline compat tool-arg shim when active: normalize an arguments
// string @ai-sdk/openai-compatible would reject (empty / whitespace / non-JSON)
// to "{}" so a parameterless tool call isn't silently dropped (vercel/ai#6687).
// A no-op passthrough when inactive → byte-identical for every non-Cline client.
function compatArgs(raw, active) {
  // Inactive → preserve the exact legacy expression (`raw || '{}'`) so the
  // default path is byte-identical.
  if (!active) return raw || '{}';
  // Active → normalize the RAW value (not the pre-coalesced one) so an empty
  // string, whitespace, or malformed JSON is counted as a real repair.
  const fixed = normalizeToolCallArgs(raw);
  const legacy = raw || '{}';
  if (fixed !== legacy) recordArgRepair();
  return fixed;
}

// streamChat is injectable so the adapter can be unit-tested without touching
// the network — mirrors the __set…ForTest convention in windsurf-api.js.
let streamChatImpl = realStreamChat;
export function __setStreamChatForTest(fn) {
  streamChatImpl = typeof fn === 'function' ? fn : realStreamChat;
}

// ── retry-on-empty (fable capacity-jitter self-heal) ────────────────────────
// fable (and other capacity-jittered upstream models) occasionally return a
// COMPLETED turn that carries no answer at all: finish_reason 'stop',
// completion_tokens ≤ 2, and zero content / reasoning / tool_call deltas. This
// is PROBABILISTIC upstream capacity jitter — NOT a deterministic tool-count
// threshold (that theory was disproven; the code never trims tools) and NOT an
// outage (kimi's 502 path is a different fault). The correct heal is to simply
// re-issue the identical request a bounded number of times, since a fresh
// attempt usually lands a real answer.
//
// Because an empty reply yields ONLY a terminal finish event (no content), the
// wrapper below forwards every delta LIVE — zero buffering, zero added latency
// on the overwhelmingly common non-empty path — and merely holds the single
// finish event to decide, once the stream has drained, whether anything real
// was produced. It retries ONLY when the turn emitted literally nothing, so it
// is safe for both the streaming and non-stream callers (which each just
// iterate this wrapper instead of the raw primitive). It deliberately does NOT
// merge into isRetryable() (UPSTREAM_INTERNAL stays non-retryable by design) and
// never trims tools.
function retryOnEmptyEnabled(env = process.env) {
  // Default ON: an empty completion is always a degenerate result and the retry
  // is bounded + only fires when the turn yielded nothing. Only an explicit
  // off-switch disables it.
  const v = String(env.DEVIN_CONNECT_RETRY_ON_EMPTY ?? '').trim().toLowerCase();
  return v !== '0' && v !== 'off' && v !== 'false' && v !== 'no';
}
function retryOnEmptyMax(env = process.env) {
  const raw = Number(env.DEVIN_CONNECT_RETRY_ON_EMPTY_MAX);
  return Number.isFinite(raw) && raw >= 0 ? raw : 2;
}
function retryOnEmptyBaseMs(env = process.env) {
  const raw = Number(env.DEVIN_CONNECT_RETRY_ON_EMPTY_MS);
  return Number.isFinite(raw) && raw >= 0 ? raw : 350;
}

/**
 * Decide whether a drained stream was the pathological empty completion.
 * @param {object} finishEv  the terminal finish event (may be null if none seen)
 * @param {boolean} sawContent  true if any non-empty content/reasoning delta arrived
 */
function isEmptyCompletion(finishEv, sawContent) {
  // The AUTHORITATIVE signal is that the turn produced ZERO usable output:
  // no content delta, no reasoning delta, no tool call. `sawContent` already
  // captures "any non-empty content/reasoning arrived", so if it's true the
  // turn is not empty. This is what actually breaks OpenCode/agent loops.
  if (sawContent || !finishEv) return false;
  // A native/emulated tool call is a real answer even with no visible text.
  if (finishEv.toolCalls && finishEv.toolCalls.length) return false;
  // Only a clean 'stop' (or an absent finish reason — a clean drain with no
  // signal) counts. 'length' / 'tool_calls' / 'content_filter' are real
  // terminal states we must not paper over with a retry.
  if (finishEv.reason != null && finishEv.reason !== 'stop') return false;
  // NOTE: completion_tokens is deliberately NOT used as a gate. Live paid probes
  // (2026-07-08, fable-5-medium + 11–14 tools) showed genuine empty replies with
  // completion_tokens of 3/5/8/9 — an earlier `ct <= 2` gate silently vetoed the
  // retry on every one of them (15/15 empty, zero heals). The empty OUTPUT
  // (sawContent === false with a clean stop) is the real signature; ct only rides
  // the log line below for diagnostics.
  return true;
}

// ── #236 prose tool-intent detector (log-only) ─────────────────────────────
// The rescue gate below requires `sawReasoning && !sawText`. A reasoning-effort
// `-none` selector (glm-5-2-none and its siblings) can NEVER satisfy it: the
// upstream emits no reasoning events at all, so a stalled turn in that family
// arrives as TEXT or as nothing, never as reasoning. The reported shape is the
// same trap one channel over — the model narrates the tool call in visible text,
// finishes with `stop` and zero tool calls, and the client waits for a call that
// never comes.
//
// We deliberately do NOT retry on it. "The model answered in prose" and "the
// model stalled mid-intent" are indistinguishable without a capture of a real
// stall, and a false-positive nudge would rewrite a legitimate answer (#236
// thread, 2026-08-03: 60+ direct turns, zero stalls, so the trigger is not even
// established). What that thread is missing is the capture — so this detector
// produces one. It logs and counts; it never alters the wire.
//
// The criterion is narrow on purpose, and its last two conditions are what keep
// ordinary final answers out of the log:
//   - the request declared tools (a request without tools has nothing to stall on);
//   - finish is `stop`/null with ZERO tool calls (a real call, or a `length` /
//     `content_filter` terminal state, is not this shape);
//   - the turn does not END like a finished sentence. A completed answer ends in
//     `.`/`!`/`?` (or the CJK equivalents); a turn that stops mid-intent ends in
//     `:`, an ellipsis, or nothing at all. This is what separates
//     "Let me read the file:" (stall) from "I used read_file to inspect it."
//     (a normal summary) — the latter names a tool too, and would otherwise
//     dominate the signal;
//   - and the tail then names one of THIS request's tools, or carries an intent
//     phrase. Tool names are matched with explicit ASCII boundaries so `bash`
//     does not match inside `bashful`, and `read_file` does not match inside
//     `read_files`.
//
// Accepted false negatives: a stall whose text happens to end in a period is not
// detected. Keeping the counter clean matters more than covering every phrasing,
// because the counter is what an operator will alarm on.
const PROSE_INTENT_TAIL_MAX = 400;
// Sentence-final punctuation, optionally followed by closing quotes/brackets.
const SENTENCE_FINAL = /[.!?。！？]["'”’`)\]]*\s*$/;
// Future/present intent markers. `let me` is excluded when followed by "know" —
// "Let me know if you need anything else" is a closing, not an intent.
const PROSE_INTENT_VERB = /\b(?:let me(?!\s+know\b)|i(?:'ll| will| am going to|'m going to| need to| should)|next,? i|now,? i)\b|(?:让我|我来|我先|接下来我|我将|我需要|正在)/i;

function proseIntentDetectEnabled(env = process.env) {
  // Log-only, so it is on by default: a detector that has to be switched on
  // before it can capture the shape it exists to capture captures nothing.
  return String(env.DEVIN_CONNECT_PROSE_INTENT_DETECT ?? '1') !== '0';
}

/** Tool names declared by the request, in declaration order. */
function requestToolNames(tools) {
  if (!Array.isArray(tools)) return [];
  const names = [];
  for (const tool of tools) {
    const name = tool?.function?.name || tool?.name;
    if (typeof name === 'string' && name.trim()) names.push(name.trim());
  }
  return names;
}

/**
 * Judge a FINISHED turn for the #236 "narrated a tool call, emitted none" shape.
 * Returns a signal string (`tool_name:<name>` or `intent_verb:<phrase>`) or null.
 * Pure and synchronous — the caller only logs/counts the result.
 */
export function detectProseToolIntent({ text, toolNames = [], finishEv } = {}) {
  if (!finishEv) return null;
  if (finishEv.reason != null && finishEv.reason !== 'stop') return null;
  if (finishEv.toolCalls && finishEv.toolCalls.length) return null;
  const full = String(text || '');
  if (!full.trim()) return null;
  const tail = full.slice(-PROSE_INTENT_TAIL_MAX);
  if (SENTENCE_FINAL.test(tail)) return null;
  for (const name of toolNames) {
    if (!name || name.length < 3) continue;
    const escaped = name.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
    if (new RegExp(`(^|[^A-Za-z0-9_])${escaped}([^A-Za-z0-9_]|$)`, 'i').test(tail)) {
      return `tool_name:${name}`;
    }
  }
  const verb = tail.match(PROSE_INTENT_VERB);
  return verb ? `intent_verb:${verb[0].trim().toLowerCase()}` : null;
}

/**
 * Log/count the #236 signal for a turn that has ALREADY been delivered.
 *
 * MUST run at the adapter layer, after tool-call emulation: before the parser
 * strips `<tool_call>` markup, a normal emulated call looks like prose that
 * stops mid-intent, so every emulated call was counted as a stall (the exact
 * traffic the counter exists to measure). The caller passes the DELIVERED text
 * and the calls it actually emitted, so a delivered call can never be reported
 * as "narrated one, emitted none".
 */
function reportProseToolIntent({ params, text, finishEv, reasoning }) {
  if (!proseIntentDetectEnabled()) return;
  if (!Array.isArray(params?.tools) || params.tools.length === 0) return;
  const toolNames = requestToolNames(params.tools);
  const signal = detectProseToolIntent({ text, toolNames, finishEv });
  if (!signal) return;
  bumpConnect('prose_tool_intent');
  log.info(
    `DEVIN_CONNECT: prose tool intent without a call (#236) — `
    + `model=${safeLogValue(params?.model)}, finish=${finishEv?.reason ?? 'null'}, `
    + `signal=${signal}, reasoning=${Boolean(reasoning)}, tools=${toolNames.length}, `
    + `sample=${JSON.stringify(leakSample(text))}`,
  );
}

/**
 * Transparent wrapper around streamChatImpl that heals probabilistic empty
 * completions by re-issuing the identical request, and rescues thinking-only
 * traps (swe-1-7 declaring tool intent in reasoning without emitting the call)
 * by appending a corrective user nudge and dropping empty assistant turns.
 * Yields the same event stream as streamChat; on a non-empty turn it is a pass-through.
 */
// There used to be a `rescueThinkingOnly` option here. It was removed in v3.9.20 because it
// had become dead configuration: #241 moved the real gate INSIDE this function (`hasTools`,
// derived from the request), after which both call sites passed a literal `true` and nothing —
// production or test — ever passed `false`. A knob with one reachable value is worse than no
// knob: it reads as a variation point that a reviewer must reason about, and any mutation
// guard aimed at it guards code that cannot be reached (the exact shape of the v3.9.13 medium
// defect). If a caller ever genuinely needs to opt out, add the option back THEN, with a test
// that passes `false` — that test is what makes it a knob rather than decoration.

// Egress/ingress boundary for tool-call ids. The client received a normalized
// `call_…` id (see toClientToolCallId); the upstream issued its own dialect
// (`get_weather:0#<uuid>` on swe-2). Restore the original id on a COPY of the
// request right before the wire encode, so native history replay (#6
// ChatToolCall / #7 tool_call_id) is byte-identical to what the upstream
// issued. Everything upstream of this point — session continuity, cache keys,
// the response body — sees the client-visible id.
function withUpstreamToolIds(params) {
  const messages = restoreUpstreamToolIds(params?.messages);
  return messages === params?.messages ? params : { ...params, messages };
}

async function* streamChatWithEmptyRetry(params, { env = process.env } = {}, opts = {}) {
  // Route-owned request budget. Each internal attempt is one upstream turn; the
  // cap is per REQUEST, shared with the handler's account failover, so the
  // empty/rescue retries can no longer multiply into the measured 45/90-turn
  // products. Backoffs are clamped to the remaining deadline.
  const budget = opts?.budget || null;
  // Weak models (fable) return DETERMINISTIC empties on complex multi-turn / large
  // system — paid E2E (2026-07-08, 27/27) proved retry never heals them, it only
  // triples the upstream load and burns the account into a 3h rate limit. So for
  // weak models we do NOT retry on empty (short-circuit). Non-weak models keep the
  // bounded retry, where an empty is more likely genuine capacity jitter.
  const weak = isWeakEmulationModel(params?.model || '');
  const max = (retryOnEmptyEnabled(env) && !weak) ? retryOnEmptyMax(env) : 0;
  // Same sanitisation as every sibling knob in this file. Without it `Infinity` and
  // `1e9` both defeated the `rescueAttempt < rescueMax` bound — and the loop uses
  // `continue`, so one client request became an unbounded sequence of upstream calls
  // that burns the account into a rate limit. A non-numeric value produced NaN,
  // which silently turned the whole feature OFF with no log line.
  //
  // The rescue also honours the empty-retry master off switch: an operator who set
  // DEVIN_CONNECT_RETRY_ON_EMPTY=0 to stop speculative re-issues on a rate-limited
  // account meant that, and it would be a surprise for a new speculative re-issue to
  // ignore it. RESCUE_MAX=0 remains the way to disable just this one.
  // Number.isFinite alone is not enough: 1e9 passes it and still hangs the request
  // for hours of upstream calls (measured — the probe had to be killed). A rescue is
  // a speculative re-issue against a live account, so its ceiling is hard.
  const RESCUE_MAX_CEILING = 5;
  const rescueRaw = Number(env.DEVIN_CONNECT_RESCUE_MAX);
  const rescueConfigured = Number.isFinite(rescueRaw) && rescueRaw >= 0
    ? Math.min(rescueRaw, RESCUE_MAX_CEILING)
    : 2;
  // Weak models are excluded for the same measured reason the empty-retry is: paid
  // E2E proved retrying the fable family never heals it and only triples upstream
  // load into a 3h rate limit. Whether fable produces reasoning-only finishes is
  // unproven either way, so this errs toward the documented account-protection.
  const rescueMax = (retryOnEmptyEnabled(env) && !weak) ? rescueConfigured : 0;
  // Cap for the reasoning digest: the nudge quotes only the END of the reasoning,
  // sliced once at nudge time (the consumers already hold the full reasoning text,
  // so per-chunk trimming would only churn the GC). 0 disables the digest; NaN
  // (a genuinely non-numeric value like `abc`) falls back to the default.
  //
  // EMPTY STRING IS NOT NaN: Number('') === 0, so `…MAX_CHARS=` with nothing after
  // the `=` disables the digest rather than falling back. That is the SAME convention
  // the three sibling knobs in this file already have (RETRY_ON_EMPTY_MAX → 0 retries,
  // RETRY_ON_EMPTY_MS → 0 backoff, RESCUE_MAX → rescue off), all measured, so it is
  // left consistent rather than special-cased here — but it is written down because
  // the previous version of this comment claimed the opposite.
  //
  // The ceiling exists for the reason RESCUE_MAX_CEILING does (see rescueConfigured above):
  // Number.isFinite lets `1e9` through. Measured — reasoning of 50000 chars with `1e9` set
  // produced a 50089-byte nudge, i.e. no cap at all, on EVERY rescue of that request.
  // Unlike RESCUE_MAX this cannot hang the request (it is one slice, not an unbounded loop),
  // so the ceiling is not an account-protection bound; it bounds how large a single upstream
  // body one env typo can produce. 32000 is a JUDGEMENT (16x the default, far above any
  // plausible tuning), NOT a measured upstream limit — the real upstream body limit has
  // never been probed.
  //
  // Math.floor is load-bearing, not tidiness. The single consumer is `slice(-n)`, and for
  // 0 < n < 1 slice truncates the argument toward zero: `slice(-0.5)` === `slice(0)` ===
  // THE WHOLE STRING. So a fractional cap under 1 inverted into "no cap at all" — measured,
  // MAX_CHARS=0.5 shipped all 50000 chars, the exact failure the ceiling was added to stop.
  // Clamping alone did not cover it because Math.min(0.5, 32000) is correctly 0.5.
  const DIGEST_MAX_CEILING = 32000;
  const digestRaw = Number(env.DEVIN_CONNECT_RESCUE_REASONING_MAX_CHARS);
  const reasoningDigestMaxChars = Number.isFinite(digestRaw) && digestRaw >= 0
    ? Math.floor(Math.min(digestRaw, DIGEST_MAX_CEILING))
    : 2000;
  let attemptParams = params;
  // Two speculative arms, two budgets (#240). They used to share the loop counter: the
  // rescue arm has always had its own `rescueAttempt`/`rescueMax` pair, but reaching the
  // next iteration via `continue` also advanced the `for (…; ; attempt++)` counter, and
  // `attempt` was what the EMPTY arm's budget was measured against. So a rescue chain
  // silently spent the empty arm's budget — one-way, since the empty arm never touched
  // `rescueAttempt`. Measured at the defaults (max 2, rescue 2): two rescues followed by
  // an empty completion delivered the empty answer to the client with ZERO retries, while
  // the same empty with no preceding rescue healed on the first retry.
  //
  // That the rescue arm carries its own counter is the evidence this was never a "one
  // shared pot" policy — a shared pot would not need the second counter. It was an
  // artifact of counting in the for-header.
  //
  // The TOTAL ceiling is deliberately unchanged: 1 + max + rescueMax upstream calls per
  // client request. That ceiling is the account protection (the fable paid-E2E lesson —
  // retries that never heal only triple upstream load into a rate limit), and it was
  // already reachable before this change in the empty-first order. Splitting the counters
  // makes the empty budget independent of ORDER; it does not raise what may be spent.
  // #236's detector is NOT here: it runs in the adapters, after emulation, on
  // the text and calls the client actually received. Judging the raw upstream
  // tail here counted every normal emulated `<tool_call>` turn as a stall
  // (the markup is stripped only downstream).
  const hasTools = Boolean(params?.tools?.length);
  let rescueAttempt = 0;
  let emptyAttempt = 0;
  for (;;) {
    let sawContent = false;
    let sawText = false;
    let sawReasoning = false;
    let sawReasoningText = '';
    let finishEv = null;
    // P-04: reasoning is provisional until this attempt is committed. A rescue
    // REPLACES the attempt and a delta already sent cannot be retracted, so a
    // reasoning-only attempt that is later abandoned used to leave its whole
    // trace on the wire while the returned/cached/continuity value kept only
    // the final attempt. Hold reasoning while a rescue is still possible and
    // release it only when the attempt can no longer be replaced: the first
    // text delta (sawText closes the rescue gate) or the terminal finish. The
    // rescue branch below drops the buffer together with the attempt.
    const rescuePossible = hasTools && rescueAttempt < rescueMax;
    let pendingReasoning = [];
    // The thinking signature (opt-in decode, DEVIN_CONNECT_SIGNATURE_TAG) seals
    // this attempt's reasoning, so it is held and released with it: committed
    // together, or dropped together when a rescue replaces the attempt.
    let pendingSignature = '';
    // Item 1 (loop break): leading think-tagged CONTENT is reclassified into the
    // reasoning channel HERE, at the stream-event level, BEFORE the rescue
    // decision below. A whole turn of ` thinking…` therefore stays a
    // reasoning-only stream.
    // Honest rescue interplay (corrected after review): such a turn does NOT
    // fire the #238 rescue — the :284 nudge path requires tools, and
    // isEmptyCompletion reads sawContent, which the thinking branch below sets
    // at :238. The real outcome: reasoning delivered as thinking blocks, text
    // empty, no retry. That is accepted: the reasoning is not lost (the client
    // receives thinking blocks, not the #238 whole-turn vanish), and sawContent
    // genuinely means "upstream said something", which is true here.
    // Deliberately NOT teaching isEmptyCompletion to read sawText — that would
    // redraw the #238/#241 rescue boundary for a cosmetic gain.
    // (Known limitation: only the ` thinking`…` dialect is recognized; Kimi K2's
    // family uses `◁think▷…◁/think▷` — extending the classifier to that dialect
    // is left for a future PR.)
    const thinkClassifier = thinkTextRerouteEnabled() ? new ThinkTextClassifier() : null;
    // One turn per upstream call, consumed immediately before it. The cap is
    // hard: when it is reached the generator throws a 504-class error instead
    // of starting another turn.
    if (budget && !budget.takeTurn()) {
      throw budgetExceededError(
        budget.expired()
          ? `request budget exhausted before DEVIN_CONNECT attempt (${budget.describe()})`
          : `request upstream-turn cap reached before DEVIN_CONNECT attempt (${budget.describe()})`,
      );
    }
    const upstreamParams = withUpstreamToolIds(attemptParams);
    const callParams = budget
      ? {
        ...upstreamParams,
        // devin-connect.js treats deadlineMs as the total wall clock for THIS
        // stream. Give it what the request has left, never more.
        deadlineMs: Math.min(
          Number.isFinite(upstreamParams?.deadlineMs) && upstreamParams.deadlineMs > 0
            ? upstreamParams.deadlineMs
            : Infinity,
          budget.remainingMs(),
        ),
      }
      : upstreamParams;
    for await (const ev of streamChatImpl(callParams)) {
      if ((ev.type === 'content' || ev.type === 'reasoning') && leakTraceEnabled(env)) {
        log.info('LEAK_TRACE stream-event', {
          channel: ev.type,
          think: thinkMarkersIn(ev.text),
          sample: leakSample(ev.text),
          len: ev.text ? ev.text.length : 0,
          reqId: opts?.reqId ?? null,
          account: opts?.account ?? null,
        });
      }
      if (ev.type === 'content' && thinkClassifier) {
        const routed = thinkClassifier.feed(ev.text);
        if (routed.thinking) {
          sawContent = true;
          sawReasoning = true;
          sawReasoningText += routed.thinking;
          if (rescuePossible && !sawText) pendingReasoning.push(routed.thinking);
          else yield { type: 'reasoning', text: routed.thinking };
        }
        if (routed.text) {
          // First text delta: the rescue gate (`!sawText`) is now closed, so
          // this attempt is committed — release its provisional reasoning
          // before the text it preceded.
          for (const text of pendingReasoning) yield { type: 'reasoning', text };
          pendingReasoning = [];
          if (pendingSignature) { yield { type: 'signature', reasoning_signature: pendingSignature }; pendingSignature = ''; }
          sawContent = true;
          sawText = true;
          yield { type: 'content', text: routed.text };
        }
      } else if (ev.type === 'content' || ev.type === 'reasoning') {
        if (ev.text) {
          sawContent = true;
          if (ev.type === 'content') {
            // First text delta commits the attempt — see the classifier branch.
            for (const text of pendingReasoning) yield { type: 'reasoning', text };
            pendingReasoning = [];
            if (pendingSignature) { yield { type: 'signature', reasoning_signature: pendingSignature }; pendingSignature = ''; }
            sawText = true;
          } else {
            sawReasoning = true;
            sawReasoningText += ev.text;
            if (rescuePossible && !sawText) {
              pendingReasoning.push(ev.text);
              continue;
            }
          }
        }
        yield ev;
      } else if (ev.type === 'finish') {
        finishEv = ev; // hold: decide retry after the stream drains
      } else if (ev.type === 'signature' && rescuePossible && !sawText) {
        pendingSignature += ev.reasoning_signature || '';
      } else {
        yield ev;
      }
    }
    // Flush anything the classifier still holds BEFORE the rescue decision: an
    // unterminated/undecided tail is delivered as text (visible beats dropped),
    // and it must count as real text for the rescue gate above.
    if (thinkClassifier) {
      const rest = thinkClassifier.flush();
      if (rest) {
        // The flushed tail is text, so the attempt is committed for the same
        // reason as a normal content delta.
        for (const text of pendingReasoning) yield { type: 'reasoning', text };
        pendingReasoning = [];
        if (pendingSignature) { yield { type: 'signature', reasoning_signature: pendingSignature }; pendingSignature = ''; }
        sawContent = true;
        sawText = true;
        yield { type: 'content', text: rest };
      }
    }
    // swe-1-7 (Kimi K2 fine-tune) intermittently spends the whole turn in reasoning
    // declaring tool intent without emitting the call; a corrective nudge measurably
    // (24/24 live probe) forces emission; empty assistant turns poison upstream into
    // repeating empty turns.
    // NOTE: Rescue triggers only when tools are present in the request (params.tools?.length > 0).
    // Plain chat requests without tools legitimately end in reasoning/text, so rescue is skipped.
    const isStopOrNull = finishEv && (finishEv.reason == null || finishEv.reason === 'stop');
    const hasNoToolCalls = !finishEv?.toolCalls || finishEv.toolCalls.length === 0;
    if (hasTools && rescueAttempt < rescueMax && isStopOrNull && hasNoToolCalls && sawReasoning && !sawText) {
      rescueAttempt++;
      log.warn(`DEVIN_CONNECT: thinking-only completion (reasoning-only, finish=${finishEv?.reason ?? 'null'}) — rescue retry ${rescueAttempt}/${rescueMax} (nudge appended)`);
      const backoff = retryOnEmptyBaseMs(env) * rescueAttempt;
      if (backoff && !(await budgetedSleep(backoff, budget))) {
        throw budgetExceededError(`request budget exhausted during DEVIN_CONNECT rescue backoff (${budget.describe()})`);
      }
      // Rebuild from the ORIGINAL request every rescue: `params` is never mutated (each
      // rescue constructs a fresh attemptParams below), so it always holds the client's
      // messages. Building from attemptParams instead accumulated — previous rescues'
      // nudges are role:'user' so the filter keeps them, and with a capped digest each
      // stale nudge is ~2.1KB at the 2000-char default: five rescues stacked ~10.4KB
      // of old nudges into the upstream call, each quoting reasoning the model already
      // moved past.
      // One rescue = exactly one fresh nudge.
      const origMsgs = params?.messages || [];
      const filteredMsgs = origMsgs.filter((msg) => {
        if (msg.role !== 'assistant') return true;
        if (Array.isArray(msg.tool_calls) && msg.tool_calls.length) return true;
        if (msg.reasoning_signature && String(msg.reasoning_content || '').length > 0) return true;
        return messageText(msg.content).trim() !== '';
      });
      let nudgeText = 'Stop reasoning. Emit the tool call markup now.';
      const digest = reasoningDigestMaxChars > 0 ? sawReasoningText.slice(-reasoningDigestMaxChars).trim() : '';
      if (digest) {
        nudgeText = `Your previous reasoning ended with: """${digest}"""\nStop reasoning. Emit the tool call markup now.`;
      }
      const rescued = [
        ...filteredMsgs,
        { role: 'user', content: nudgeText },
      ];
      attemptParams = { ...attemptParams, messages: rescued };
      // The abandoned attempt's provisional reasoning dies with this loop
      // iteration (`pendingReasoning` is declared inside it): it was never sent
      // (the buffer is only released at commit), so the client, the returned
      // result, the cache and session continuity all stay on the replacement
      // attempt. The attempt_reset below still tells the consumer to clear the
      // other accumulators (content/tool calls/billing) for the same reason.
      // Tell the consumer to discard what it accumulated: the next attempt REPLACES
      // this one, it does not continue it. Without this every rescue attempt's
      // reasoning was concatenated into a single answer — measured
      // "PASS1. PASS2. PASS3. " for three attempts, i.e. the user was shown all
      // three tries glued together. The outer retry loops in toChatCompletion /
      // streamChatCompletion reset their accumulators per THEIR OWN attempt, but the
      // rescue loop lives in here, so they never saw an attempt boundary.
      yield { type: 'attempt_reset' };
      continue;
    }

    if (emptyAttempt < max && isEmptyCompletion(finishEv, sawContent)) {
      emptyAttempt++;
      log.warn(`DEVIN_CONNECT: empty completion (finish=${finishEv.reason ?? 'null'}, completion_tokens=${finishEv.usage?.completion_tokens ?? 'n/a'}) — retry ${emptyAttempt}/${max}`);
      // Backoff still grows with the number of EMPTY retries (base×1, base×2, …), which is
      // what it did before when `attempt` happened to equal that count. Keying it to the
      // shared counter instead would make an unrelated rescue lengthen the next empty wait.
      const backoff = retryOnEmptyBaseMs(env) * emptyAttempt;
      if (backoff && !(await budgetedSleep(backoff, budget))) {
        throw budgetExceededError(`request budget exhausted during DEVIN_CONNECT empty-retry backoff (${budget.describe()})`);
      }
      continue;
    }
    if (weak && finishEv && isEmptyCompletion(finishEv, sawContent)) {
      log.warn(`DEVIN_CONNECT: weak model ${params.model} empty completion (finish=${finishEv.reason ?? 'null'}) — NOT retrying (deterministic, retry would amplify rate limit)`);
    }
    // No rescue: this is the committed attempt. Release any reasoning still held
    // while a rescue was possible, before its terminal finish event.
    for (const text of pendingReasoning) yield { type: 'reasoning', text };
    pendingReasoning = [];
    if (pendingSignature) { yield { type: 'signature', reasoning_signature: pendingSignature }; pendingSignature = ''; }
    if (finishEv) yield finishEv;
    return;
  }
}

// Sleep that can never outlive the request budget. Returns false when the
// budget has no time left — including when it cannot cover the full backoff —
// so the caller stops instead of starting an attempt that the route deadline
// would kill mid-flight.
async function budgetedSleep(ms, budget) {
  if (!ms) return true;
  if (!budget) {
    await new Promise((r) => setTimeout(r, ms));
    return true;
  }
  const left = budget.remainingMs();
  // A backoff the budget cannot fully honour means there is no room left for
  // the retry it was pacing. Sleeping the clamped remainder and then re-issuing
  // starts the attempt with a sliver of time, and the deadline kills it
  // mid-flight — the v3.9.57 CI flake (run 39861008441). The final
  // `remainingMs() > 0` check cannot catch that: it is the same condition
  // budgetedSleep already returned, and canStartAttempt() adds only the turn
  // cap. Stop here; each caller surfaces its own error.
  if (left < ms) return false;
  await new Promise((r) => setTimeout(r, ms));
  return budget.remainingMs() > 0;
}

const OBJECT_COMPLETION = 'chat.completion';
const OBJECT_CHUNK = 'chat.completion.chunk';

function newId() {
  return `chatcmpl-${randomUUID().replace(/-/g, '')}`;
}

function nowSeconds() {
  return Math.floor(Date.now() / 1000);
}

/**
 * Collect a full DEVIN_CONNECT completion and shape it as a non-streaming
 * `chat.completion` object.
 *
 * @param {object} params  forwarded to streamChat (messages, model, token, …)
 * @param {object} [opts]
 * @param {string} [opts.id]            response id (default chatcmpl-…)
 * @param {number} [opts.created]       unix seconds (default now)
 * @param {string} [opts.displayModel]  model name echoed back to the client
 * @param {boolean} [opts.emulateTools] when true, parse <tool_call> markup out of
 *                                      the buffered answer and surface OpenAI
 *                                      tool_calls (text-emulation, swe-1.6 etc).
 * @returns {Promise<{status:number, body:object, stop:string|null}>}
 */
export async function toChatCompletion(params, opts = {}) {
  const { id = newId(), created = nowSeconds(), displayModel, maxRetries = 1, retryBaseMs = 400, emulateTools = false, stop = null, clineCompat = false, stopCarrier = false, budget = null } = opts;
  const model = displayModel || params.model;

  // Non-stream path buffers the whole answer, so a transient failure (network
  // blip, 5xx, rate limit) can be retried cleanly — a discarded partial buffer
  // never duplicates tokens. Terminal errors (MODEL_BLOCKED / UNAUTHORIZED)
  // are not retryable and throw straight through for the handler to map.
  let content = '';
  let reasoning = '';
  let reasoningSignature = '';
  let signedSegment = false;
  let multipleSignedSegments = false;
  let finishReason = 'stop';
  // The upstream ended the stream without ever naming a stop reason (devin-connect
  // raises `reason:'error'` + `unterminated:true` for a delivered payload with no
  // #5). Remembered separately because the tool-call arms below overwrite
  // finishReason with 'tool_calls' whenever a call exists — which would report a
  // turn whose arguments were never validated as a clean, executable finish.
  let finishUnterminated = false;
  let usage = null;
  // Per-request billing detail (credit / ACU cost) when the operator has calibrated
  // DEVIN_CONNECT_BILLING_TAGS. devin-connect.js has always yielded this on the
  // finish event; it stopped here because the adapter never forwarded it, so the
  // dashboard's lifetime-credit column could only ever render 0.
  let billing = null;
  // Native tool calls (DEVIN_CONNECT_TOOL_CALL_TAGS calibrated) ride the terminal
  // finish event as ev.toolCalls (devin-connect.js:927). Null/empty on free tier
  // and un-calibrated deployments, where prompt emulation owns tool calls.
  let nativeToolCalls = [];
  for (let attempt = 0; ; attempt++) {
    try {
      content = ''; reasoning = ''; reasoningSignature = ''; signedSegment = false; multipleSignedSegments = false; finishReason = 'stop'; finishUnterminated = false; usage = null; nativeToolCalls = [];
      for await (const ev of streamChatWithEmptyRetry(params, undefined, opts)) {
        // A rescue attempt REPLACES the previous one; drop what it produced or the
        // client is handed every attempt concatenated.
        if (ev.type === 'attempt_reset') { content = ''; reasoning = ''; reasoningSignature = ''; signedSegment = false; multipleSignedSegments = false; nativeToolCalls = []; finishUnterminated = false; billing = null; continue; }
        if (ev.type === 'content') content += ev.text;
        else if (ev.type === 'reasoning') {
          if (signedSegment) multipleSignedSegments = true;
          reasoning += ev.text;
        }
        // Thinking signature (opt-in decode, DEVIN_CONNECT_SIGNATURE_TAG). The
        // per-frame events are increments; the finish event's total is not used
        // because a rescue attempt_reset must discard the abandoned increments.
        else if (ev.type === 'signature') {
          reasoningSignature += ev.reasoning_signature || '';
          signedSegment = true;
        }
        else if (ev.type === 'finish') {
          if (ev.reason) finishReason = ev.reason;
          if (ev.unterminated === true) finishUnterminated = true;
          if (ev.usage) usage = ev.usage;
          if (ev.billing) billing = ev.billing;
          if (ev.toolCalls && ev.toolCalls.length) nativeToolCalls = ev.toolCalls;
        }
      }
      break;
    } catch (err) {
      if (!isRetryable(err) || attempt >= maxRetries) throw err;
      // The inner generator already consumes a turn per upstream call; when
      // the route budget has no turn/time left, re-issuing here would only be
      // refused. Surface the real upstream error instead of a synthetic one.
      if (budget && !budget.canStartAttempt()) throw err;
      const backoff = retryBaseMs * 2 ** attempt;
      log.warn(`DEVIN_CONNECT: retryable error (${err.code || err.message}); retry ${attempt + 1}/${maxRetries} in ${backoff}ms`);
      if (!(await budgetedSleep(backoff, budget))) throw err;
      // Re-read the gate after the wait: the pre-sleep check is a snapshot, and
      // the retry must be admitted by the budget as it stands now. (The deadline
      // sliver itself is budgetedSleep()'s job — it refuses a backoff the budget
      // cannot cover.) The real upstream error is what surfaces.
      if (budget && !budget.canStartAttempt()) throw err;
    }
  }

  // Tool calls come from one of two sources, never both. Native decode wins:
  // when DEVIN_CONNECT_TOOL_CALL_TAGS is calibrated, streamChat surfaces real
  // ChatToolCall structs ({id, name, arguments}) on the finish event, so the
  // text never carries <tool_call> markup to parse. Otherwise (free tier /
  // un-calibrated) the connect models have no native function-calling slot, so
  // tool defs were injected into the prompt (normalizeMessagesForCascade) and
  // the model answers with <tool_call>…</tool_call> markup we pull back out,
  // mirroring the Cascade non-stream path (handlers/chat.js buildToolCalls).
  //
  // SEED-C3c: the emulated parse runs BEFORE the stop gate, never after it. The
  // stop sequence fences the model's PROSE; a complete, declared, allowlisted
  // tool call carries opaque structured arguments. Gating the raw text first cut
  // inside that JSON and handed non-stream clients a truncated `<tool_call>`
  // fragment with no tool at all, while the stream path (which parses before its
  // prose gate, see sendContent below) returned the real call — the same request
  // answered two different ways depending on `stream`. Native structured calls
  // are opaque for the same reason.
  //
  // This does NOT decide mixed prose/tool ordering (an END that appears before a
  // later call, split-chunk order, salvage parity): only the accepted, declared,
  // allowlisted structured call is exempt from the cut.
  let stopHit = false;
  let matchedStop = null;
  let toolCalls = [];
  // The upstream's OWN terminal verdict (enum 13 → 'error'), captured before the
  // tool-call arms below, which set 'tool_calls' unconditionally whenever a call
  // was decoded. An error turn that happened to emit a decodable call is still an
  // error — the call's arguments are as unverified as the rest of the turn.
  const terminalError = finishReason === 'error';
  if (nativeToolCalls.length) {
    // arguments is the raw JSON string off the wire (decodeToolCalls); map it to
    // the same shape parseToolCallsFromText produces so the message builder below
    // is source-agnostic.
    toolCalls = nativeToolCalls.map((tc) => ({
      id: tc.id, name: tc.name, argumentsJson: tc.arguments,
    }));
    finishReason = 'tool_calls';
  } else if (emulateTools) {
    const parsed = parseToolCallsFromText(content, {
      modelKey: params.model, provider: null, route: 'devin_connect',
    });
    if (parsed.toolCalls.length) {
      content = parsed.text;
      toolCalls = filterToolCallsByAllowlist(parsed.toolCalls, params.tools || []);   // M2 ToolGuard parity
    }
  }
  // proto-openai-03: enforce the client's `stop` locally (the Devin wire has no
  // native stop field). Truncate at the earliest stop-sequence hit and report
  // finish_reason:'stop'. The cut applies to the prose that is left after the
  // structured-call parse above; tool arguments are never matched against it.
  {
    const stopped = applyStop(content, stop);
    if (stopped.hit) { content = stopped.text; matchedStop = stopped.stop; stopHit = true; }
  }
  if (stopHit && !toolCalls.length) finishReason = 'stop';

  // Fallback promotion: promote reasoning to content when no tool calls and no content exist
  // so plain prompts never return an empty visible answer to clients.
  //
  // MOVED, not copied. Emitting the same text as both `content` and
  // `reasoning_content` made every client that renders reasoning show the answer
  // twice — and on the Anthropic route it produced a `thinking` block and a `text`
  // block with byte-identical content, which is exactly what the reporter's own
  // client (kimi CLI) displays. Nothing is lost by dropping the duplicate: the
  // text is still delivered, just once, in the field the client will actually
  // render as the answer.
  // The promoted text must go through the SAME tool-call extraction the content
  // path just ran. The rescue nudge literally asks the model to "Emit the tool call
  // markup now", so a model that complies but keeps writing into the reasoning
  // channel produces exactly this shape — and promoting it unparsed delivered raw
  // `<tool_call>` XML to the client as the visible answer with finish_reason='stop'.
  // The agent loop does not advance and the user sees markup, which is worse than
  // the empty turn the promotion exists to prevent.
  let promotedReasoning = !toolCalls.length && !content && !!reasoning && !stopHit;
  if (promotedReasoning) {
    if (emulateTools && !stopHit) {
      const promotedParse = parseToolCallsFromText(reasoning, {
        modelKey: params.model, provider: null, route: 'devin_connect',
      });
      if (promotedParse.toolCalls.length) {
        // M2 ToolGuard parity: only keep calls the client declared
        const allowedCalls = filterToolCallsByAllowlist(promotedParse.toolCalls, params.tools || []);
        // It was a tool call all along. Deliver it as one; keep any surrounding prose
        // as content, and leave reasoning_content in place since nothing was moved
        // out of it into the answer.
        toolCalls = allowedCalls;
        content = promotedParse.text;
        finishReason = 'tool_calls';
        promotedReasoning = false;
      } else {
        content = reasoning;
      }
    } else {
      content = reasoning;
    }
  }

  // Promotion creates visible prose after the first gate. Gate it at this final boundary.
  if (promotedReasoning) {
    const stopped = applyStop(content, stop);
    content = stopped.text;
    if (stopped.hit) { matchedStop = stopped.stop; stopHit = true; }
    if (stopHit && !toolCalls.length) finishReason = 'stop';
  }

  // OpenAI convention: content is a string (may be empty), never undefined.
  const message = { role: 'assistant', content: content || '' };
  if (reasoning && !promotedReasoning) message.reasoning_content = reasoning;
  // The signature seals exactly this reasoning text; promoted reasoning is no
  // longer a thinking block, so a signature beside it would seal nothing.
  if (reasoning && !promotedReasoning && reasoningSignature && !multipleSignedSegments) message.reasoning_signature = reasoningSignature;
  if (toolCalls.length) {
    message.tool_calls = toolCalls.map((tc, i) => ({
      id: toClientToolCallId(tc.id, `call_${i}_${Date.now().toString(36)}`),
      type: 'function',
      function: { name: tc.name || 'unknown', arguments: compatArgs(tc.argumentsJson || tc.arguments, clineCompat) },
    }));
    // content is null when the turn is a tool call (the inline text is usually
    // a hallucinated preview the caller shouldn't show).
    message.content = null;
    finishReason = 'tool_calls';
  }

  // Budget honesty (live finding: gemini-3-6-flash-medium). Thinking is counted
  // in completion_tokens but the upstream does not bound it by max_tokens, so a
  // turn can overrun the caller's cap and still claim a clean 'stop'. Report
  // 'length' when the upstream's own count reached or passed the requested cap.
  // NOT a local truncation: every delta has already been delivered above, and
  // the exact count only exists on the terminal frame — see applyOutputCap.
  //
  // A turn cut by the CLIENT's stop sequence keeps 'stop': the caller asked for
  // that terminal cause and the Messages route carries it as `stop_sequence`
  // (the carrier below is gated on 'stop'). Tool calls / content_filter are
  // stronger states and applyOutputCap leaves them alone.
  //
  // A turn the upstream ended in ERROR (enum 13) stays an error even when a
  // decodable tool call arrived with it — the tool-call arms above would otherwise
  // launder a failed turn into a clean, executable `tool_calls` finish. Same for a
  // turn that never named a stop reason at all. Placed after EVERY tool-call arm so
  // a future one cannot slip past it, and before applyOutputCap (which only
  // upgrades 'stop').
  if (terminalError) finishReason = 'error';
  else if (finishUnterminated && finishReason === 'tool_calls') finishReason = 'error';
  if (!stopHit) finishReason = applyOutputCap(finishReason, usage, params.completion?.maxTokens);

  // #236 log-only detector — AFTER emulation and the stop gate, on the text and
  // calls the client actually receives. `stopHit` means a local stop sequence
  // cut the answer, which is not a model stall.
  if (!stopHit) {
    reportProseToolIntent({
      params,
      text: content,
      finishEv: { reason: finishReason, toolCalls },
      reasoning,
    });
  }

  const choice = { index: 0, message, finish_reason: finishReason };
  // SEED-C3a carrier. The internal Anthropic Messages route needs to know WHICH
  // stop sequence truncated this answer (it echoes it as `stop_sequence`), and
  // the matched bytes are gone from `content`, so the cause has to ride along.
  // Serialisable plain field, not a Symbol: the streaming twin travels through
  // JSON.stringify into the messages translator's SSE parser. Gated on the
  // explicit route opt-in (chat.js sets `stopCarrier` only for
  // trusted Messages context) so a direct OpenAI client sees the public shape
  // byte-for-byte. Only a genuine local stop carries it — a tool/length/refusal
  // finish cannot be upgraded by a stray value.
  if (stopCarrier && matchedStop && finishReason === 'stop') choice._windsurf_stop_sequence = matchedStop;

  const body = {
    id,
    object: OBJECT_COMPLETION,
    created,
    model,
    system_fingerprint: systemFingerprint(model),
    choices: [choice],
  };
  if (usage) body.usage = usage;
  // Private field: per-request credit/ACU cost, absent unless the billing tags are
  // calibrated. Under the _windsurf prefix so it can't be mistaken for an OpenAI
  // field by a downstream relay that metering reads.
  if (billing) body._windsurf_billing = billing;
  return { status: 200, body, stop: matchedStop };
}

/**
 * Stream a DEVIN_CONNECT completion as OpenAI `chat.completion.chunk` SSE
 * frames. `send` is the SSE writer used in handlers/chat.js — a function taking
 * a JS object that it JSON-encodes onto the `data:` line. This helper does NOT
 * write `data: [DONE]` or close the response; the caller owns the socket
 * lifecycle (heartbeat, unregister, res.end) exactly as the Cascade path does.
 *
 * Emission order mirrors the Cascade stream:
 *   1. role-priming chunk (delta {role, content:''}) — DEFERRED until the first
 *      real delta (or the finish tail) so a pre-open transient/dead-token error
 *      leaves the caller's first-connect recovery armed (see `prime` below)
 *   2. reasoning_content deltas — provisional while a rescue could still replace
 *      the attempt; released only when the attempt is committed (first text
 *      delta or terminal finish), so an abandoned attempt never reaches the wire
 *   3. content deltas as they arrive
 *   4. finish chunk (delta {}, finish_reason)
 *   5. usage-only chunk (choices [], usage) when usage is known
 *
 * @returns {Promise<{content:string, reasoning:string, finish_reason:string, usage:object|null}>}
 *          the assembled result, so callers can cache it after streaming.
 */
export async function streamChatCompletion(params, send, opts = {}) {
  const { id = newId(), created = nowSeconds(), displayModel, emulateTools = false, includeUsage = false, stop = null, clineCompat = false, stopCarrier = false, maxRetries = 1, retryBaseMs = 400, budget = null } = opts;
  const model = displayModel || params.model;
  const base = { id, object: OBJECT_CHUNK, created, model, system_fingerprint: systemFingerprint(model) };

  // 1. Role-priming chunk. This USED to fire eagerly here, before streamChat
  //    opened the upstream — but the very first send() flips `emitted=true` in
  //    the caller (handlers/chat.js), which disarms the handler's !emitted-gated
  //    account recovery branches (re-login / failover). A dead token — which the
  //    non-stream path re-logs-in / fails over — then surfaced as a hard client
  //    error on the stream path. So we DEFER the prime behind `primed` until the
  //    first REAL delta (content / reasoning / tool_call) actually arrives, i.e.
  //    until the upstream has demonstrably opened. `primed` also gates this
  //    adapter's own one-shot transport retry (nothing emitted ⇒ safe replay).
  //    The empty / immediate-finish path primes from the finish tail below, so a
  //    legitimately empty response is still a well-formed OpenAI stream
  //    (role → finish → optional usage). Operators who relied on the eager role
  //    chunk can restore it with DEVIN_CONNECT_EAGER_PRIME=1.
  let primed = false;
  // V-08: `send` is the handler's bounded writer and returns a promise once the
  // socket reports backpressure. Awaiting it here (and in every emitter below)
  // is what lets a slow reader hold this producer instead of buying unbounded
  // upstream work. A synchronous consumer returns undefined, so `await` is a
  // no-op on the direct-Chat wiring that still passes a plain function.
  const prime = async () => {
    if (primed) return;
    primed = true;
    await send({ ...base, choices: [{ index: 0, delta: { role: 'assistant', content: '' }, finish_reason: null }] });
  };
  if (String(process.env.DEVIN_CONNECT_EAGER_PRIME || '') === '1') await prime();

  let content = '';
  let reasoning = '';
  let finishReason = 'stop';
  // Same guard as the non-stream twin: the tool-call arms below force
  // finish_reason='tool_calls' whenever a call was emitted, which would label a
  // turn the upstream never terminated as a clean, executable finish.
  let finishUnterminated = false;
  let usage = null;
  // Exactly what the client received as visible content, after tool-markup
  // stripping and the stop gate. `content` is the RAW upstream accumulator and
  // is used for tool parsing/session continuity; caching must use `delivered`
  // or a filtered/echoed <tool_call> block becomes the visible cached answer.
  let delivered = '';
  // Same billing passthrough as the non-stream path.
  let streamBilling = null;
  // Native tool calls (DEVIN_CONNECT_TOOL_CALL_TAGS calibrated) ride the terminal
  // finish event, not the content stream — captured here, emitted after the loop.
  let nativeToolCalls = [];
  // proto-openai-03: stream-side stop enforcement. The gate holds back a short
  // tail so a stop sequence straddling two content chunks is still caught; on a
  // hit we emit the safe prefix, flip finish_reason:'stop', and stop the stream.
  // `let` because a pre-emit transport retry must start from a fresh gate.
  let stopGate = new StopSequenceGate(stop);
  let stopHit = false;
  let matchedStop = null;
  // Emit content through the stop gate. Returns true when the stream should end.
  const sendContent = async (text) => {
    if (!text) return false;
    if (!stopGate.active) {
      await send({ ...base, choices: [{ index: 0, delta: { content: text }, finish_reason: null }] });
      delivered += text;
      return false;
    }
    const { emit, hit } = stopGate.push(text);
    if (emit) {
      await send({ ...base, choices: [{ index: 0, delta: { content: emit }, finish_reason: null }] });
      delivered += emit;
    }
    if (hit) { finishReason = 'stop'; stopHit = true; matchedStop = stopGate.matched; }
    return hit;
  };

  // Tool emulation: run content deltas through the same streaming parser the
  // Cascade path uses. It strips <tool_call> markup from the text deltas and
  // surfaces fully-closed calls; we emit each as an OpenAI tool_calls delta
  // (whole arguments at once, keyed by index — matching Cascade, not
  // token-by-token argument streaming). finish_reason flips to tool_calls.
  const toolParser = emulateTools
    ? new ToolCallStreamParser({ modelKey: params.model, provider: null, route: 'devin_connect' })
    : null;
  const collectedToolCalls = [];
  const emitToolCalls = async (calls, filter = true) => {
    // ToolGuard: same allowlist filter the chat main path applies — the
    // connect stream path used to emit any EMULATED call without checking it
    // against the client's declared tools[] (audit blind spot). Native calls
    // (structured ev.toolCalls from upstream) are trusted and pass through.
    if (filter) calls = filterToolCallsByAllowlist(calls || [], params.tools || []);
    for (const tc of calls || []) {
      await prime(); // a tool_call is a real delta — open the message before it
      const idx = collectedToolCalls.length;
      // Normalize once, before the call is recorded: the emitted delta AND the
      // returned toolCalls (session-continuity commit, cache) must agree with
      // what the client saw, or the next turn's echoed id no longer matches.
      const emitted = {
        ...tc,
        id: toClientToolCallId(tc.id, `call_${idx}_${Date.now().toString(36)}`),
      };
      collectedToolCalls.push(emitted);
      await send({ ...base, choices: [{ index: 0, delta: {
        tool_calls: [{
          index: idx,
          id: emitted.id,
          type: 'function',
          function: { name: emitted.name || 'unknown', arguments: compatArgs(emitted.argumentsJson, clineCompat) },
        }],
      }, finish_reason: null }] });
    }
  };

  // Q5 ownership: the ADAPTER owns transport retries. One bounded retry for a
  // retryable fault (5xx / ECONNRESET / server unavailable) is safe ONLY while
  // nothing has crossed the wire — `primed` is flipped by the first real delta
  // (and by the eager-prime opt-in), so a partial answer is never replayed.
  // Account failover stays in the handler.
  for (let attempt = 0; ; attempt++) {
    try {
      for await (const ev of streamChatWithEmptyRetry(params, undefined, opts)) {
        // A rescue attempt REPLACES the previous one. The generator holds
        // reasoning while an attempt can still be rescued, so no abandoned
        // reasoning delta ever reaches this consumer; the accumulators are
        // reset anyway so `content`, `delivered`, tool calls and billing can
        // never carry a discarded attempt into the returned result, the cache
        // or session continuity. (`reasoning` is reset for the same contract,
        // even though the generator already guarantees it is committed-only.)
        if (ev.type === 'attempt_reset') {
          reasoning = '';
          content = '';
          delivered = '';
          collectedToolCalls.length = 0;
          nativeToolCalls = [];
          continue;
        }
        if (ev.type === 'reasoning') {
          await prime(); // first real delta: emit the deferred role chunk first
          reasoning += ev.text;
          await send({ ...base, choices: [{ index: 0, delta: { reasoning_content: ev.text }, finish_reason: null }] });
        } else if (ev.type === 'signature') {
          // Opt-in decode (DEVIN_CONNECT_SIGNATURE_TAG). Upstream sends it after
          // the last reasoning frame and before content, so messages.js can close
          // the thinking block with a signature_delta. The generator releases it
          // only with a committed attempt, so it always seals exactly `reasoning`.
          if (reasoning && ev.reasoning_signature) {
            await send({ ...base, choices: [{ index: 0, delta: { reasoning_signature: ev.reasoning_signature }, finish_reason: null }] });
          }
        } else if (ev.type === 'content') {
          await prime(); // first real delta: emit the deferred role chunk first
          content += ev.text;
          if (toolParser) {
            const { text, toolCalls } = toolParser.feed(ev.text);
            await emitToolCalls(toolCalls);
            if (await sendContent(text)) break;
          } else {
            if (await sendContent(ev.text)) break;
          }
        } else if (ev.type === 'finish') {
          if (ev.reason) finishReason = ev.reason;
          if (ev.unterminated === true) finishUnterminated = true;
          if (ev.usage) usage = ev.usage;
          if (ev.billing) streamBilling = ev.billing;
          if (ev.toolCalls && ev.toolCalls.length) nativeToolCalls = ev.toolCalls;
        }
      }
      break;
    } catch (err) {
      if (primed || attempt >= maxRetries || !isRetryable(err)) throw err;
      // A client abort during the previous attempt (or during the backoff
      // below) must not start a fresh upstream call for a caller that is gone.
      if (params.signal?.aborted) throw err;
      // The route budget owns the total attempt count: re-issuing with no
      // turn/time left would only be refused by the next upstream call.
      if (budget && !budget.canStartAttempt()) throw err;
      const backoff = retryBaseMs * 2 ** attempt;
      log.warn(`DEVIN_CONNECT: stream retryable error (${err.code || err.status || err.message}); retry ${attempt + 1}/${maxRetries} in ${backoff}ms`);
      bumpConnect('transient_replays');
      if (!(await budgetedSleep(backoff, budget))) throw err;
      // Same post-wait gate as the non-stream loop: the budget's state after the
      // sleep, not the pre-sleep snapshot, decides whether the stream may be
      // re-issued. The real upstream error is what surfaces.
      if (budget && !budget.canStartAttempt()) throw err;
      if (params.signal?.aborted) throw err;
      // Nothing was emitted, so a clean restart cannot duplicate output.
      content = ''; reasoning = ''; finishReason = 'stop'; finishUnterminated = false; usage = null;
      streamBilling = null; nativeToolCalls = []; delivered = '';
      collectedToolCalls.length = 0;
      stopGate = new StopSequenceGate(stop);
      stopHit = false; matchedStop = null;
    }
  }

  // The upstream's OWN terminal verdict (enum 13 → 'error'), captured before the
  // tool-call arms below. Those arms set 'tool_calls' unconditionally, which turned
  // an upstream ERROR that happened to carry a decodable call into a clean,
  // executable finish.
  const terminalError = finishReason === 'error';
  // Drain any tool_call still buffered at end-of-stream, plus the trailing text.
  // Skip when a stop sequence already ended the stream (stopHit) — the tail after
  // the stop must not leak out.
  if (toolParser && !stopHit) {
    const { text, toolCalls } = toolParser.flush();
    await emitToolCalls(toolCalls);
    await sendContent(text);
    if (collectedToolCalls.length) finishReason = 'tool_calls';
  }

  // Native tool calls win over text emulation (the two are mutually exclusive:
  // when native decode is calibrated the text carries no <tool_call> markup).
  // Only emit native if emulation produced nothing, so a call is never counted
  // twice. Native arrives whole on the finish event, so it's emitted here rather
  // than inline — same wire shape as the emulated deltas above.
  if (nativeToolCalls.length && !collectedToolCalls.length) {
    await emitToolCalls(nativeToolCalls.map((tc) => ({
      id: tc.id, name: tc.name, argumentsJson: tc.arguments,
    })), false);   // native calls are trusted (structured upstream data), no allowlist filter
    finishReason = 'tool_calls';
  }

  // Fallback promotion: promote reasoning to content when no tool calls and no content exist
  // so plain prompts never return an empty visible answer to clients.
  //
  // KNOWN AND ACCEPTED COST, unlike the non-stream path above where the duplicate
  // is simply removed. Here the reasoning deltas have ALREADY gone out on the wire
  // as `reasoning_content`, and a stream cannot retract what it emitted — so a
  // client that renders reasoning will show this text twice. The alternatives were
  // weighed and are worse:
  //
  //   - Buffer reasoning and decide at end-of-stream: kills the reason streaming
  //     exists (watching the model think in real time) for EVERY turn, to spare a
  //     duplicate on the minority that end up reasoning-only.
  //   - Emit a short marker instead of the text: a strict client
  //     (kimi CLI: "response containing only thinking content") is satisfied by
  //     any text, but the user then loses the actual answer — the reasoning IS the
  //     answer on a plain prompt, which is the whole case this branch serves.
  //   - Don't promote on the stream path at all: leaves the original bug in place
  //     for streaming clients, which is the common configuration.
  //
  // So the duplicate is the price of rescuing a strict client mid-stream. It is
  // recorded here rather than left to look like an oversight; if a future change
  // can dedupe at the translator layer (messages.js already knows whether the
  // text block equals the thinking block), that is the place to do it.
  if (!content && !collectedToolCalls.length && !nativeToolCalls.length && reasoning && !stopHit) {
    // Same extraction the content path uses, for the same reason as the non-stream
    // path: the rescue nudge asks for tool markup, so a model that complies while
    // still writing into the reasoning channel would otherwise have raw
    // `<tool_call>` XML streamed to it as the visible answer.
    const promotedParse = emulateTools
      ? parseToolCallsFromText(reasoning, { modelKey: params.model, provider: null, route: 'devin_connect' })
      : { toolCalls: [], text: reasoning };
    if (promotedParse.toolCalls.length) {
      if (promotedParse.text) await sendContent(promotedParse.text);
      await emitToolCalls(promotedParse.toolCalls.map((tc) => ({
        id: tc.id, name: tc.name, argumentsJson: tc.argumentsJson ?? tc.arguments,
      })));
      content = promotedParse.text;
      finishReason = 'tool_calls';
    } else {
      await sendContent(reasoning);
      content = reasoning;
    }
  }

  // proto-openai-03: release the gate's held tail (the last few chars it was
  // withholding in case they started a stop sequence). No-op after a hit.
  if (!stopHit && stopGate.active) {
    const tail = stopGate.flush();
    if (tail) {
      await send({ ...base, choices: [{ index: 0, delta: { content: tail }, finish_reason: null }] });
      delivered += tail;
    }
  }

  // Budget honesty — same rule and same reasons as the non-stream twin above.
  // Applied only after the gate's held tail has been released, so the terminal
  // frame still follows every delivered delta: the answer is reported as over
  // budget, never retracted or cut mid-chunk. A local stop sequence keeps
  // 'stop' (the caller asked for that cause; the carrier below depends on it).
  // A turn the upstream ended in ERROR (enum 13) stays an error even when a
  // decodable tool call arrived with it; a turn that never named a stop reason is
  // not a clean tool-call finish either. The calls were already emitted as deltas
  // (a stream cannot retract them), so the terminal frame is the only place that
  // can say "these arguments are unverified". Without this, every tool-call arm
  // above (`toolParser.flush`, native #6, promoted reasoning) forced 'tool_calls'.
  if (terminalError) finishReason = 'error';
  else if (finishUnterminated && finishReason === 'tool_calls') finishReason = 'error';
  if (!stopHit) finishReason = applyOutputCap(finishReason, usage, params.completion?.maxTokens);

  // #236 log-only detector — AFTER emulation and the stop gate, on what the
  // client actually received. A delivered call (emulated or native) sets
  // finish_reason='tool_calls', so it can never be counted as prose intent;
  // `stopHit` is a local cut, not a stall.
  if (!stopHit) {
    reportProseToolIntent({
      params,
      text: delivered,
      finishEv: { reason: finishReason, toolCalls: collectedToolCalls },
      reasoning,
    });
  }

  // 4. Terminal finish chunk. Reaching here means streamChat drained cleanly
  //    (the upstream opened and completed); if it had thrown before any delta,
  //    the exception would have propagated with `primed` — and therefore the
  //    caller's `emitted` — still false, leaving first-connect recovery armed.
  //    An empty / immediate-finish response yielded no delta to prime from, so
  //    prime here to keep the stream well-formed: role → finish → optional usage.
  await prime();
  const finishChoice = { index: 0, delta: {}, finish_reason: finishReason };
  // SEED-C3a carrier — see the non-stream twin above. This frame is what the
  // internal Messages translator sees (chat.js serialises it onto the `data:`
  // line, messages.js parses it back), so the field must survive JSON round-trip
  // and is emitted only on the route opt-in and only for a genuine local stop.
  if (stopCarrier && matchedStop && finishReason === 'stop') finishChoice._windsurf_stop_sequence = matchedStop;
  await send({ ...base, choices: [finishChoice] });

  // 5. Usage-only chunk (OpenAI streams usage in a trailing choices:[] frame).
  //    O1: only when the caller opted in via stream_options.include_usage;
  //    OpenAI omits this frame by default.
  if (usage && includeUsage) {
    await send({ ...base, choices: [], usage });
  }

  return { content, delivered, reasoning, finish_reason: finishReason, usage, billing: streamBilling, toolCalls: collectedToolCalls, stop: matchedStop };
}

export const __testing = {
  newId, nowSeconds, OBJECT_COMPLETION, OBJECT_CHUNK,
  isEmptyCompletion, streamChatWithEmptyRetry,
  retryOnEmptyEnabled, retryOnEmptyMax,
};