diff --git a/packages/ollama-adapter/src/adapter.ts b/packages/ollama-adapter/src/adapter.ts index e9d29bd6..8a3bac82 100644 --- a/packages/ollama-adapter/src/adapter.ts +++ b/packages/ollama-adapter/src/adapter.ts @@ -25,6 +25,7 @@ import { resolveOverride, type OllamaAdapterOverride, } from "./overrides"; +import { createThinkSplitState, reclassifyThinkingEvents } from "./think-tags"; type OllamaChatBody = { options?: Record; @@ -78,6 +79,12 @@ export const createOllamaAdapter: AdapterFactory = ( ): ProviderAdapter => { const config = parseOllamaAdapterConfig(quirks); const inner = createOpenAIAdapter(source); + // One split state per adapter instance: the registry resolves a fresh + // adapter per request (see `createAdapterRegistry`'s own doc comment), + // so this safely tracks "are we inside a `` span" across every + // chunk of one response without leaking state between requests. + const streamThinkState = createThinkSplitState(); + const jsonThinkState = createThinkSplitState(); return { ...inner, buildRequest: (messages, model, options) => @@ -85,5 +92,9 @@ export const createOllamaAdapter: AdapterFactory = ( inner.buildRequest(messages, model, options), resolveOverride(config, model), ), + parseResponse: (sseData) => + reclassifyThinkingEvents(inner.parseResponse(sseData), streamThinkState), + parseJSONResponse: (body) => + reclassifyThinkingEvents(inner.parseJSONResponse(body), jsonThinkState), }; }; diff --git a/packages/ollama-adapter/src/index.ts b/packages/ollama-adapter/src/index.ts index 6145719e..d7bd1463 100644 --- a/packages/ollama-adapter/src/index.ts +++ b/packages/ollama-adapter/src/index.ts @@ -6,3 +6,8 @@ export { parseOllamaAdapterConfig, resolveOverride, } from "./overrides"; +export { + createThinkSplitState, + reclassifyThinkingEvents, + type ThinkSplitState, +} from "./think-tags"; diff --git a/packages/ollama-adapter/src/think-tags.test.ts b/packages/ollama-adapter/src/think-tags.test.ts new file mode 100644 index 00000000..24611f8d --- /dev/null +++ b/packages/ollama-adapter/src/think-tags.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, test } from "bun:test"; +import type { InferenceEvent } from "@intx/types/runtime"; + +import { createThinkSplitState, reclassifyThinkingEvents } from "./think-tags"; + +function textDelta(token: string, seq = 1): InferenceEvent { + return { + type: "inference.text.delta", + seq, + data: { token, partial: { text: token }, index: 0 }, + }; +} + +describe("reclassifyThinkingEvents", () => { + test("a whole ... span in one token becomes thinking-delta, not text-delta", () => { + const state = createThinkSplitState(); + const out = reclassifyThinkingEvents( + [textDelta("plan the approachHere is the answer.")], + state, + ); + expect(out).toHaveLength(2); + expect(out[0]?.type).toBe("inference.thinking.delta"); + expect((out[0]?.data as { token: string }).token).toBe("plan the approach"); + expect(out[1]?.type).toBe("inference.text.delta"); + expect((out[1]?.data as { token: string }).token).toBe( + "Here is the answer.", + ); + }); + + test("a span split across multiple chunks stays classified as thinking across the boundary", () => { + const state = createThinkSplitState(); + const first = reclassifyThinkingEvents( + [textDelta("step one, ")], + state, + ); + const second = reclassifyThinkingEvents( + [textDelta("step twofinal reply")], + state, + ); + + expect(first).toHaveLength(1); + expect(first[0]?.type).toBe("inference.thinking.delta"); + expect(second).toHaveLength(2); + expect(second[0]?.type).toBe("inference.thinking.delta"); + expect((second[0]?.data as { token: string }).token).toBe("step two"); + expect(second[1]?.type).toBe("inference.text.delta"); + expect((second[1]?.data as { token: string }).token).toBe("final reply"); + }); + + test("ordinary text with no tag passes through as text-delta unchanged", () => { + const state = createThinkSplitState(); + const out = reclassifyThinkingEvents( + [textDelta("just a normal reply")], + state, + ); + expect(out).toEqual([textDelta("just a normal reply")]); + }); + + test("non-text events (tool calls, done) pass through untouched", () => { + const state = createThinkSplitState(); + const toolCallStart: InferenceEvent = { + type: "inference.tool_call.start", + seq: 1, + data: { + callId: "call-1", + name: "slack__post_message", + partial: { text: "" }, + }, + }; + const out = reclassifyThinkingEvents([toolCallStart], state); + expect(out).toEqual([toolCallStart]); + }); + + test("thinking events never share an index with the text stream, so the harness's per-index blockMap can't collide them", () => { + const state = createThinkSplitState(); + const out = reclassifyThinkingEvents( + [textDelta("internal notesvisible reply", 3)], + state, + ); + const thinkingEvent = out.find( + (event) => event.type === "inference.thinking.delta", + ); + const textEvent = out.find( + (event) => event.type === "inference.text.delta", + ); + expect((thinkingEvent?.data as { index?: number }).index).not.toBe( + (textEvent?.data as { index?: number }).index, + ); + expect((textEvent?.data as { index?: number }).index).toBe(0); + }); + + test("cumulative partial.text/partial.thinking reflect only their own kind, never the raw tags", () => { + const state = createThinkSplitState(); + const out = reclassifyThinkingEvents( + [textDelta("internal notesvisible reply")], + state, + ); + const thinkingEvent = out.find( + (event) => event.type === "inference.thinking.delta", + ); + const textEvent = out.find( + (event) => event.type === "inference.text.delta", + ); + expect( + (thinkingEvent?.data as { partial: { thinking?: string } }).partial + .thinking, + ).toBe("internal notes"); + expect( + (textEvent?.data as { partial: { text: string } }).partial.text, + ).toBe("visible reply"); + expect( + (textEvent?.data as { partial: { text: string } }).partial.text, + ).not.toContain(""); + }); +}); diff --git a/packages/ollama-adapter/src/think-tags.ts b/packages/ollama-adapter/src/think-tags.ts new file mode 100644 index 00000000..830693e9 --- /dev/null +++ b/packages/ollama-adapter/src/think-tags.ts @@ -0,0 +1,147 @@ +// Ollama's OpenAI-compatible endpoint never populates the `reasoning`/ +// `reasoning_content` delta fields `@intx/inference`'s OpenAI provider +// looks for (see `providers/openai.js`'s `reasoningFieldNames` handling). +// gpt-oss and qwen instead emit their chain-of-thought inline inside the +// ordinary `content` field, wrapped in `…`. Left alone, that +// text is indistinguishable from the reply and rides every hop downstream +// as a genuine `inference.text.delta` — this is the CL-6654 leak. This +// module reclassifies it into `inference.thinking.delta` before anything +// else ever sees it, at the one place that already knows these tokens came +// from Ollama. +import type { InferenceEvent } from "@intx/types/runtime"; + +/** Carries the split state across every chunk of one streamed response — + * a `` tag can land at a chunk boundary, so "are we inside a + * thinking span" has to survive from one `parseResponse` call to the next. */ +export type ThinkSplitState = { + inThink: boolean; + /** Once a `` tag has appeared at all, every later text-delta gets + * its `partial` recomputed from this module's own tally rather than the + * built-in adapter's — otherwise a token after the closing tag would + * still carry the raw-tagged cumulative text the built-in parser tracked + * on its own. Before the first tag, events pass through byte-identical. */ + everInThink: boolean; + textAcc: string; + thinkingAcc: string; +}; + +export function createThinkSplitState(): ThinkSplitState { + return { inThink: false, everInThink: false, textAcc: "", thinkingAcc: "" }; +} + +const THINK_OPEN = ""; +const THINK_CLOSE = ""; + +/** The harness (`@intx/inference`'s `dist/harness.js`) keys its per-index + * `blockMap` by `event.data.index` and throws a `ProtocolMismatchError` the + * moment two different block kinds land at the same index — so the + * thinking half of a split can never reuse the text index the built-in + * OpenAI adapter already assigned to Ollama's one undifferentiated content + * stream. Ollama's own indexer starts at 0 and only counts up (one index + * per real block: text, then any tool calls), so a fixed negative + * sentinel is guaranteed to never collide with one it hands out. */ +const THINKING_BLOCK_INDEX = -1; + +type TokenSplit = { + readonly textToken: string; + readonly thinkingToken: string; +}; + +/** Peels ``/`` spans out of one token, folding the result + * into the running cumulative text/thinking strings. A tag never splits + * across two tokens in practice, but a token straddling the tag boundary + * (open and content in the same token, or close and content) is still + * handled correctly by looping until the whole token is consumed. */ +function splitToken(state: ThinkSplitState, token: string): TokenSplit { + let remaining = token; + let textToken = ""; + let thinkingToken = ""; + + while (remaining.length > 0) { + if (!state.inThink) { + const openIndex = remaining.indexOf(THINK_OPEN); + if (openIndex === -1) { + textToken += remaining; + break; + } + textToken += remaining.slice(0, openIndex); + remaining = remaining.slice(openIndex + THINK_OPEN.length); + state.inThink = true; + } else { + const closeIndex = remaining.indexOf(THINK_CLOSE); + if (closeIndex === -1) { + thinkingToken += remaining; + break; + } + thinkingToken += remaining.slice(0, closeIndex); + remaining = remaining.slice(closeIndex + THINK_CLOSE.length); + state.inThink = false; + } + } + + state.textAcc += textToken; + state.thinkingAcc += thinkingToken; + return { textToken, thinkingToken }; +} + +/** + * Rewrites one `parseResponse`/`parseJSONResponse` result so a + * `…` span in an `inference.text.delta`'s token becomes an + * `inference.thinking.delta` instead — every other event (tool calls, + * usage, `inference.done`, ...) passes through untouched. `state` is + * mutated in place so a caller threads the same instance across every + * chunk of one response. + */ +export function reclassifyThinkingEvents( + events: readonly InferenceEvent[], + state: ThinkSplitState, +): InferenceEvent[] { + const output: InferenceEvent[] = []; + + for (const event of events) { + if (event.type !== "inference.text.delta") { + output.push(event); + continue; + } + + const token = event.data.token; + if (!state.inThink && !state.everInThink && !token.includes(THINK_OPEN)) { + output.push(event); + continue; + } + state.everInThink = true; + + const { textToken, thinkingToken } = splitToken(state, token); + const partial = { + text: state.textAcc, + ...(state.thinkingAcc !== "" ? { thinking: state.thinkingAcc } : {}), + }; + + if (thinkingToken !== "") { + output.push({ + type: "inference.thinking.delta", + seq: event.seq, + data: { + token: thinkingToken, + partial, + index: THINKING_BLOCK_INDEX, + }, + }); + } + if (textToken !== "") { + output.push({ + type: "inference.text.delta", + seq: event.seq, + data: { + token: textToken, + partial, + ...(event.data.index !== undefined + ? { index: event.data.index } + : {}), + }, + }); + } + } + + return output; +}