From d6d75ade74a5d56374f4944a5a0ff831b7bec869 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 12:29:07 +0200 Subject: [PATCH 01/20] refactor(llm): extract stream content parsing into a named seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pull the provider-shape → (text, reasoning) walker out of chat_string into _make_stream_content_parser, pinned by a characterization test. No behavior change. --- packages/ai/src/ai/common/chat.py | 84 +++++++++++-------- .../ai/tests/ai/common/test_chat_content.py | 68 +++++++++++++++ 2 files changed, 115 insertions(+), 37 deletions(-) create mode 100644 packages/ai/tests/ai/common/test_chat_content.py diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index 64c403914..872fa045e 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -93,6 +93,50 @@ def flush(): return feed +def _make_stream_content_parser(has_reasoning_sink: bool): + """Split streamed content (str, or Anthropic/LangChain-v1 typed blocks) into + (visible_text, reasoning_text); also strips inline ```` from str content. + """ + think_split = _make_think_tag_splitter() + state = {'signature_noted': False} + + def feed(content): + text = '' + thinking = '' + if isinstance(content, list): + for b in content: + if not isinstance(b, dict): + continue + btype = b.get('type', '') + if btype == 'thinking': + # carries either text deltas or a signature-only final delta. + piece_text = b.get('thinking') or b.get('text') or '' + if piece_text: + thinking += piece_text + elif b.get('signature') and not state['signature_noted'] and has_reasoning_sink: + thinking += ( + '_Extended thinking ran, but this stream only delivered the ' + 'block verification signature, not the readable chain-of-thought ' + 'text. The answer below still reflects internal reasoning._\n\n' + ) + state['signature_noted'] = True + elif btype == 'reasoning': + # LangChain v1 standard block (thinking → reasoning). + piece_text = b.get('reasoning') or b.get('text') or '' + if piece_text: + thinking += piece_text + elif btype == 'text' or not btype: + text += b.get('text', '') + elif isinstance(content, str): + text, thinking_inline = think_split(content) + if thinking_inline: + thinking += thinking_inline + return text, thinking + + feed.flush = think_split.flush # type: ignore[attr-defined] + return feed + + class ChatBase: """ Abstract base class for all chat drivers with configurable token allocation. @@ -595,43 +639,9 @@ def on_chunk_w(t): try: parts = [] finish_reason: Optional[str] = None - _signature_only_note_sent = False - _think_split = _make_think_tag_splitter() + parse = _make_stream_content_parser(on_reasoning_chunk_w is not None) for piece in _llm.stream(prompt, **_stop_kwargs()): - # content: str for OpenAI-style, list of typed blocks for Anthropic. - content = piece.content - text = '' - thinking_delta = '' - if isinstance(content, list): - for b in content: - if not isinstance(b, dict): - continue - btype = b.get('type', '') - if btype == 'thinking': - # carries either text deltas or a signature-only final delta. - piece_text = b.get('thinking') or b.get('text') or '' - if piece_text: - thinking_delta += piece_text - elif b.get('signature') and not _signature_only_note_sent: - if on_reasoning_chunk_w is not None: - thinking_delta += ( - '_Extended thinking ran, but this stream only delivered the ' - 'block verification signature, not the readable chain-of-thought ' - 'text. The answer below still reflects internal reasoning._\n\n' - ) - _signature_only_note_sent = True - elif btype == 'reasoning': - # LangChain v1 standard block (thinking → reasoning). - piece_text = b.get('reasoning') or b.get('text') or '' - if piece_text: - thinking_delta += piece_text - elif btype == 'text' or not btype: - text += b.get('text', '') - elif isinstance(content, str): - # Strip inline `...` (Perplexity sonar-reasoning fallback). - text, _thinking_inline = _think_split(content) - if _thinking_inline: - thinking_delta += _thinking_inline + text, thinking_delta = parse(piece.content) if thinking_delta and on_reasoning_chunk_w is not None: on_reasoning_chunk_w(thinking_delta) if text: @@ -641,7 +651,7 @@ def on_chunk_w(t): if reason: finish_reason = reason # Drain chars buffered by the splitter (partial-tag tail). - tail_visible, tail_reasoning = _think_split.flush() + tail_visible, tail_reasoning = parse.flush() if tail_visible: on_chunk_w(tail_visible) parts.append(tail_visible) diff --git a/packages/ai/tests/ai/common/test_chat_content.py b/packages/ai/tests/ai/common/test_chat_content.py new file mode 100644 index 000000000..dca531756 --- /dev/null +++ b/packages/ai/tests/ai/common/test_chat_content.py @@ -0,0 +1,68 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins _make_stream_content_parser: the provider-shape → (text, reasoning) seam +that the LLM adapters will absorb. Behavior must not drift during that refactor. +""" + +from ai.common.chat import _make_stream_content_parser + + +def test_str_passthrough(): + # think-splitter holds a possible partial `` tail until flush. + parse = _make_stream_content_parser(True) + text, thinking = parse('hello world') + tail_text, _ = parse.flush() + assert text + tail_text == 'hello world' + assert thinking == '' + + +def test_anthropic_thinking_then_text(): + parse = _make_stream_content_parser(True) + text, thinking = parse( + [ + {'type': 'thinking', 'thinking': 'reasoning...'}, + {'type': 'text', 'text': 'answer'}, + ] + ) + assert text == 'answer' + assert thinking == 'reasoning...' + + +def test_reasoning_block_langchain_v1(): + parse = _make_stream_content_parser(True) + assert parse([{'type': 'reasoning', 'reasoning': 'r'}]) == ('', 'r') + + +def test_text_block_without_type(): + parse = _make_stream_content_parser(True) + assert parse([{'text': 'plain'}]) == ('plain', '') + + +def test_non_dict_blocks_skipped(): + parse = _make_stream_content_parser(True) + assert parse(['stray', {'type': 'text', 'text': 'y'}]) == ('y', '') + + +def test_signature_only_note_emitted_once(): + parse = _make_stream_content_parser(True) + _, first = parse([{'type': 'thinking', 'signature': 'sig1'}]) + assert 'verification signature' in first + _, second = parse([{'type': 'thinking', 'signature': 'sig2'}]) + assert second == '' + + +def test_signature_note_suppressed_without_reasoning_sink(): + parse = _make_stream_content_parser(False) + _, thinking = parse([{'type': 'thinking', 'signature': 'sig'}]) + assert thinking == '' + + +def test_inline_think_split_across_feed_and_flush(): + parse = _make_stream_content_parser(True) + text, thinking = parse('beforecotafter') + tail_text, tail_thinking = parse.flush() + assert text + tail_text == 'beforeafter' + assert thinking + tail_thinking == 'cot' From ec1a7df373c51a984e30f5542eb115942cc93a11 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 12:53:52 +0200 Subject: [PATCH 02/20] feat(llm): add normalized provider Adapter interface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Event (thinking/text/done) + Adapter Protocol with opaque history items — the seam ChatBase will consume. Not wired yet. RFC: discussion #1679. --- packages/ai/src/ai/common/llm_adapter.py | 36 +++++++++++++++++ .../ai/tests/ai/common/test_llm_adapter.py | 40 +++++++++++++++++++ 2 files changed, 76 insertions(+) create mode 100644 packages/ai/src/ai/common/llm_adapter.py create mode 100644 packages/ai/tests/ai/common/test_llm_adapter.py diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py new file mode 100644 index 000000000..1aec6e4b7 --- /dev/null +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -0,0 +1,36 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Normalized LLM provider interface: one Event shape for every provider. + +ChatBase consumes Adapters and never touches provider-native content shapes. +Design: repo discussion #1679 (RFC — virtualized provider Adapter). +""" + +from dataclasses import dataclass, field +from typing import Any, Iterator, Protocol, runtime_checkable + + +@dataclass +class Event: + """One normalized streaming event: a display delta, or the terminal ``done``.""" + + type: str # "thinking" | "text" | "done" + text: str = '' + items: list[Any] = field(default_factory=list) + + +@runtime_checkable +class Adapter(Protocol): + """Provider adapter: owns provider-native ``history``, streams normalized Events. + + Yields ``Event("thinking"|"text")`` deltas in order, then exactly one terminal + ``Event("done", items=...)``. ``items`` is provider-native and OPAQUE: append it + to ``history`` verbatim — never inspect, edit, reorder, or reserialize it. + """ + + history: list[Any] + + def stream(self, user_text: str) -> Iterator[Event]: ... diff --git a/packages/ai/tests/ai/common/test_llm_adapter.py b/packages/ai/tests/ai/common/test_llm_adapter.py new file mode 100644 index 000000000..8726f3e0f --- /dev/null +++ b/packages/ai/tests/ai/common/test_llm_adapter.py @@ -0,0 +1,40 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins the normalized Adapter interface: Event shape + structural Adapter check.""" + +from ai.common.llm_adapter import Adapter, Event + + +def test_event_defaults(): + e = Event('text') + assert (e.type, e.text, e.items) == ('text', '', []) + + +def test_event_done_carries_opaque_items(): + e = Event('done', items=[{'role': 'assistant', 'content': 'x'}]) + assert e.type == 'done' + assert e.items == [{'role': 'assistant', 'content': 'x'}] + + +def test_adapter_is_structural_protocol(): + class FakeAdapter: + def __init__(self): + self.history: list = [] + + def stream(self, user_text): + yield Event('thinking', 'reasoning') + yield Event('text', 'answer') + self.history.append('raw-turn') + yield Event('done', items=['raw-turn']) + + fake = FakeAdapter() + assert isinstance(fake, Adapter) + + events = list(fake.stream('question')) + assert events[0] == Event('thinking', 'reasoning') + assert events[1] == Event('text', 'answer') + assert events[-1].type == 'done' and events[-1].items == ['raw-turn'] + assert fake.history == ['raw-turn'] From a2e18f06ea420a7cd59ad8fd3823f3e41bf13989 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 14:01:33 +0200 Subject: [PATCH 03/20] refactor(llm): move stream content parser to the adapter module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Relocate _make_stream_content_parser / _make_think_tag_splitter into llm_adapter so adapters can reuse them without a chat↔adapter import cycle. --- packages/ai/src/ai/common/chat.py | 100 +----------------- packages/ai/src/ai/common/llm_adapter.py | 99 +++++++++++++++++ .../ai/tests/ai/common/test_chat_content.py | 2 +- 3 files changed, 101 insertions(+), 100 deletions(-) diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index 872fa045e..750d7b174 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -20,6 +20,7 @@ from ai.common.util import parseJson from ai.common.validation import validate_model_name, validate_max_tokens, validate_prompt from ai.common.llm_native_stream import STOP_SEQUENCES_VAR, dispatch_native_chat_stream +from ai.common.llm_adapter import _make_stream_content_parser def _stop_kwargs() -> dict: @@ -38,105 +39,6 @@ def _stop_kwargs() -> dict: return {'stop': stop} if stop else {} -def _make_think_tag_splitter(): - """Split ``...`` CoT out of the content stream (Ollama, Perplexity). - - Returns a ``feed(text) -> (visible, reasoning)`` closure; tags may span deltas. - """ - OPEN, CLOSE = '', '' - state = {'mode': 'visible', 'buf': ''} - - def feed(text: str): - if not text: - return '', '' - buf = state['buf'] + text - visible_parts: list = [] - reasoning_parts: list = [] - while buf: - if state['mode'] == 'visible': - idx = buf.find(OPEN) - if idx < 0: - # Hold back trailing chars that could be a partial ''. - safe = len(buf) - (len(OPEN) - 1) - if safe > 0: - visible_parts.append(buf[:safe]) - buf = buf[safe:] - break - if idx: - visible_parts.append(buf[:idx]) - buf = buf[idx + len(OPEN) :] - state['mode'] = 'thinking' - else: - idx = buf.find(CLOSE) - if idx < 0: - safe = len(buf) - (len(CLOSE) - 1) - if safe > 0: - reasoning_parts.append(buf[:safe]) - buf = buf[safe:] - break - if idx: - reasoning_parts.append(buf[:idx]) - buf = buf[idx + len(CLOSE) :] - state['mode'] = 'visible' - state['buf'] = buf - return ''.join(visible_parts), ''.join(reasoning_parts) - - def flush(): - """Emit anything buffered at end-of-stream (e.g. an unterminated tag).""" - tail = state['buf'] - state['buf'] = '' - if state['mode'] == 'thinking': - return '', tail - return tail, '' - - feed.flush = flush # type: ignore[attr-defined] - return feed - - -def _make_stream_content_parser(has_reasoning_sink: bool): - """Split streamed content (str, or Anthropic/LangChain-v1 typed blocks) into - (visible_text, reasoning_text); also strips inline ```` from str content. - """ - think_split = _make_think_tag_splitter() - state = {'signature_noted': False} - - def feed(content): - text = '' - thinking = '' - if isinstance(content, list): - for b in content: - if not isinstance(b, dict): - continue - btype = b.get('type', '') - if btype == 'thinking': - # carries either text deltas or a signature-only final delta. - piece_text = b.get('thinking') or b.get('text') or '' - if piece_text: - thinking += piece_text - elif b.get('signature') and not state['signature_noted'] and has_reasoning_sink: - thinking += ( - '_Extended thinking ran, but this stream only delivered the ' - 'block verification signature, not the readable chain-of-thought ' - 'text. The answer below still reflects internal reasoning._\n\n' - ) - state['signature_noted'] = True - elif btype == 'reasoning': - # LangChain v1 standard block (thinking → reasoning). - piece_text = b.get('reasoning') or b.get('text') or '' - if piece_text: - thinking += piece_text - elif btype == 'text' or not btype: - text += b.get('text', '') - elif isinstance(content, str): - text, thinking_inline = think_split(content) - if thinking_inline: - thinking += thinking_inline - return text, thinking - - feed.flush = think_split.flush # type: ignore[attr-defined] - return feed - - class ChatBase: """ Abstract base class for all chat drivers with configurable token allocation. diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 1aec6e4b7..32bdc368f 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -34,3 +34,102 @@ class Adapter(Protocol): history: list[Any] def stream(self, user_text: str) -> Iterator[Event]: ... + + +def _make_think_tag_splitter(): + """Split ``...`` CoT out of the content stream (Ollama, Perplexity). + + Returns a ``feed(text) -> (visible, reasoning)`` closure; tags may span deltas. + """ + OPEN, CLOSE = '', '' + state = {'mode': 'visible', 'buf': ''} + + def feed(text: str): + if not text: + return '', '' + buf = state['buf'] + text + visible_parts: list = [] + reasoning_parts: list = [] + while buf: + if state['mode'] == 'visible': + idx = buf.find(OPEN) + if idx < 0: + # Hold back trailing chars that could be a partial ''. + safe = len(buf) - (len(OPEN) - 1) + if safe > 0: + visible_parts.append(buf[:safe]) + buf = buf[safe:] + break + if idx: + visible_parts.append(buf[:idx]) + buf = buf[idx + len(OPEN) :] + state['mode'] = 'thinking' + else: + idx = buf.find(CLOSE) + if idx < 0: + safe = len(buf) - (len(CLOSE) - 1) + if safe > 0: + reasoning_parts.append(buf[:safe]) + buf = buf[safe:] + break + if idx: + reasoning_parts.append(buf[:idx]) + buf = buf[idx + len(CLOSE) :] + state['mode'] = 'visible' + state['buf'] = buf + return ''.join(visible_parts), ''.join(reasoning_parts) + + def flush(): + """Emit anything buffered at end-of-stream (e.g. an unterminated tag).""" + tail = state['buf'] + state['buf'] = '' + if state['mode'] == 'thinking': + return '', tail + return tail, '' + + feed.flush = flush # type: ignore[attr-defined] + return feed + + +def _make_stream_content_parser(has_reasoning_sink: bool): + """Split streamed content (str, or Anthropic/LangChain-v1 typed blocks) into + (visible_text, reasoning_text); also strips inline ```` from str content. + """ + think_split = _make_think_tag_splitter() + state = {'signature_noted': False} + + def feed(content): + text = '' + thinking = '' + if isinstance(content, list): + for b in content: + if not isinstance(b, dict): + continue + btype = b.get('type', '') + if btype == 'thinking': + # carries either text deltas or a signature-only final delta. + piece_text = b.get('thinking') or b.get('text') or '' + if piece_text: + thinking += piece_text + elif b.get('signature') and not state['signature_noted'] and has_reasoning_sink: + thinking += ( + '_Extended thinking ran, but this stream only delivered the ' + 'block verification signature, not the readable chain-of-thought ' + 'text. The answer below still reflects internal reasoning._\n\n' + ) + state['signature_noted'] = True + elif btype == 'reasoning': + # LangChain v1 standard block (thinking → reasoning). + piece_text = b.get('reasoning') or b.get('text') or '' + if piece_text: + thinking += piece_text + elif btype == 'text' or not btype: + text += b.get('text', '') + elif isinstance(content, str): + text, thinking_inline = think_split(content) + if thinking_inline: + thinking += thinking_inline + return text, thinking + + feed.flush = think_split.flush # type: ignore[attr-defined] + return feed diff --git a/packages/ai/tests/ai/common/test_chat_content.py b/packages/ai/tests/ai/common/test_chat_content.py index dca531756..c234ff575 100644 --- a/packages/ai/tests/ai/common/test_chat_content.py +++ b/packages/ai/tests/ai/common/test_chat_content.py @@ -7,7 +7,7 @@ that the LLM adapters will absorb. Behavior must not drift during that refactor. """ -from ai.common.chat import _make_stream_content_parser +from ai.common.llm_adapter import _make_stream_content_parser def test_str_passthrough(): From b8203915286c45d2fc67e5bad7ec8c1895a24634 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 14:32:09 +0200 Subject: [PATCH 04/20] feat(llm): add LangChainAdapter Wraps a LangChain chat model as an Event stream (reusing the shared parser); covers the non-reasoning providers. done.items is the assistant text turn. --- packages/ai/src/ai/common/llm_adapter.py | 32 +++++++++++ .../ai/common/test_llm_adapter_langchain.py | 53 +++++++++++++++++++ 2 files changed, 85 insertions(+) create mode 100644 packages/ai/tests/ai/common/test_llm_adapter_langchain.py diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 32bdc368f..58aea0404 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -133,3 +133,35 @@ def feed(content): feed.flush = think_split.flush # type: ignore[attr-defined] return feed + + +class LangChainAdapter: + """Wraps a LangChain chat model so non-reasoning providers speak the Event contract. + + ``done.items`` is the assistant text turn — LangChain carries no opaque reasoning state. + """ + + def __init__(self, llm: Any, history: list[Any] | None = None): + self.llm = llm + self.history: list[Any] = history if history is not None else [] + + def stream(self, user_text: str) -> Iterator[Event]: + self.history.append({'role': 'user', 'content': user_text}) + parse = _make_stream_content_parser(True) + parts: list[str] = [] + for piece in self.llm.stream(self.history): + text, thinking = parse(piece.content) + if thinking: + yield Event('thinking', thinking) + if text: + parts.append(text) + yield Event('text', text) + tail_text, tail_thinking = parse.flush() + if tail_thinking: + yield Event('thinking', tail_thinking) + if tail_text: + parts.append(tail_text) + yield Event('text', tail_text) + assistant = {'role': 'assistant', 'content': ''.join(parts)} + self.history.append(assistant) + yield Event('done', items=[assistant]) diff --git a/packages/ai/tests/ai/common/test_llm_adapter_langchain.py b/packages/ai/tests/ai/common/test_llm_adapter_langchain.py new file mode 100644 index 000000000..f58a76528 --- /dev/null +++ b/packages/ai/tests/ai/common/test_llm_adapter_langchain.py @@ -0,0 +1,53 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins LangChainAdapter: normalizes LangChain content to Events and records history.""" + +from ai.common.llm_adapter import Event, LangChainAdapter + + +class _Piece: + def __init__(self, content): + self.content = content + + +class _FakeLLM: + def __init__(self, pieces): + self._pieces = pieces + self.seen = None + + def stream(self, messages): + self.seen = list(messages) + for p in self._pieces: + yield _Piece(p) + + +def test_normalizes_blocks_and_records_history(): + llm = _FakeLLM([[{'type': 'thinking', 'thinking': 'r'}, {'type': 'text', 'text': 'Hi there'}]]) + adapter = LangChainAdapter(llm) + + events = list(adapter.stream('q')) + + assert Event('thinking', 'r') in events + assert ''.join(e.text for e in events if e.type == 'text') == 'Hi there' + + done = events[-1] + assert done.type == 'done' + assert done.items == [{'role': 'assistant', 'content': 'Hi there'}] + + # history: user turn seen by the model, assistant turn appended after + assert llm.seen == [{'role': 'user', 'content': 'q'}] + assert adapter.history == [ + {'role': 'user', 'content': 'q'}, + {'role': 'assistant', 'content': 'Hi there'}, + ] + + +def test_str_content_flushes_buffered_tail(): + # str content rides the think-splitter, which buffers a possible partial tag. + llm = _FakeLLM(['hello world']) + adapter = LangChainAdapter(llm) + text = ''.join(e.text for e in adapter.stream('q') if e.type == 'text') + assert text == 'hello world' From 6ae43013f5591dd71725037e846f32f522848c2f Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 14:50:07 +0200 Subject: [PATCH 05/20] feat(llm): add AnthropicAdapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Streams the Messages API (thinking/text deltas → Events); done.items is the assembled content with signatures intact — appended to history verbatim. --- packages/ai/src/ai/common/llm_adapter.py | 39 ++++++++ .../ai/common/test_llm_adapter_anthropic.py | 97 +++++++++++++++++++ 2 files changed, 136 insertions(+) create mode 100644 packages/ai/tests/ai/common/test_llm_adapter_anthropic.py diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 58aea0404..0cc14f906 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -165,3 +165,42 @@ def stream(self, user_text: str) -> Iterator[Event]: assistant = {'role': 'assistant', 'content': ''.join(parts)} self.history.append(assistant) yield Event('done', items=[assistant]) + + +class AnthropicAdapter: + """history is a Messages-API `messages` list; done.items is the assembled content + (thinking `signature` / redacted blocks intact) — append verbatim, never rebuild. + """ + + def __init__( + self, + client: Any, + model: str, + max_tokens: int = 16000, + thinking: dict | None = None, + history: list[Any] | None = None, + ): + self.client = client + self.model = model + self.max_tokens = max_tokens + self.thinking = thinking + self.history: list[Any] = history if history is not None else [] + + def stream(self, user_text: str) -> Iterator[Event]: + self.history.append({'role': 'user', 'content': user_text}) + kwargs: dict[str, Any] = {'model': self.model, 'max_tokens': self.max_tokens, 'messages': self.history} + if self.thinking: + kwargs['thinking'] = self.thinking + with self.client.messages.stream(**kwargs) as stream: + for ev in stream: + if getattr(ev, 'type', '') != 'content_block_delta': + continue + delta = ev.delta + dtype = getattr(delta, 'type', '') + if dtype == 'thinking_delta': + yield Event('thinking', getattr(delta, 'thinking', '') or '') + elif dtype == 'text_delta': + yield Event('text', getattr(delta, 'text', '') or '') + final = stream.get_final_message() + self.history.append({'role': 'assistant', 'content': final.content}) + yield Event('done', items=final.content) diff --git a/packages/ai/tests/ai/common/test_llm_adapter_anthropic.py b/packages/ai/tests/ai/common/test_llm_adapter_anthropic.py new file mode 100644 index 000000000..6b99099f2 --- /dev/null +++ b/packages/ai/tests/ai/common/test_llm_adapter_anthropic.py @@ -0,0 +1,97 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins AnthropicAdapter: thinking/text deltas → Events, opaque content in done.items.""" + +from ai.common.llm_adapter import AnthropicAdapter, Event + + +class _Delta: + def __init__(self, type, thinking='', text=''): + self.type = type + self.thinking = thinking + self.text = text + + +class _Ev: + def __init__(self, type, delta=None): + self.type = type + self.delta = delta + + +class _Final: + def __init__(self, content): + self.content = content + + +class _Stream: + def __init__(self, events, final): + self._events = events + self._final = final + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def __iter__(self): + return iter(self._events) + + def get_final_message(self): + return self._final + + +class _Messages: + def __init__(self, stream): + self._stream = stream + self.seen = None + + def stream(self, **kwargs): + self.seen = {**kwargs, 'messages': list(kwargs['messages'])} + return self._stream + + +class _Client: + def __init__(self, stream): + self.messages = _Messages(stream) + + +def test_deltas_to_events_and_opaque_done_items(): + final_content = [ + {'type': 'thinking', 'thinking': 'reasoning', 'signature': 'sig'}, + {'type': 'text', 'text': 'answer'}, + ] + events = [ + _Ev('content_block_delta', _Delta('thinking_delta', thinking='reasoning')), + _Ev('content_block_delta', _Delta('text_delta', text='answer')), + _Ev('message_stop'), + ] + client = _Client(_Stream(events, _Final(final_content))) + adapter = AnthropicAdapter( + client, model='claude-sonnet-4-6', thinking={'type': 'adaptive', 'display': 'summarized'} + ) + + out = list(adapter.stream('q')) + + assert Event('thinking', 'reasoning') in out + assert Event('text', 'answer') in out + done = out[-1] + assert done.type == 'done' + assert done.items == final_content # opaque, verbatim (signature intact) + + assert adapter.history == [ + {'role': 'user', 'content': 'q'}, + {'role': 'assistant', 'content': final_content}, + ] + assert client.messages.seen['thinking'] == {'type': 'adaptive', 'display': 'summarized'} + assert client.messages.seen['messages'] == [{'role': 'user', 'content': 'q'}] + + +def test_thinking_omitted_when_none(): + client = _Client(_Stream([_Ev('content_block_delta', _Delta('text_delta', text='hi'))], _Final([]))) + adapter = AnthropicAdapter(client, model='claude-3-haiku') + list(adapter.stream('q')) + assert 'thinking' not in client.messages.seen From 1d8ebccf98a6dedd0bea910a9a5ca95ad305f311 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 18:15:40 +0200 Subject: [PATCH 06/20] feat(llm): add OpenAIAdapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Streams the Responses API (reasoning/text deltas → Events); done.items are the output items with encrypted reasoning, extended into history verbatim. --- packages/ai/src/ai/common/llm_adapter.py | 31 +++++++ .../ai/common/test_llm_adapter_openai.py | 93 +++++++++++++++++++ 2 files changed, 124 insertions(+) create mode 100644 packages/ai/tests/ai/common/test_llm_adapter_openai.py diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 0cc14f906..6c60093a0 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -204,3 +204,34 @@ def stream(self, user_text: str) -> Iterator[Event]: final = stream.get_final_message() self.history.append({'role': 'assistant', 'content': final.content}) yield Event('done', items=final.content) + + +class OpenAIAdapter: + """history is a Responses-API `input` item list; done.items are the output items + (encrypted reasoning + assistant) — extend history verbatim, never replay a subset. + """ + + def __init__(self, client: Any, model: str, effort: str = 'medium', history: list[Any] | None = None): + self.client = client + self.model = model + self.effort = effort + self.history: list[Any] = history if history is not None else [] + + def stream(self, user_text: str) -> Iterator[Event]: + self.history.append({'role': 'user', 'content': user_text}) + with self.client.responses.stream( + model=self.model, + input=self.history, + store=False, # stateless: encrypted_content rides back on the reasoning items. + reasoning={'effort': self.effort, 'summary': 'auto', 'context': 'all_turns'}, + ) as stream: + for ev in stream: + etype = getattr(ev, 'type', '') + if etype in ('response.reasoning_summary_text.delta', 'response.reasoning_text.delta'): + yield Event('thinking', getattr(ev, 'delta', '') or '') + elif etype == 'response.output_text.delta': + yield Event('text', getattr(ev, 'delta', '') or '') + final = stream.get_final_response() + items = [item.model_dump() for item in final.output] + self.history.extend(items) + yield Event('done', items=items) diff --git a/packages/ai/tests/ai/common/test_llm_adapter_openai.py b/packages/ai/tests/ai/common/test_llm_adapter_openai.py new file mode 100644 index 000000000..bd12b14b9 --- /dev/null +++ b/packages/ai/tests/ai/common/test_llm_adapter_openai.py @@ -0,0 +1,93 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins OpenAIAdapter: Responses deltas → Events, opaque output items in done.items.""" + +from ai.common.llm_adapter import Event, OpenAIAdapter + + +class _Ev: + def __init__(self, type, delta=''): + self.type = type + self.delta = delta + + +class _Item: + def __init__(self, dump): + self._dump = dump + + def model_dump(self): + return self._dump + + +class _Final: + def __init__(self, output): + self.output = output + + +class _Stream: + def __init__(self, events, final): + self._events = events + self._final = final + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def __iter__(self): + return iter(self._events) + + def get_final_response(self): + return self._final + + +class _Responses: + def __init__(self, stream): + self._stream = stream + self.seen = None + + def stream(self, **kwargs): + self.seen = {**kwargs, 'input': list(kwargs['input'])} + return self._stream + + +class _Client: + def __init__(self, stream): + self.responses = _Responses(stream) + + +def test_deltas_to_events_and_opaque_output_items(): + events = [ + _Ev('response.reasoning_summary_text.delta', 'think'), + _Ev('response.output_text.delta', 'ans'), + ] + output = [ + _Item({'type': 'reasoning', 'encrypted_content': 'enc'}), + _Item({'type': 'message', 'content': 'ans'}), + ] + client = _Client(_Stream(events, _Final(output))) + adapter = OpenAIAdapter(client, model='gpt-5.6') + + out = list(adapter.stream('q')) + + assert Event('thinking', 'think') in out + assert Event('text', 'ans') in out + done = out[-1] + assert done.type == 'done' + assert done.items == [ + {'type': 'reasoning', 'encrypted_content': 'enc'}, + {'type': 'message', 'content': 'ans'}, + ] + + # history: user turn, then output items extended verbatim + assert adapter.history == [ + {'role': 'user', 'content': 'q'}, + {'type': 'reasoning', 'encrypted_content': 'enc'}, + {'type': 'message', 'content': 'ans'}, + ] + assert client.responses.seen['store'] is False + assert client.responses.seen['input'] == [{'role': 'user', 'content': 'q'}] From 4f1976264628e1b95b6541e05882390773a297fa Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 21:23:18 +0200 Subject: [PATCH 07/20] =?UTF-8?q?feat(llm):=20add=20drive=5Fadapter=20Even?= =?UTF-8?q?t=E2=86=92callback=20shim?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consumes an adapter's Event stream into the existing on_chunk/on_reasoning_chunk callbacks and returns (answer_text, opaque done.items). Not wired yet. --- packages/ai/src/ai/common/llm_adapter.py | 26 +++++++++++- .../tests/ai/common/test_llm_adapter_drive.py | 40 +++++++++++++++++++ 2 files changed, 65 insertions(+), 1 deletion(-) create mode 100644 packages/ai/tests/ai/common/test_llm_adapter_drive.py diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 6c60093a0..00ba4b7b5 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -10,7 +10,7 @@ """ from dataclasses import dataclass, field -from typing import Any, Iterator, Protocol, runtime_checkable +from typing import Any, Callable, Iterator, Optional, Protocol, runtime_checkable @dataclass @@ -36,6 +36,30 @@ class Adapter(Protocol): def stream(self, user_text: str) -> Iterator[Event]: ... +def drive_adapter( + adapter: Adapter, + user_text: str, + on_text: Optional[Callable[[str], None]] = None, + on_thinking: Optional[Callable[[str], None]] = None, +) -> tuple[str, list[Any]]: + """Consume an adapter's Event stream: fan text/thinking deltas to the sinks, + return the joined answer text and the terminal opaque ``done.items``. + """ + parts: list[str] = [] + items: list[Any] = [] + for ev in adapter.stream(user_text): + if ev.type == 'text': + parts.append(ev.text) + if on_text is not None: + on_text(ev.text) + elif ev.type == 'thinking': + if on_thinking is not None: + on_thinking(ev.text) + elif ev.type == 'done': + items = ev.items + return ''.join(parts), items + + def _make_think_tag_splitter(): """Split ``...`` CoT out of the content stream (Ollama, Perplexity). diff --git a/packages/ai/tests/ai/common/test_llm_adapter_drive.py b/packages/ai/tests/ai/common/test_llm_adapter_drive.py new file mode 100644 index 000000000..04fe1c5b3 --- /dev/null +++ b/packages/ai/tests/ai/common/test_llm_adapter_drive.py @@ -0,0 +1,40 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins drive_adapter: Event stream → callbacks + (answer_text, opaque done.items).""" + +from ai.common.llm_adapter import Event, drive_adapter + + +class _FakeAdapter: + def __init__(self, events): + self._events = events + self.history: list = [] + + def stream(self, user_text): + yield from self._events + + +def test_fans_deltas_and_returns_text_and_items(): + events = [ + Event('thinking', 'reasoning'), + Event('text', 'Hel'), + Event('text', 'lo'), + Event('done', items=[{'role': 'assistant', 'content': 'Hello'}]), + ] + texts, thinks = [], [] + answer, items = drive_adapter(_FakeAdapter(events), 'q', texts.append, thinks.append) + + assert answer == 'Hello' + assert items == [{'role': 'assistant', 'content': 'Hello'}] + assert texts == ['Hel', 'lo'] + assert thinks == ['reasoning'] + + +def test_sinks_optional(): + events = [Event('text', 'x'), Event('done', items=[])] + answer, items = drive_adapter(_FakeAdapter(events), 'q') + assert answer == 'x' + assert items == [] From 4faf8b326a426155c2c671ab1b55289d56ac8e3f Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Sun, 26 Jul 2026 23:50:24 +0200 Subject: [PATCH 08/20] feat(llm): drive the LangChain streaming path through the adapter chat_string now streams via LangChainAdapter + drive_adapter instead of the inline walker; stop sequences and finish_reason preserved. Native/Responses paths unchanged. --- packages/ai/src/ai/common/chat.py | 32 +++------ packages/ai/src/ai/common/llm_adapter.py | 9 ++- .../ai/common/test_chat_string_wiring.py | 72 +++++++++++++++++++ .../ai/common/test_llm_adapter_langchain.py | 17 ++++- 4 files changed, 101 insertions(+), 29 deletions(-) create mode 100644 packages/ai/tests/ai/common/test_chat_string_wiring.py diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index 750d7b174..a695f3f78 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -20,7 +20,7 @@ from ai.common.util import parseJson from ai.common.validation import validate_model_name, validate_max_tokens, validate_prompt from ai.common.llm_native_stream import STOP_SEQUENCES_VAR, dispatch_native_chat_stream -from ai.common.llm_adapter import _make_stream_content_parser +from ai.common.llm_adapter import LangChainAdapter, drive_adapter def _stop_kwargs() -> dict: @@ -539,30 +539,14 @@ def on_chunk_w(t): result = None if on_chunk_w is not None and _llm is not None and hasattr(_llm, 'stream'): try: - parts = [] - finish_reason: Optional[str] = None - parse = _make_stream_content_parser(on_reasoning_chunk_w is not None) - for piece in _llm.stream(prompt, **_stop_kwargs()): - text, thinking_delta = parse(piece.content) - if thinking_delta and on_reasoning_chunk_w is not None: - on_reasoning_chunk_w(thinking_delta) - if text: - on_chunk_w(text) - parts.append(text) - reason = (piece.response_metadata or {}).get('finish_reason') - if reason: - finish_reason = reason - # Drain chars buffered by the splitter (partial-tag tail). - tail_visible, tail_reasoning = parse.flush() - if tail_visible: - on_chunk_w(tail_visible) - parts.append(tail_visible) - if tail_reasoning and on_reasoning_chunk_w is not None: - on_reasoning_chunk_w(tail_reasoning) - if parts: - result = ''.join(parts) + # Stream the LangChain path through the normalized adapter; drive_adapter + # fans text/thinking to the callbacks and returns the joined answer. + adapter = LangChainAdapter(_llm, stream_kwargs=_stop_kwargs()) + answer, _items = drive_adapter(adapter, prompt, on_chunk_w, on_reasoning_chunk_w) + if answer: + result = answer if on_finish is not None: - on_finish(finish_reason) + on_finish(adapter.finish_reason) except Exception as e: warning( f'Streaming disabled for model={self._model} ' diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 00ba4b7b5..2aea4548b 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -165,21 +165,26 @@ class LangChainAdapter: ``done.items`` is the assistant text turn — LangChain carries no opaque reasoning state. """ - def __init__(self, llm: Any, history: list[Any] | None = None): + def __init__(self, llm: Any, history: list[Any] | None = None, stream_kwargs: dict | None = None): self.llm = llm self.history: list[Any] = history if history is not None else [] + self.stream_kwargs = stream_kwargs or {} + self.finish_reason: Optional[str] = None def stream(self, user_text: str) -> Iterator[Event]: self.history.append({'role': 'user', 'content': user_text}) parse = _make_stream_content_parser(True) parts: list[str] = [] - for piece in self.llm.stream(self.history): + for piece in self.llm.stream(self.history, **self.stream_kwargs): text, thinking = parse(piece.content) if thinking: yield Event('thinking', thinking) if text: parts.append(text) yield Event('text', text) + reason = (getattr(piece, 'response_metadata', None) or {}).get('finish_reason') + if reason: + self.finish_reason = reason tail_text, tail_thinking = parse.flush() if tail_thinking: yield Event('thinking', tail_thinking) diff --git a/packages/ai/tests/ai/common/test_chat_string_wiring.py b/packages/ai/tests/ai/common/test_chat_string_wiring.py new file mode 100644 index 000000000..7ae8a73e4 --- /dev/null +++ b/packages/ai/tests/ai/common/test_chat_string_wiring.py @@ -0,0 +1,72 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Integration: chat_string streams the LangChain path through the normalized adapter.""" + +from ai.common.chat import ChatBase + + +class _Piece: + def __init__(self, content, response_metadata=None): + self.content = content + self.response_metadata = response_metadata + + +class _FakeLLM: + def __init__(self, pieces): + self._pieces = pieces + self.kwargs = None + + def stream(self, messages, **kwargs): + self.kwargs = kwargs + yield from self._pieces + + +class _Chat(ChatBase): + # Bypass the config-driven __init__; wire only what chat_string reads. + def __init__(self, llm): + self._llm = llm + self._model = 'test-model' + self._modelTotalTokens = 100000 + self._modelOutputTokens = 4096 + self._is_reasoning = False + self._raw_client = None + self._native_stream_provider = '' + self._raw_openai_client = None + + def getTokens(self, value): + return len(value) + + def _ensure_openai_compat_reasoning_stream(self): + pass + + +def test_streams_text_and_reasoning_via_adapter(): + llm = _FakeLLM( + [ + _Piece([{'type': 'thinking', 'thinking': 'cot'}, {'type': 'text', 'text': 'the answer'}]), + ] + ) + chat = _Chat(llm) + chunks, thinks, finishes = [], [], [] + + result = chat.chat_string('hi', on_chunk=chunks.append, on_finish=finishes.append, on_reasoning_chunk=thinks.append) + + assert result == 'the answer' + assert ''.join(chunks) == 'the answer' + assert ''.join(thinks) == 'cot' + + +def test_stop_sequences_reach_the_model(): + from ai.common.llm_native_stream import STOP_SEQUENCES_VAR + + llm = _FakeLLM([_Piece('plain answer here')]) + chat = _Chat(llm) + token = STOP_SEQUENCES_VAR.set(['\nObservation:']) + try: + chat.chat_string('hi', on_chunk=lambda t: None) + finally: + STOP_SEQUENCES_VAR.reset(token) + assert llm.kwargs == {'stop': ['\nObservation:']} diff --git a/packages/ai/tests/ai/common/test_llm_adapter_langchain.py b/packages/ai/tests/ai/common/test_llm_adapter_langchain.py index f58a76528..6bbd07006 100644 --- a/packages/ai/tests/ai/common/test_llm_adapter_langchain.py +++ b/packages/ai/tests/ai/common/test_llm_adapter_langchain.py @@ -9,19 +9,22 @@ class _Piece: - def __init__(self, content): + def __init__(self, content, response_metadata=None): self.content = content + self.response_metadata = response_metadata class _FakeLLM: def __init__(self, pieces): self._pieces = pieces self.seen = None + self.kwargs = None - def stream(self, messages): + def stream(self, messages, **kwargs): self.seen = list(messages) + self.kwargs = kwargs for p in self._pieces: - yield _Piece(p) + yield p if isinstance(p, _Piece) else _Piece(p) def test_normalizes_blocks_and_records_history(): @@ -51,3 +54,11 @@ def test_str_content_flushes_buffered_tail(): adapter = LangChainAdapter(llm) text = ''.join(e.text for e in adapter.stream('q') if e.type == 'text') assert text == 'hello world' + + +def test_passes_stream_kwargs_and_tracks_finish_reason(): + llm = _FakeLLM([_Piece([{'type': 'text', 'text': 'done'}], response_metadata={'finish_reason': 'stop'})]) + adapter = LangChainAdapter(llm, stream_kwargs={'stop': ['\nObservation:']}) + list(adapter.stream('q')) + assert llm.kwargs == {'stop': ['\nObservation:']} + assert adapter.finish_reason == 'stop' From 9396a331c464fa48f03e9710a94e23d7bb5567ea Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Mon, 27 Jul 2026 08:36:54 +0200 Subject: [PATCH 09/20] fix(llm): flatten non-streaming _chat content to a string _chat returned Anthropic's typed-block list verbatim, crashing agents/expectJson on .strip(); flatten_content keeps text blocks, drops thinking. Fixes the #1658 class. --- packages/ai/src/ai/common/chat.py | 6 +++--- packages/ai/src/ai/common/llm_adapter.py | 11 +++++++++++ packages/ai/tests/ai/common/test_chat_content.py | 12 +++++++++++- 3 files changed, 25 insertions(+), 4 deletions(-) diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index a695f3f78..ab9a0dd02 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -20,7 +20,7 @@ from ai.common.util import parseJson from ai.common.validation import validate_model_name, validate_max_tokens, validate_prompt from ai.common.llm_native_stream import STOP_SEQUENCES_VAR, dispatch_native_chat_stream -from ai.common.llm_adapter import LangChainAdapter, drive_adapter +from ai.common.llm_adapter import LangChainAdapter, drive_adapter, flatten_content def _stop_kwargs() -> dict: @@ -188,8 +188,8 @@ def _chat(self, prompt: str) -> str: # so non-agent callers (and backends/mocks without a stop param) are unaffected. results = self._llm.invoke(prompt, **_stop_kwargs()) - # Return the results - return results.content + # Flatten provider content (Anthropic returns typed blocks) so callers get a string. + return flatten_content(results.content) def getTokens(self, value: str) -> int: """ diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 2aea4548b..feea5c1de 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -159,6 +159,17 @@ def feed(content): return feed +def flatten_content(content: Any) -> str: + """Collapse a full (non-streamed) message content to visible text, dropping thinking. + + Non-streaming callers (agents, expectJson) must get a string, never a block list. + """ + parse = _make_stream_content_parser(False) + text, _thinking = parse(content) + tail, _ = parse.flush() + return text + tail + + class LangChainAdapter: """Wraps a LangChain chat model so non-reasoning providers speak the Event contract. diff --git a/packages/ai/tests/ai/common/test_chat_content.py b/packages/ai/tests/ai/common/test_chat_content.py index c234ff575..c5f929d2a 100644 --- a/packages/ai/tests/ai/common/test_chat_content.py +++ b/packages/ai/tests/ai/common/test_chat_content.py @@ -7,7 +7,7 @@ that the LLM adapters will absorb. Behavior must not drift during that refactor. """ -from ai.common.llm_adapter import _make_stream_content_parser +from ai.common.llm_adapter import _make_stream_content_parser, flatten_content def test_str_passthrough(): @@ -60,6 +60,16 @@ def test_signature_note_suppressed_without_reasoning_sink(): assert thinking == '' +def test_flatten_str_passthrough(): + assert flatten_content('plain answer') == 'plain answer' + + +def test_flatten_anthropic_blocks_drops_thinking(): + # the original crash: a typed-block list must become a plain string. + content = [{'type': 'thinking', 'thinking': 'cot'}, {'type': 'text', 'text': 'answer'}] + assert flatten_content(content) == 'answer' + + def test_inline_think_split_across_feed_and_flush(): parse = _make_stream_content_parser(True) text, thinking = parse('beforecotafter') From 1cb146528ebfeb496867aeb43b5435b15a4cd337 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Mon, 27 Jul 2026 16:47:01 +0200 Subject: [PATCH 10/20] feat(llm): route the native Anthropic path through an adapter NativeAnthropicAdapter yields Events over the same payload + create(stream=True) path; try_anthropic_native_chat_stream now drives it. Behavior preserved. --- .../ai/src/ai/common/llm_native_stream.py | 123 +++++++++--------- .../common/test_native_anthropic_adapter.py | 86 ++++++++++++ 2 files changed, 150 insertions(+), 59 deletions(-) create mode 100644 packages/ai/tests/ai/common/test_native_anthropic_adapter.py diff --git a/packages/ai/src/ai/common/llm_native_stream.py b/packages/ai/src/ai/common/llm_native_stream.py index 85c6a3c96..34cf2eacf 100644 --- a/packages/ai/src/ai/common/llm_native_stream.py +++ b/packages/ai/src/ai/common/llm_native_stream.py @@ -19,6 +19,8 @@ from rocketlib import debug, warning +from ai.common.llm_adapter import Event, drive_adapter + # Per-call carrier for API-level stop sequences (e.g. CrewAI's ReAct "\nObservation:"). # Set on the ask path in llm_base._question and read at every model sink so the stop # reaches the provider API instead of relying only on post-hoc text truncation. A @@ -167,67 +169,65 @@ def anthropic_extended_thinking_active(chat: Any) -> bool: return bool(mk.get('thinking')) -def _stream_anthropic_messages_api( - chat: Any, - prompt: str, - on_chunk: Callable[[str], None], - on_finish: Optional[Callable[[Optional[str]], None]], - on_reasoning_chunk: Optional[Callable[[str], None]], -) -> str: - llm = chat._llm - # INVARIANT: read synchronously within the ask() call, while LLMBase._question still - # holds the contextvar (before its finally reset). The stream is consumed here, not - # deferred to the caller, so this never runs after the reset (which would send None). - payload: dict[str, Any] = dict(llm._get_request_payload(prompt, stop=STOP_SEQUENCES_VAR.get() or None, stream=True)) - _raw_client = getattr(llm, '_client', None) - client = _raw_client() if callable(_raw_client) else _raw_client - if client is None: - raise RuntimeError('ChatAnthropic has no _client for native streaming') - - parts: list[str] = [] - finish_reason: Optional[str] = None - reasoning_deltas = 0 - raw_stream = _open_raw_message_stream(client, payload) - - try: - for event in raw_stream: - et = _event_type_name(event) - if et == 'content_block_delta': - delta = getattr(event, 'delta', None) - if delta is None: - continue - dt = _delta_type_name(delta) - if dt == 'thinking_delta' and on_reasoning_chunk is not None: - piece = getattr(delta, 'thinking', None) or '' - if piece: - on_reasoning_chunk(piece) - reasoning_deltas += 1 - elif dt == 'text_delta' and on_chunk is not None: - piece = getattr(delta, 'text', None) or '' - if piece: - on_chunk(piece) - parts.append(piece) - elif et == 'message_delta': - md = getattr(event, 'delta', None) - if md is not None: - sr = getattr(md, 'stop_reason', None) - if sr is not None: - finish_reason = _map_claude_stop_reason(sr) - finally: - closer = getattr(raw_stream, 'close', None) - if callable(closer): - try: - closer() - except Exception: - pass +class NativeAnthropicAdapter: + """Bridges the native Anthropic Messages stream to the Event contract. - if not parts: - raise RuntimeError('Anthropic SDK stream produced no text') + Same payload + raw create(stream=True) path as before; now yields normalized Events. + """ - if on_finish is not None: - on_finish(finish_reason) + def __init__(self, chat: Any, history: Optional[list] = None): + self.chat = chat + self.history = history if history is not None else [] + self.finish_reason: Optional[str] = None - return ''.join(parts) + def stream(self, user_text: str): + # INVARIANT: consume synchronously within ask() while STOP_SEQUENCES_VAR is still set. + self.history.append({'role': 'user', 'content': user_text}) + llm = self.chat._llm + payload: dict[str, Any] = dict( + llm._get_request_payload(user_text, stop=STOP_SEQUENCES_VAR.get() or None, stream=True) + ) + _raw_client = getattr(llm, '_client', None) + client = _raw_client() if callable(_raw_client) else _raw_client + if client is None: + raise RuntimeError('ChatAnthropic has no _client for native streaming') + + parts: list[str] = [] + raw_stream = _open_raw_message_stream(client, payload) + try: + for event in raw_stream: + et = _event_type_name(event) + if et == 'content_block_delta': + delta = getattr(event, 'delta', None) + if delta is None: + continue + dt = _delta_type_name(delta) + if dt == 'thinking_delta': + piece = getattr(delta, 'thinking', None) or '' + if piece: + yield Event('thinking', piece) + elif dt == 'text_delta': + piece = getattr(delta, 'text', None) or '' + if piece: + parts.append(piece) + yield Event('text', piece) + elif et == 'message_delta': + md = getattr(event, 'delta', None) + if md is not None: + sr = getattr(md, 'stop_reason', None) + if sr is not None: + self.finish_reason = _map_claude_stop_reason(sr) + finally: + closer = getattr(raw_stream, 'close', None) + if callable(closer): + try: + closer() + except Exception: + pass + + assistant = {'role': 'assistant', 'content': ''.join(parts)} + self.history.append(assistant) + yield Event('done', items=[assistant]) def try_anthropic_native_chat_stream( @@ -242,7 +242,12 @@ def try_anthropic_native_chat_stream( return None try: - text = _stream_anthropic_messages_api(chat, prompt, on_chunk, on_finish, on_reasoning_chunk) + adapter = NativeAnthropicAdapter(chat) + text, _items = drive_adapter(adapter, prompt, on_chunk, on_reasoning_chunk) + if not text: + raise RuntimeError('Anthropic SDK stream produced no text') + if on_finish is not None: + on_finish(adapter.finish_reason) return text except Exception as e: warning( diff --git a/packages/ai/tests/ai/common/test_native_anthropic_adapter.py b/packages/ai/tests/ai/common/test_native_anthropic_adapter.py new file mode 100644 index 000000000..9f658594a --- /dev/null +++ b/packages/ai/tests/ai/common/test_native_anthropic_adapter.py @@ -0,0 +1,86 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins NativeAnthropicAdapter: native Messages stream → Events, same payload path.""" + +from ai.common.llm_adapter import Event +from ai.common.llm_native_stream import NativeAnthropicAdapter, try_anthropic_native_chat_stream + + +class _Delta: + def __init__(self, type, thinking='', text='', stop_reason=None): + self.type = type + self.thinking = thinking + self.text = text + self.stop_reason = stop_reason + + +class _RawEvent: + def __init__(self, type, delta=None): + self.type = type + self.delta = delta + + +class _Messages: + def __init__(self, events): + self._events = events + self.seen = None + + def create(self, **kwargs): + self.seen = kwargs + return iter(self._events) + + +class _Client: + def __init__(self, events): + self.messages = _Messages(events) + + +class _LLM: + def __init__(self, payload, client): + self._payload = payload + self._client = client + + def _get_request_payload(self, prompt, stop=None, stream=False): + return {**self._payload, 'messages': [{'role': 'user', 'content': prompt}]} + + +class _Chat: + def __init__(self, llm): + self._llm = llm + self._extended_thinking = True + + +def _chat_with(events, payload=None): + return _Chat(_LLM(payload or {'model': 'claude-sonnet-4-6', 'max_tokens': 1000}, _Client(events))) + + +def test_native_adapter_yields_events_and_maps_finish(): + events = [ + _RawEvent('content_block_delta', _Delta('thinking_delta', thinking='cot')), + _RawEvent('content_block_delta', _Delta('text_delta', text='answer')), + _RawEvent('message_delta', _Delta('message_delta', stop_reason='end_turn')), + ] + adapter = NativeAnthropicAdapter(_chat_with(events)) + out = list(adapter.stream('q')) + + assert Event('thinking', 'cot') in out + assert Event('text', 'answer') in out + assert out[-1].type == 'done' + assert out[-1].items == [{'role': 'assistant', 'content': 'answer'}] + assert adapter.finish_reason == 'stop' + + +def test_handler_drives_adapter_and_reports_finish(): + events = [ + _RawEvent('content_block_delta', _Delta('text_delta', text='hi')), + _RawEvent('message_delta', _Delta('message_delta', stop_reason='max_tokens')), + ] + finishes = [] + text = try_anthropic_native_chat_stream( + _chat_with(events), 'q', on_chunk=lambda t: None, on_finish=finishes.append, on_reasoning_chunk=lambda t: None + ) + assert text == 'hi' + assert finishes == ['length'] # max_tokens → length From 72039585bb8140d2e1d2ff5f6141ba9d63fc92a4 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Mon, 27 Jul 2026 17:01:14 +0200 Subject: [PATCH 11/20] feat(llm): route the OpenAI Responses path through an adapter NativeOpenAIResponsesAdapter yields Events over responses.create(stream=True); _chat_string_responses drives it, keeping the non-streaming fallback. Behavior preserved. --- packages/ai/src/ai/common/chat.py | 67 ++++------------- packages/ai/src/ai/common/llm_adapter.py | 45 +++++++++++ .../test_native_openai_responses_adapter.py | 74 +++++++++++++++++++ 3 files changed, 134 insertions(+), 52 deletions(-) create mode 100644 packages/ai/tests/ai/common/test_native_openai_responses_adapter.py diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index ab9a0dd02..6cebaf34e 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -20,7 +20,7 @@ from ai.common.util import parseJson from ai.common.validation import validate_model_name, validate_max_tokens, validate_prompt from ai.common.llm_native_stream import STOP_SEQUENCES_VAR, dispatch_native_chat_stream -from ai.common.llm_adapter import LangChainAdapter, drive_adapter, flatten_content +from ai.common.llm_adapter import LangChainAdapter, NativeOpenAIResponsesAdapter, drive_adapter, flatten_content def _stop_kwargs() -> dict: @@ -400,64 +400,27 @@ def _chat_string_responses( falling back to non-streaming invoke() only if nothing reached the UI yet. """ prompt = validate_prompt(prompt, self._modelTotalTokens, self.getTokens) - - text_parts: list = [] - finish_reason: Optional[str] = None try: - stream = self._raw_client.responses.create( - model=self._model, - input=prompt, - reasoning={'summary': 'auto'}, - max_output_tokens=self._modelOutputTokens, - stream=True, - ) - for event in stream: - etype = getattr(event, 'type', '') or '' - if etype == 'response.reasoning_summary_text.delta': - delta = getattr(event, 'delta', '') or '' - if delta and on_reasoning_chunk is not None: - on_reasoning_chunk(delta) - elif etype == 'response.output_text.delta': - delta = getattr(event, 'delta', '') or '' - if delta: - text_parts.append(delta) - if on_chunk is not None: - on_chunk(delta) - elif etype == 'response.completed': - resp = getattr(event, 'response', None) - if resp is not None: - status = getattr(resp, 'status', None) - if status == 'completed': - finish_reason = 'stop' - elif status == 'incomplete': - details = getattr(resp, 'incomplete_details', None) - reason = getattr(details, 'reason', None) if details else None - finish_reason = reason or 'length' - else: - finish_reason = status or 'stop' - elif etype in ('response.failed', 'response.error'): - finish_reason = 'error' + adapter = NativeOpenAIResponsesAdapter(self) + text, _items = drive_adapter(adapter, prompt, on_chunk, on_reasoning_chunk) + if on_finish is not None: + on_finish(adapter.finish_reason) + return text except Exception as e: warning(f'Reasoning streaming disabled for model={self._model} ({type(e).__name__}): {e}.') - # Only retry non-streaming if nothing has reached the UI; otherwise - # the full fallback would arrive on top of the partial we already streamed. + # Only retry non-streaming if nothing reached the UI; otherwise the full + # fallback would arrive on top of the partial we already streamed. if emitted is None or not emitted['any']: results = self._llm.invoke(prompt, **_stop_kwargs()) - content = getattr(results, 'content', '') or '' - content_text = content if isinstance(content, str) else str(content) - text_parts = [content_text] - # Push the fallback answer through on_chunk so the open UI bubble - # gets the visible text (the caller dedupes the final pipeline result). + content_text = flatten_content(getattr(results, 'content', '')) if content_text and on_chunk is not None: on_chunk(content_text) - finish_reason = 'stop' - else: - finish_reason = 'error' - - if on_finish is not None: - on_finish(finish_reason) - - return ''.join(text_parts) + if on_finish is not None: + on_finish('stop') + return content_text + if on_finish is not None: + on_finish('error') + return '' def chat_string( self, diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index feea5c1de..190633518 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -275,3 +275,48 @@ def stream(self, user_text: str) -> Iterator[Event]: items = [item.model_dump() for item in final.output] self.history.extend(items) yield Event('done', items=items) + + +class NativeOpenAIResponsesAdapter: + """Bridges the OpenAI Responses reasoning stream (create/stream=True) to Events.""" + + def __init__(self, chat: Any, history: list[Any] | None = None): + self.chat = chat + self.history: list[Any] = history if history is not None else [] + self.finish_reason: Optional[str] = None + + def stream(self, user_text: str) -> Iterator[Event]: + self.history.append({'role': 'user', 'content': user_text}) + chat = self.chat + parts: list[str] = [] + stream = chat._raw_client.responses.create( + model=chat._model, + input=user_text, + reasoning={'summary': 'auto'}, + max_output_tokens=chat._modelOutputTokens, + stream=True, + ) + for event in stream: + etype = getattr(event, 'type', '') or '' + if etype == 'response.reasoning_summary_text.delta': + delta = getattr(event, 'delta', '') or '' + if delta: + yield Event('thinking', delta) + elif etype == 'response.output_text.delta': + delta = getattr(event, 'delta', '') or '' + if delta: + parts.append(delta) + yield Event('text', delta) + elif etype == 'response.completed': + resp = getattr(event, 'response', None) + status = getattr(resp, 'status', None) if resp is not None else None + if status == 'incomplete': + details = getattr(resp, 'incomplete_details', None) + self.finish_reason = (getattr(details, 'reason', None) if details else None) or 'length' + else: + self.finish_reason = 'stop' if status == 'completed' else (status or 'stop') + elif etype in ('response.failed', 'response.error'): + self.finish_reason = 'error' + assistant = {'role': 'assistant', 'content': ''.join(parts)} + self.history.append(assistant) + yield Event('done', items=[assistant]) diff --git a/packages/ai/tests/ai/common/test_native_openai_responses_adapter.py b/packages/ai/tests/ai/common/test_native_openai_responses_adapter.py new file mode 100644 index 000000000..71bcae5cd --- /dev/null +++ b/packages/ai/tests/ai/common/test_native_openai_responses_adapter.py @@ -0,0 +1,74 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins NativeOpenAIResponsesAdapter: Responses stream → Events + finish mapping.""" + +from ai.common.llm_adapter import Event, NativeOpenAIResponsesAdapter + + +class _Ev: + def __init__(self, type, delta='', response=None): + self.type = type + self.delta = delta + self.response = response + + +class _Resp: + def __init__(self, status, incomplete_details=None): + self.status = status + self.incomplete_details = incomplete_details + + +class _Details: + def __init__(self, reason): + self.reason = reason + + +class _Responses: + def __init__(self, events): + self._events = events + self.seen = None + + def create(self, **kwargs): + self.seen = kwargs + return iter(self._events) + + +class _RawClient: + def __init__(self, events): + self.responses = _Responses(events) + + +class _Chat: + def __init__(self, events): + self._raw_client = _RawClient(events) + self._model = 'gpt-5.6' + self._modelOutputTokens = 4096 + + +def test_yields_events_and_completed_maps_to_stop(): + events = [ + _Ev('response.reasoning_summary_text.delta', delta='think'), + _Ev('response.output_text.delta', delta='ans'), + _Ev('response.completed', response=_Resp('completed')), + ] + adapter = NativeOpenAIResponsesAdapter(_Chat(events)) + out = list(adapter.stream('q')) + + assert Event('thinking', 'think') in out + assert Event('text', 'ans') in out + assert out[-1].type == 'done' + assert out[-1].items == [{'role': 'assistant', 'content': 'ans'}] + assert adapter.finish_reason == 'stop' + + +def test_incomplete_maps_to_reason(): + events = [ + _Ev('response.output_text.delta', delta='x'), + _Ev('response.completed', response=_Resp('incomplete', _Details('max_output_tokens'))), + ] + adapter = NativeOpenAIResponsesAdapter(_Chat(events)) + list(adapter.stream('q')) + assert adapter.finish_reason == 'max_output_tokens' From fba74bb5caeaccca46ec7fab47c5074e86b54aab Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Tue, 28 Jul 2026 10:40:13 +0200 Subject: [PATCH 12/20] refactor(llm): unify the non-streaming path through the adapter _chat now drains the adapter (same iterator/normalization as streaming) instead of invoke+flatten, so content handling is one mechanism. Fallback kept for stream-less backends. --- packages/ai/src/ai/common/chat.py | 15 +++++++++------ .../ai/tests/ai/common/test_chat_string_wiring.py | 7 +++++++ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index 6cebaf34e..528ae1ab1 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -184,12 +184,15 @@ def _chat(self, prompt: str) -> str: Should raise appropriate exceptions for API failures, authentication errors, or other provider-specific issues """ - # Ask the LLM. The stop kwarg is only added when the agent set stop sequences, - # so non-agent callers (and backends/mocks without a stop param) are unaffected. - results = self._llm.invoke(prompt, **_stop_kwargs()) - - # Flatten provider content (Anthropic returns typed blocks) so callers get a string. - return flatten_content(results.content) + # Non-streaming: drain the adapter with no display sinks — same iterator and + # normalization as streaming, so item capture is structural on every path. + _llm = self._llm + if hasattr(_llm, 'stream'): + adapter = LangChainAdapter(_llm, stream_kwargs=_stop_kwargs()) + text, _items = drive_adapter(adapter, prompt) + return text + # Backends/mocks without .stream: invoke and normalize the content to a string. + return flatten_content(_llm.invoke(prompt, **_stop_kwargs()).content) def getTokens(self, value: str) -> int: """ diff --git a/packages/ai/tests/ai/common/test_chat_string_wiring.py b/packages/ai/tests/ai/common/test_chat_string_wiring.py index 7ae8a73e4..b1ab27105 100644 --- a/packages/ai/tests/ai/common/test_chat_string_wiring.py +++ b/packages/ai/tests/ai/common/test_chat_string_wiring.py @@ -70,3 +70,10 @@ def test_stop_sequences_reach_the_model(): finally: STOP_SEQUENCES_VAR.reset(token) assert llm.kwargs == {'stop': ['\nObservation:']} + + +def test_chat_nonstreaming_drains_adapter(): + # The unify: _chat (no display callbacks) drains the same adapter and returns text. + llm = _FakeLLM([_Piece([{'type': 'thinking', 'thinking': 'x'}, {'type': 'text', 'text': 'plain answer'}])]) + chat = _Chat(llm) + assert chat._chat('q') == 'plain answer' From d0cd81310f9f8d3f7f321e79cf2a056d2c429db8 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Tue, 28 Jul 2026 11:15:03 +0200 Subject: [PATCH 13/20] feat(llm_anthropic): per-node extended-thinking control with context gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the interim hard-off (#1681) with an opt-in `extendedThinking` node toggle (default off). Thinking is no longer baked into the client; the native adapter adds it per call, so it rides the interactive streaming path only — agent/expectJson never gets it. --- nodes/src/nodes/llm_anthropic/anthropic.py | 36 ++++------ nodes/src/nodes/llm_anthropic/services.json | 66 +++++++++++++------ .../ai/src/ai/common/llm_native_stream.py | 3 + .../common/test_native_anthropic_adapter.py | 20 +++++- 4 files changed, 82 insertions(+), 43 deletions(-) diff --git a/nodes/src/nodes/llm_anthropic/anthropic.py b/nodes/src/nodes/llm_anthropic/anthropic.py index 4851c98e2..ccb604192 100644 --- a/nodes/src/nodes/llm_anthropic/anthropic.py +++ b/nodes/src/nodes/llm_anthropic/anthropic.py @@ -36,14 +36,6 @@ from langchain_anthropic import ChatAnthropic -# Interim: extended thinking is disabled at the node until the normalized provider adapter -# lands (RFC #1679). With thinking on, Anthropic returns typed content blocks (thinking + text) -# that the agent / expectJson path cannot consume (incident #1658; flatten defense-in-depth in -# #1659). We gate activation here rather than flipping capabilities.reasoning, because that flag -# is stamped by tools/sync_models and re-stamps to true on the next sync. Flip back to re-enable. -_EXTENDED_THINKING_ENABLED = False - - def _estimate_token_ids(text: str) -> list: """Estimate token ids at ~4 chars/token.""" return [0] * max(1, (len(text) + 3) // 4) @@ -78,23 +70,23 @@ def __init__(self, provider: str, connConfig: Dict[str, Any], bag: Dict[str, Any # Init the chat base super().__init__(provider, connConfig, bag) - # Get the LLM - kwargs: Dict[str, Any] = { - 'model': model, - 'api_key': apikey, - 'max_tokens': self._modelOutputTokens, - 'custom_get_token_ids': _estimate_token_ids, - } - if self._is_reasoning and _EXTENDED_THINKING_ENABLED: - kwargs.update(build_anthropic_thinking_kwargs(model_gate, self._modelOutputTokens)) - - self._extended_thinking = bool(kwargs.get('thinking')) - # Only route through the native handler when thinking is actually on; - # non-reasoning models stay on the default LangChain path. + # Extended thinking is opt-in per node (config extendedThinking, default off) and only + # for reasoning models. It is NOT baked into the client: the native streaming adapter adds + # it per call, so thinking activates on the interactive streaming path only — never on the + # agent / expectJson path (which has no streaming and stays on the plain LangChain client). + self._thinking_mode_kwargs: Dict[str, Any] = {} + if self._is_reasoning and bool(config.get('extendedThinking')): + self._thinking_mode_kwargs = build_anthropic_thinking_kwargs(model_gate, self._modelOutputTokens) + self._extended_thinking = bool(self._thinking_mode_kwargs) if self._extended_thinking: self._native_stream_provider = 'anthropic' - self._llm = ChatAnthropic(**kwargs) + self._llm = ChatAnthropic( + model=model, + api_key=apikey, + max_tokens=self._modelOutputTokens, + custom_get_token_ids=_estimate_token_ids, + ) # Save our chat class into the bag bag['chat'] = self diff --git a/nodes/src/nodes/llm_anthropic/services.json b/nodes/src/nodes/llm_anthropic/services.json index 78a338127..b7db2c2d7 100644 --- a/nodes/src/nodes/llm_anthropic/services.json +++ b/nodes/src/nodes/llm_anthropic/services.json @@ -332,48 +332,60 @@ "title": "Tokens", "description": "Total Tokens" }, + "extendedThinking": { + "type": "boolean", + "title": "Extended thinking", + "default": false, + "description": "Enable Anthropic extended thinking (reasoning) for this node. Off by default. Only affects reasoning-capable models, and only on the interactive streaming path — agent / expectJson calls never use thinking." + }, "anthropic.custom": { "object": "custom", "properties": [ "model", "modelTotalTokens", "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-sonnet-4-6": { "object": "claude-sonnet-4-6", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-4-6": { "object": "claude-opus-4-6", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-haiku-4-5": { "object": "claude-haiku-4-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-sonnet-4-5": { "object": "claude-sonnet-4-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-4-5": { "object": "claude-opus-4-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.profile": { @@ -523,98 +535,112 @@ "object": "claude-opus-4-7", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-3-haiku": { "object": "claude-3-haiku", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-fable-5": { "object": "claude-fable-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-fable-latest": { "object": "claude-fable-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-haiku-latest": { "object": "claude-haiku-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-4": { "object": "claude-opus-4", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-4-1": { "object": "claude-opus-4-1", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-4-7-fast": { "object": "claude-opus-4-7-fast", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-4-8": { "object": "claude-opus-4-8", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-4-8-fast": { "object": "claude-opus-4-8-fast", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-latest": { "object": "claude-opus-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-sonnet-4": { "object": "claude-sonnet-4", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-sonnet-5": { "object": "claude-sonnet-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-sonnet-latest": { "object": "claude-sonnet-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource" + "llm.cloud.modelSource", + "extendedThinking" ] }, "anthropic.claude-opus-5": { diff --git a/packages/ai/src/ai/common/llm_native_stream.py b/packages/ai/src/ai/common/llm_native_stream.py index 34cf2eacf..0d9823c54 100644 --- a/packages/ai/src/ai/common/llm_native_stream.py +++ b/packages/ai/src/ai/common/llm_native_stream.py @@ -187,6 +187,9 @@ def stream(self, user_text: str): payload: dict[str, Any] = dict( llm._get_request_payload(user_text, stop=STOP_SEQUENCES_VAR.get() or None, stream=True) ) + # Thinking is added per call (not baked into the client) so it rides only this native + # streaming path — the agent / expectJson path never gets it. + payload.update(getattr(self.chat, '_thinking_mode_kwargs', None) or {}) _raw_client = getattr(llm, '_client', None) client = _raw_client() if callable(_raw_client) else _raw_client if client is None: diff --git a/packages/ai/tests/ai/common/test_native_anthropic_adapter.py b/packages/ai/tests/ai/common/test_native_anthropic_adapter.py index 9f658594a..eecf8adc1 100644 --- a/packages/ai/tests/ai/common/test_native_anthropic_adapter.py +++ b/packages/ai/tests/ai/common/test_native_anthropic_adapter.py @@ -48,9 +48,10 @@ def _get_request_payload(self, prompt, stop=None, stream=False): class _Chat: - def __init__(self, llm): + def __init__(self, llm, thinking_mode=None): self._llm = llm self._extended_thinking = True + self._thinking_mode_kwargs = thinking_mode or {} def _chat_with(events, payload=None): @@ -84,3 +85,20 @@ def test_handler_drives_adapter_and_reports_finish(): ) assert text == 'hi' assert finishes == ['length'] # max_tokens → length + + +def test_thinking_mode_injected_into_payload_per_call(): + client = _Client([_RawEvent('content_block_delta', _Delta('text_delta', text='ok'))]) + chat = _Chat( + _LLM({'model': 'claude-sonnet-4-6', 'max_tokens': 1000}, client), + thinking_mode={'thinking': {'type': 'adaptive', 'display': 'summarized'}}, + ) + list(NativeAnthropicAdapter(chat).stream('q')) + assert client.messages.seen['thinking'] == {'type': 'adaptive', 'display': 'summarized'} + + +def test_no_thinking_in_payload_when_mode_empty(): + client = _Client([_RawEvent('content_block_delta', _Delta('text_delta', text='ok'))]) + chat = _Chat(_LLM({'model': 'm', 'max_tokens': 10}, client)) # no thinking mode + list(NativeAnthropicAdapter(chat).stream('q')) + assert 'thinking' not in client.messages.seen From 5be6e36da9ea07d907c8f938da4e54b86d78c1fd Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Tue, 28 Jul 2026 13:47:58 +0200 Subject: [PATCH 14/20] test(agent_crewai): drive NativeAnthropicAdapter for native stop-threading _stream_anthropic_messages_api folded into NativeAnthropicAdapter on this branch; update the stop-threading test to drive the adapter. Same payload wiring, no behavior change. --- nodes/test/agent_crewai/test_stop_words.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/nodes/test/agent_crewai/test_stop_words.py b/nodes/test/agent_crewai/test_stop_words.py index 9af1e5c80..0fc151d5e 100644 --- a/nodes/test/agent_crewai/test_stop_words.py +++ b/nodes/test/agent_crewai/test_stop_words.py @@ -239,7 +239,7 @@ def _load_native_stream(): def _run_native_capture(ns, stop_value): - """Drive _stream_anthropic_messages_api with fakes; return the stop the payload got.""" + """Drive the native Anthropic adapter with fakes; return the stop the payload got.""" captured: dict = {} class _FakeLLM: @@ -255,11 +255,8 @@ class _FakeChat: ns._open_raw_message_stream = lambda client, payload: iter(()) # no events -> no text token = ns.STOP_SEQUENCES_VAR.set(stop_value) try: - # Produces no text, so the function raises after building the payload — we only - # assert the payload wiring, which happens before any streaming. - ns._stream_anthropic_messages_api(_FakeChat(), 'prompt', lambda t: None, None, None) - except RuntimeError: - pass + # Drain the adapter; the payload (with the stop) is built before any streaming. + list(ns.NativeAnthropicAdapter(_FakeChat()).stream('prompt')) finally: ns.STOP_SEQUENCES_VAR.reset(token) return captured.get('stop', 'UNSET') From afce1ff3b9ee83d25a06b64351f94891208c6ce0 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Tue, 28 Jul 2026 18:18:11 +0200 Subject: [PATCH 15/20] fix(sync-models): make the extendedThinking toggle capabilities-aware MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The model sync now adds the extendedThinking UI toggle to reasoning-capable profiles (incl. models added on future syncs) and removes it from non-reasoning ones — the UI stays self-maintaining. Also drops the inert toggle from claude-3-haiku (non-reasoning). --- nodes/src/nodes/llm_anthropic/services.json | 3 +- tools/sync_models/src/core/patcher.py | 32 ++++++++--- tools/sync_models/src/providers/base.py | 2 +- .../test/test_extended_thinking_field.py | 57 +++++++++++++++++++ 4 files changed, 83 insertions(+), 11 deletions(-) create mode 100644 tools/sync_models/test/test_extended_thinking_field.py diff --git a/nodes/src/nodes/llm_anthropic/services.json b/nodes/src/nodes/llm_anthropic/services.json index b7db2c2d7..e18264dbb 100644 --- a/nodes/src/nodes/llm_anthropic/services.json +++ b/nodes/src/nodes/llm_anthropic/services.json @@ -543,8 +543,7 @@ "object": "claude-3-haiku", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "llm.cloud.modelSource" ] }, "anthropic.claude-fable-5": { diff --git a/tools/sync_models/src/core/patcher.py b/tools/sync_models/src/core/patcher.py index 5ab34628b..0a10c0e55 100644 --- a/tools/sync_models/src/core/patcher.py +++ b/tools/sync_models/src/core/patcher.py @@ -277,7 +277,10 @@ def _detect_namespace(fields: Dict[str, Any]) -> str: return '' -def _repair_field_objects(fields: Dict[str, Any]) -> bool: +_EXTENDED_THINKING = 'extendedThinking' + + +def _repair_field_objects(fields: Dict[str, Any], profiles: Dict[str, Any] | None = None) -> bool: """ Ensure every profile field object that exposes ``llm.cloud.apikey`` also exposes ``llm.cloud.modelSource``, and that ``llm.cloud.modelSource`` is @@ -319,6 +322,19 @@ def _repair_field_objects(fields: Dict[str, Any]) -> bool: break # only one apikey entry per object has_apikey = 'llm.cloud.apikey' in props + + # extendedThinking toggle: present iff the model reasons (capabilities.reasoning). + # Keeps the UI self-maintaining across syncs; skips 'custom' and non-model objects. + obj = value.get('object') + if profiles is not None and has_apikey and isinstance(obj, str) and obj != 'custom' and obj in profiles: + reasons = bool(((profiles.get(obj) or {}).get('capabilities') or {}).get('reasoning')) + if reasons and _EXTENDED_THINKING not in props: + props.append(_EXTENDED_THINKING) + repaired = True + elif not reasons and _EXTENDED_THINKING in props: + props.remove(_EXTENDED_THINKING) + repaired = True + has_model_source = 'llm.cloud.modelSource' in props if has_apikey and not has_model_source: props.append('llm.cloud.modelSource') @@ -357,12 +373,12 @@ def _update_fields_for_added( field_key = f'{namespace}.{profile_key}' - # 1. Field object + # 1. Field object — reasoning models get the extendedThinking toggle (before modelSource). if field_key not in fields: - fields[field_key] = { - 'object': profile_key, - 'properties': ['llm.cloud.apikey', 'llm.cloud.modelSource'], - } + props = ['llm.cloud.apikey', 'llm.cloud.modelSource'] + if bool((profile.get('capabilities') or {}).get('reasoning')): + props.insert(1, _EXTENDED_THINKING) + fields[field_key] = {'object': profile_key, 'properties': props} # 2 & 3. enum + conditional live inside the profile selector field profile_field = fields.get(f'{namespace}.profile') @@ -518,8 +534,8 @@ def patch( ns = _detect_namespace(fields) - # Repair existing field objects that are missing llm.cloud.modelSource - _repair_field_objects(fields) + # Repair existing field objects (modelSource + capabilities-aware extendedThinking) + _repair_field_objects(fields, updated_profiles) for key in sorted(_added): profile = updated_profiles.get(key, {}) diff --git a/tools/sync_models/src/providers/base.py b/tools/sync_models/src/providers/base.py index c09e829d3..cf4962121 100644 --- a/tools/sync_models/src/providers/base.py +++ b/tools/sync_models/src/providers/base.py @@ -623,7 +623,7 @@ def _run_merge( _fields = json.loads(_raw[_f_start:_f_end]) _fields_copy = copy.deepcopy(_fields) - _needs_repair = _repair_field_objects(_fields_copy) + _needs_repair = _repair_field_objects(_fields_copy, updated_profiles) except Exception: pass diff --git a/tools/sync_models/test/test_extended_thinking_field.py b/tools/sync_models/test/test_extended_thinking_field.py new file mode 100644 index 000000000..c9e169dae --- /dev/null +++ b/tools/sync_models/test/test_extended_thinking_field.py @@ -0,0 +1,57 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# ============================================================================= + +"""Pins the capabilities-aware `extendedThinking` UI toggle in the field patcher. + +Reasoning models (capabilities.reasoning) carry the toggle; non-reasoning models do +not — so the UI stays self-maintaining across model syncs. +""" + +from __future__ import annotations + +from core.patcher import _repair_field_objects, _update_fields_for_added + + +def test_repair_adds_toggle_to_reasoning_and_removes_from_non_reasoning(): + fields = { + 'ns.r': {'object': 'r', 'properties': ['llm.cloud.apikey', 'llm.cloud.modelSource']}, + 'ns.n': {'object': 'n', 'properties': ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource']}, + } + profiles = {'r': {'capabilities': {'reasoning': True}}, 'n': {'capabilities': {}}} + + _repair_field_objects(fields, profiles) + + # toggle sits before modelSource (which the patcher keeps last) + assert fields['ns.r']['properties'] == ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource'] + assert fields['ns.n']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] + + +def test_repair_skips_custom(): + fields = { + 'ns.custom': { + 'object': 'custom', + 'properties': ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource'], + } + } + _repair_field_objects(fields, {'custom': {}}) + assert 'extendedThinking' in fields['ns.custom']['properties'] + + +def test_repair_without_profiles_leaves_toggle_untouched(): + fields = {'ns.r': {'object': 'r', 'properties': ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource']}} + _repair_field_objects(fields) # no profiles -> only the modelSource repair runs + assert 'extendedThinking' in fields['ns.r']['properties'] + + +def test_added_reasoning_profile_gets_toggle(): + fields = {'ns.profile': {'enum': [], 'conditional': []}} + _update_fields_for_added(fields, 'ns', 'newr', {'capabilities': {'reasoning': True}, 'title': 'New R'}, set()) + assert fields['ns.newr']['properties'] == ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource'] + + +def test_added_non_reasoning_profile_has_no_toggle(): + fields = {'ns.profile': {'enum': [], 'conditional': []}} + _update_fields_for_added(fields, 'ns', 'newn', {'capabilities': {}, 'title': 'New N'}, set()) + assert fields['ns.newn']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] From 1d35fac50fd4a3b24b5f25e97984fa85a1efaece Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Tue, 28 Jul 2026 18:26:50 +0200 Subject: [PATCH 16/20] =?UTF-8?q?refactor(llm):=20address=20CodeRabbit=20?= =?UTF-8?q?=E2=80=94=20dead=20adapters,=20store=3DFalse,=20Responses=20fal?= =?UTF-8?q?lback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove unused AnthropicAdapter/OpenAIAdapter (.stream() wrappers) + their tests; production routes through the Native* adapters. - NativeOpenAIResponsesAdapter now passes store=False (no server-side retention). - _chat_string_responses raises on empty output so it falls back (matching the Anthropic path) instead of returning '' silently. --- packages/ai/src/ai/common/chat.py | 3 + packages/ai/src/ai/common/llm_adapter.py | 71 +------------- .../ai/common/test_llm_adapter_anthropic.py | 97 ------------------- .../ai/common/test_llm_adapter_openai.py | 93 ------------------ 4 files changed, 4 insertions(+), 260 deletions(-) delete mode 100644 packages/ai/tests/ai/common/test_llm_adapter_anthropic.py delete mode 100644 packages/ai/tests/ai/common/test_llm_adapter_openai.py diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index 528ae1ab1..3911cc077 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -406,6 +406,9 @@ def _chat_string_responses( try: adapter = NativeOpenAIResponsesAdapter(self) text, _items = drive_adapter(adapter, prompt, on_chunk, on_reasoning_chunk) + if not text: + # No text (e.g. response.failed) → route to the fallback below, like the Anthropic path. + raise RuntimeError('OpenAI Responses stream produced no text') if on_finish is not None: on_finish(adapter.finish_reason) return text diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 190633518..911735522 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -207,76 +207,6 @@ def stream(self, user_text: str) -> Iterator[Event]: yield Event('done', items=[assistant]) -class AnthropicAdapter: - """history is a Messages-API `messages` list; done.items is the assembled content - (thinking `signature` / redacted blocks intact) — append verbatim, never rebuild. - """ - - def __init__( - self, - client: Any, - model: str, - max_tokens: int = 16000, - thinking: dict | None = None, - history: list[Any] | None = None, - ): - self.client = client - self.model = model - self.max_tokens = max_tokens - self.thinking = thinking - self.history: list[Any] = history if history is not None else [] - - def stream(self, user_text: str) -> Iterator[Event]: - self.history.append({'role': 'user', 'content': user_text}) - kwargs: dict[str, Any] = {'model': self.model, 'max_tokens': self.max_tokens, 'messages': self.history} - if self.thinking: - kwargs['thinking'] = self.thinking - with self.client.messages.stream(**kwargs) as stream: - for ev in stream: - if getattr(ev, 'type', '') != 'content_block_delta': - continue - delta = ev.delta - dtype = getattr(delta, 'type', '') - if dtype == 'thinking_delta': - yield Event('thinking', getattr(delta, 'thinking', '') or '') - elif dtype == 'text_delta': - yield Event('text', getattr(delta, 'text', '') or '') - final = stream.get_final_message() - self.history.append({'role': 'assistant', 'content': final.content}) - yield Event('done', items=final.content) - - -class OpenAIAdapter: - """history is a Responses-API `input` item list; done.items are the output items - (encrypted reasoning + assistant) — extend history verbatim, never replay a subset. - """ - - def __init__(self, client: Any, model: str, effort: str = 'medium', history: list[Any] | None = None): - self.client = client - self.model = model - self.effort = effort - self.history: list[Any] = history if history is not None else [] - - def stream(self, user_text: str) -> Iterator[Event]: - self.history.append({'role': 'user', 'content': user_text}) - with self.client.responses.stream( - model=self.model, - input=self.history, - store=False, # stateless: encrypted_content rides back on the reasoning items. - reasoning={'effort': self.effort, 'summary': 'auto', 'context': 'all_turns'}, - ) as stream: - for ev in stream: - etype = getattr(ev, 'type', '') - if etype in ('response.reasoning_summary_text.delta', 'response.reasoning_text.delta'): - yield Event('thinking', getattr(ev, 'delta', '') or '') - elif etype == 'response.output_text.delta': - yield Event('text', getattr(ev, 'delta', '') or '') - final = stream.get_final_response() - items = [item.model_dump() for item in final.output] - self.history.extend(items) - yield Event('done', items=items) - - class NativeOpenAIResponsesAdapter: """Bridges the OpenAI Responses reasoning stream (create/stream=True) to Events.""" @@ -292,6 +222,7 @@ def stream(self, user_text: str) -> Iterator[Event]: stream = chat._raw_client.responses.create( model=chat._model, input=user_text, + store=False, # stateless: don't retain prompts/responses server-side (30-day default). reasoning={'summary': 'auto'}, max_output_tokens=chat._modelOutputTokens, stream=True, diff --git a/packages/ai/tests/ai/common/test_llm_adapter_anthropic.py b/packages/ai/tests/ai/common/test_llm_adapter_anthropic.py deleted file mode 100644 index 6b99099f2..000000000 --- a/packages/ai/tests/ai/common/test_llm_adapter_anthropic.py +++ /dev/null @@ -1,97 +0,0 @@ -# ============================================================================= -# MIT License -# Copyright (c) 2026 Aparavi Software AG -# ============================================================================= - -"""Pins AnthropicAdapter: thinking/text deltas → Events, opaque content in done.items.""" - -from ai.common.llm_adapter import AnthropicAdapter, Event - - -class _Delta: - def __init__(self, type, thinking='', text=''): - self.type = type - self.thinking = thinking - self.text = text - - -class _Ev: - def __init__(self, type, delta=None): - self.type = type - self.delta = delta - - -class _Final: - def __init__(self, content): - self.content = content - - -class _Stream: - def __init__(self, events, final): - self._events = events - self._final = final - - def __enter__(self): - return self - - def __exit__(self, *a): - return False - - def __iter__(self): - return iter(self._events) - - def get_final_message(self): - return self._final - - -class _Messages: - def __init__(self, stream): - self._stream = stream - self.seen = None - - def stream(self, **kwargs): - self.seen = {**kwargs, 'messages': list(kwargs['messages'])} - return self._stream - - -class _Client: - def __init__(self, stream): - self.messages = _Messages(stream) - - -def test_deltas_to_events_and_opaque_done_items(): - final_content = [ - {'type': 'thinking', 'thinking': 'reasoning', 'signature': 'sig'}, - {'type': 'text', 'text': 'answer'}, - ] - events = [ - _Ev('content_block_delta', _Delta('thinking_delta', thinking='reasoning')), - _Ev('content_block_delta', _Delta('text_delta', text='answer')), - _Ev('message_stop'), - ] - client = _Client(_Stream(events, _Final(final_content))) - adapter = AnthropicAdapter( - client, model='claude-sonnet-4-6', thinking={'type': 'adaptive', 'display': 'summarized'} - ) - - out = list(adapter.stream('q')) - - assert Event('thinking', 'reasoning') in out - assert Event('text', 'answer') in out - done = out[-1] - assert done.type == 'done' - assert done.items == final_content # opaque, verbatim (signature intact) - - assert adapter.history == [ - {'role': 'user', 'content': 'q'}, - {'role': 'assistant', 'content': final_content}, - ] - assert client.messages.seen['thinking'] == {'type': 'adaptive', 'display': 'summarized'} - assert client.messages.seen['messages'] == [{'role': 'user', 'content': 'q'}] - - -def test_thinking_omitted_when_none(): - client = _Client(_Stream([_Ev('content_block_delta', _Delta('text_delta', text='hi'))], _Final([]))) - adapter = AnthropicAdapter(client, model='claude-3-haiku') - list(adapter.stream('q')) - assert 'thinking' not in client.messages.seen diff --git a/packages/ai/tests/ai/common/test_llm_adapter_openai.py b/packages/ai/tests/ai/common/test_llm_adapter_openai.py deleted file mode 100644 index bd12b14b9..000000000 --- a/packages/ai/tests/ai/common/test_llm_adapter_openai.py +++ /dev/null @@ -1,93 +0,0 @@ -# ============================================================================= -# MIT License -# Copyright (c) 2026 Aparavi Software AG -# ============================================================================= - -"""Pins OpenAIAdapter: Responses deltas → Events, opaque output items in done.items.""" - -from ai.common.llm_adapter import Event, OpenAIAdapter - - -class _Ev: - def __init__(self, type, delta=''): - self.type = type - self.delta = delta - - -class _Item: - def __init__(self, dump): - self._dump = dump - - def model_dump(self): - return self._dump - - -class _Final: - def __init__(self, output): - self.output = output - - -class _Stream: - def __init__(self, events, final): - self._events = events - self._final = final - - def __enter__(self): - return self - - def __exit__(self, *a): - return False - - def __iter__(self): - return iter(self._events) - - def get_final_response(self): - return self._final - - -class _Responses: - def __init__(self, stream): - self._stream = stream - self.seen = None - - def stream(self, **kwargs): - self.seen = {**kwargs, 'input': list(kwargs['input'])} - return self._stream - - -class _Client: - def __init__(self, stream): - self.responses = _Responses(stream) - - -def test_deltas_to_events_and_opaque_output_items(): - events = [ - _Ev('response.reasoning_summary_text.delta', 'think'), - _Ev('response.output_text.delta', 'ans'), - ] - output = [ - _Item({'type': 'reasoning', 'encrypted_content': 'enc'}), - _Item({'type': 'message', 'content': 'ans'}), - ] - client = _Client(_Stream(events, _Final(output))) - adapter = OpenAIAdapter(client, model='gpt-5.6') - - out = list(adapter.stream('q')) - - assert Event('thinking', 'think') in out - assert Event('text', 'ans') in out - done = out[-1] - assert done.type == 'done' - assert done.items == [ - {'type': 'reasoning', 'encrypted_content': 'enc'}, - {'type': 'message', 'content': 'ans'}, - ] - - # history: user turn, then output items extended verbatim - assert adapter.history == [ - {'role': 'user', 'content': 'q'}, - {'type': 'reasoning', 'encrypted_content': 'enc'}, - {'type': 'message', 'content': 'ans'}, - ] - assert client.responses.seen['store'] is False - assert client.responses.seen['input'] == [{'role': 'user', 'content': 'q'}] From b5560802abf5543f96880d9affdfafbc1a114091 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Wed, 29 Jul 2026 17:26:58 +0200 Subject: [PATCH 17/20] fix(sync-models): scope the extendedThinking toggle to nodes that define the field MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The capabilities-aware repair only adds/manages extendedThinking on nodes whose fields define it (today: llm_anthropic), so it never leaks into other providers (openai, gemini, …) that neither define nor read the flag. --- tools/sync_models/src/core/patcher.py | 15 ++++++-- .../test/test_extended_thinking_field.py | 38 +++++++++++++++---- 2 files changed, 43 insertions(+), 10 deletions(-) diff --git a/tools/sync_models/src/core/patcher.py b/tools/sync_models/src/core/patcher.py index 0a10c0e55..56673ade5 100644 --- a/tools/sync_models/src/core/patcher.py +++ b/tools/sync_models/src/core/patcher.py @@ -324,9 +324,17 @@ def _repair_field_objects(fields: Dict[str, Any], profiles: Dict[str, Any] | Non has_apikey = 'llm.cloud.apikey' in props # extendedThinking toggle: present iff the model reasons (capabilities.reasoning). - # Keeps the UI self-maintaining across syncs; skips 'custom' and non-model objects. + # Only managed for nodes that DEFINE the field (today: llm_anthropic) so it never + # leaks into providers that neither define nor read it. Skips 'custom'/non-model objects. obj = value.get('object') - if profiles is not None and has_apikey and isinstance(obj, str) and obj != 'custom' and obj in profiles: + if ( + profiles is not None + and _EXTENDED_THINKING in fields + and has_apikey + and isinstance(obj, str) + and obj != 'custom' + and obj in profiles + ): reasons = bool(((profiles.get(obj) or {}).get('capabilities') or {}).get('reasoning')) if reasons and _EXTENDED_THINKING not in props: props.append(_EXTENDED_THINKING) @@ -376,7 +384,8 @@ def _update_fields_for_added( # 1. Field object — reasoning models get the extendedThinking toggle (before modelSource). if field_key not in fields: props = ['llm.cloud.apikey', 'llm.cloud.modelSource'] - if bool((profile.get('capabilities') or {}).get('reasoning')): + # Only nodes that define the extendedThinking field get the toggle (today: anthropic). + if _EXTENDED_THINKING in fields and bool((profile.get('capabilities') or {}).get('reasoning')): props.insert(1, _EXTENDED_THINKING) fields[field_key] = {'object': profile_key, 'properties': props} diff --git a/tools/sync_models/test/test_extended_thinking_field.py b/tools/sync_models/test/test_extended_thinking_field.py index c9e169dae..7c489ab3f 100644 --- a/tools/sync_models/test/test_extended_thinking_field.py +++ b/tools/sync_models/test/test_extended_thinking_field.py @@ -5,17 +5,22 @@ """Pins the capabilities-aware `extendedThinking` UI toggle in the field patcher. -Reasoning models (capabilities.reasoning) carry the toggle; non-reasoning models do -not — so the UI stays self-maintaining across model syncs. +The toggle is only managed for nodes that DEFINE the field (today: llm_anthropic). +Within such a node, reasoning models carry it and non-reasoning models do not, so the +UI stays self-maintaining across model syncs without leaking into other providers. """ from __future__ import annotations from core.patcher import _repair_field_objects, _update_fields_for_added +# A node whose fields define the extendedThinking widget (like llm_anthropic). +_FIELD_DEF = {'extendedThinking': {'type': 'boolean', 'title': 'Extended thinking', 'default': False}} + def test_repair_adds_toggle_to_reasoning_and_removes_from_non_reasoning(): fields = { + **_FIELD_DEF, 'ns.r': {'object': 'r', 'properties': ['llm.cloud.apikey', 'llm.cloud.modelSource']}, 'ns.n': {'object': 'n', 'properties': ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource']}, } @@ -28,30 +33,49 @@ def test_repair_adds_toggle_to_reasoning_and_removes_from_non_reasoning(): assert fields['ns.n']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] +def test_toggle_never_added_when_field_undefined(): + # A provider node that does NOT define extendedThinking (e.g. openai) must be left alone, + # even for reasoning models — the toggle must not leak in. + fields = {'ns.r': {'object': 'r', 'properties': ['llm.cloud.apikey', 'llm.cloud.modelSource']}} + profiles = {'r': {'capabilities': {'reasoning': True}}} + _repair_field_objects(fields, profiles) + assert fields['ns.r']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] + + def test_repair_skips_custom(): fields = { + **_FIELD_DEF, 'ns.custom': { 'object': 'custom', 'properties': ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource'], - } + }, } _repair_field_objects(fields, {'custom': {}}) assert 'extendedThinking' in fields['ns.custom']['properties'] def test_repair_without_profiles_leaves_toggle_untouched(): - fields = {'ns.r': {'object': 'r', 'properties': ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource']}} + fields = { + **_FIELD_DEF, + 'ns.r': {'object': 'r', 'properties': ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource']}, + } _repair_field_objects(fields) # no profiles -> only the modelSource repair runs assert 'extendedThinking' in fields['ns.r']['properties'] -def test_added_reasoning_profile_gets_toggle(): - fields = {'ns.profile': {'enum': [], 'conditional': []}} +def test_added_reasoning_profile_gets_toggle_when_field_defined(): + fields = {**_FIELD_DEF, 'ns.profile': {'enum': [], 'conditional': []}} _update_fields_for_added(fields, 'ns', 'newr', {'capabilities': {'reasoning': True}, 'title': 'New R'}, set()) assert fields['ns.newr']['properties'] == ['llm.cloud.apikey', 'extendedThinking', 'llm.cloud.modelSource'] +def test_added_reasoning_profile_no_toggle_when_field_undefined(): + fields = {'ns.profile': {'enum': [], 'conditional': []}} # no extendedThinking field def + _update_fields_for_added(fields, 'ns', 'newr', {'capabilities': {'reasoning': True}, 'title': 'New R'}, set()) + assert fields['ns.newr']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] + + def test_added_non_reasoning_profile_has_no_toggle(): - fields = {'ns.profile': {'enum': [], 'conditional': []}} + fields = {**_FIELD_DEF, 'ns.profile': {'enum': [], 'conditional': []}} _update_fields_for_added(fields, 'ns', 'newn', {'capabilities': {}, 'title': 'New N'}, set()) assert fields['ns.newn']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] From 8314681461810b2c3431d1b61a2ec87650dfd9ba Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Wed, 29 Jul 2026 17:33:37 +0200 Subject: [PATCH 18/20] docs(llm_anthropic): soften the extendedThinking field description --- nodes/src/nodes/llm_anthropic/services.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nodes/src/nodes/llm_anthropic/services.json b/nodes/src/nodes/llm_anthropic/services.json index e18264dbb..272ed1e52 100644 --- a/nodes/src/nodes/llm_anthropic/services.json +++ b/nodes/src/nodes/llm_anthropic/services.json @@ -336,7 +336,7 @@ "type": "boolean", "title": "Extended thinking", "default": false, - "description": "Enable Anthropic extended thinking (reasoning) for this node. Off by default. Only affects reasoning-capable models, and only on the interactive streaming path — agent / expectJson calls never use thinking." + "description": "Enable Anthropic extended thinking (reasoning) for this node. Off by default. Applies to reasoning-capable models on the interactive chat path." }, "anthropic.custom": { "object": "custom", From 6196feb45daa2291556ae087e377985a3b3347e1 Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Thu, 30 Jul 2026 17:45:44 +0200 Subject: [PATCH 19/20] fix(llm): keep the non-streaming path on invoke via adapter.collect() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _chat drains through LangChainAdapter.collect() (invoke) instead of stream(), so the streaming fallback in chat_string recovers with a genuinely different mechanism again — while content normalization stays unified through the adapter. Addresses review #1. --- packages/ai/src/ai/common/chat.py | 13 ++++------- packages/ai/src/ai/common/llm_adapter.py | 11 +++++++++ .../ai/common/test_chat_string_wiring.py | 18 ++++++++++++--- .../ai/common/test_llm_adapter_langchain.py | 23 +++++++++++++++++++ 4 files changed, 53 insertions(+), 12 deletions(-) diff --git a/packages/ai/src/ai/common/chat.py b/packages/ai/src/ai/common/chat.py index 3911cc077..701c7fdc4 100644 --- a/packages/ai/src/ai/common/chat.py +++ b/packages/ai/src/ai/common/chat.py @@ -184,15 +184,10 @@ def _chat(self, prompt: str) -> str: Should raise appropriate exceptions for API failures, authentication errors, or other provider-specific issues """ - # Non-streaming: drain the adapter with no display sinks — same iterator and - # normalization as streaming, so item capture is structural on every path. - _llm = self._llm - if hasattr(_llm, 'stream'): - adapter = LangChainAdapter(_llm, stream_kwargs=_stop_kwargs()) - text, _items = drive_adapter(adapter, prompt) - return text - # Backends/mocks without .stream: invoke and normalize the content to a string. - return flatten_content(_llm.invoke(prompt, **_stop_kwargs()).content) + # Non-streaming: invoke through the adapter — same shared normalization as streaming, + # but a genuinely different mechanism, so the streaming fallback can still recover. + text, _items = LangChainAdapter(self._llm, stream_kwargs=_stop_kwargs()).collect(prompt) + return text def getTokens(self, value: str) -> int: """ diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index 911735522..bbcb0d651 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -206,6 +206,17 @@ def stream(self, user_text: str) -> Iterator[Event]: self.history.append(assistant) yield Event('done', items=[assistant]) + def collect(self, user_text: str) -> tuple[str, list[Any]]: + """Non-streaming drain: invoke() + shared normalization. A genuinely different + mechanism from stream(), so it can still recover when streaming fails. + """ + self.history.append({'role': 'user', 'content': user_text}) + result = self.llm.invoke(self.history, **self.stream_kwargs) + text = flatten_content(getattr(result, 'content', '')) + assistant = {'role': 'assistant', 'content': text} + self.history.append(assistant) + return text, [assistant] + class NativeOpenAIResponsesAdapter: """Bridges the OpenAI Responses reasoning stream (create/stream=True) to Events.""" diff --git a/packages/ai/tests/ai/common/test_chat_string_wiring.py b/packages/ai/tests/ai/common/test_chat_string_wiring.py index b1ab27105..c75f926eb 100644 --- a/packages/ai/tests/ai/common/test_chat_string_wiring.py +++ b/packages/ai/tests/ai/common/test_chat_string_wiring.py @@ -23,6 +23,10 @@ def stream(self, messages, **kwargs): self.kwargs = kwargs yield from self._pieces + def invoke(self, messages, **kwargs): + self.kwargs = kwargs + return self._pieces[0] + class _Chat(ChatBase): # Bypass the config-driven __init__; wire only what chat_string reads. @@ -72,8 +76,16 @@ def test_stop_sequences_reach_the_model(): assert llm.kwargs == {'stop': ['\nObservation:']} -def test_chat_nonstreaming_drains_adapter(): - # The unify: _chat (no display callbacks) drains the same adapter and returns text. +def test_chat_nonstreaming_invokes_via_adapter(): + # _chat drains via collect() → invoke() (a different mechanism than stream(), so the + # streaming fallback can recover); content is normalized (thinking dropped) and stop passes. + from ai.common.llm_native_stream import STOP_SEQUENCES_VAR + llm = _FakeLLM([_Piece([{'type': 'thinking', 'thinking': 'x'}, {'type': 'text', 'text': 'plain answer'}])]) chat = _Chat(llm) - assert chat._chat('q') == 'plain answer' + token = STOP_SEQUENCES_VAR.set(['\nObservation:']) + try: + assert chat._chat('q') == 'plain answer' + finally: + STOP_SEQUENCES_VAR.reset(token) + assert llm.kwargs == {'stop': ['\nObservation:']} diff --git a/packages/ai/tests/ai/common/test_llm_adapter_langchain.py b/packages/ai/tests/ai/common/test_llm_adapter_langchain.py index 6bbd07006..4b3d1a994 100644 --- a/packages/ai/tests/ai/common/test_llm_adapter_langchain.py +++ b/packages/ai/tests/ai/common/test_llm_adapter_langchain.py @@ -26,6 +26,12 @@ def stream(self, messages, **kwargs): for p in self._pieces: yield p if isinstance(p, _Piece) else _Piece(p) + def invoke(self, messages, **kwargs): + self.seen = list(messages) + self.kwargs = kwargs + p = self._pieces[0] + return p if isinstance(p, _Piece) else _Piece(p) + def test_normalizes_blocks_and_records_history(): llm = _FakeLLM([[{'type': 'thinking', 'thinking': 'r'}, {'type': 'text', 'text': 'Hi there'}]]) @@ -62,3 +68,20 @@ def test_passes_stream_kwargs_and_tracks_finish_reason(): list(adapter.stream('q')) assert llm.kwargs == {'stop': ['\nObservation:']} assert adapter.finish_reason == 'stop' + + +def test_collect_uses_invoke_and_normalizes(): + # collect() is the non-streaming drain: invoke() + shared normalization, thinking dropped. + llm = _FakeLLM([[{'type': 'thinking', 'thinking': 'r'}, {'type': 'text', 'text': 'answer'}]]) + adapter = LangChainAdapter(llm, stream_kwargs={'stop': ['x']}) + + text, items = adapter.collect('q') + + assert text == 'answer' + assert items == [{'role': 'assistant', 'content': 'answer'}] + assert llm.seen == [{'role': 'user', 'content': 'q'}] # invoke received the history + assert llm.kwargs == {'stop': ['x']} # stop kwargs threaded to invoke + assert adapter.history == [ + {'role': 'user', 'content': 'q'}, + {'role': 'assistant', 'content': 'answer'}, + ] From 8c5dacb5465dcffd74e34a9a97f354ddbc87477a Mon Sep 17 00:00:00 2001 From: Ariel Vernaza Date: Thu, 30 Jul 2026 19:18:46 +0200 Subject: [PATCH 20/20] =?UTF-8?q?fix(llm):=20address=20review=20=E2=80=94?= =?UTF-8?q?=20parse=5Fbool,=20opus-5=20toggle,=20custom,=20Responses=20tes?= =?UTF-8?q?ts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - llm_anthropic: parse_bool for the extendedThinking config (handles 'false'/'0'/'no'). - services.json: add the toggle to claude-opus-5/-fast, drop it from the inert custom profile, and order extendedThinking before modelSource (the sync invariant). - patcher: manage 'custom' too (non-reasoning → toggle removed), so it self-heals. - NativeAnthropic/NativeOpenAIResponses adapters drop the unused history constructor arg. - Add Responses tests: failed→error, store=False, and the no-text → invoke fallback. --- nodes/src/nodes/llm_anthropic/anthropic.py | 3 +- nodes/src/nodes/llm_anthropic/services.json | 77 ++++++++++--------- packages/ai/src/ai/common/llm_adapter.py | 5 +- .../ai/src/ai/common/llm_native_stream.py | 5 +- .../ai/common/test_chat_string_wiring.py | 18 +++++ .../test_native_openai_responses_adapter.py | 12 +++ tools/sync_models/src/core/patcher.py | 4 +- .../test/test_extended_thinking_field.py | 5 +- 8 files changed, 82 insertions(+), 47 deletions(-) diff --git a/nodes/src/nodes/llm_anthropic/anthropic.py b/nodes/src/nodes/llm_anthropic/anthropic.py index ccb604192..941fd8fa2 100644 --- a/nodes/src/nodes/llm_anthropic/anthropic.py +++ b/nodes/src/nodes/llm_anthropic/anthropic.py @@ -33,6 +33,7 @@ from ai.common.chat import ChatBase from ai.common.config import Config from ai.common.llm_native_stream import build_anthropic_thinking_kwargs, gate_model_name +from ai.common.utils import parse_bool from langchain_anthropic import ChatAnthropic @@ -75,7 +76,7 @@ def __init__(self, provider: str, connConfig: Dict[str, Any], bag: Dict[str, Any # it per call, so thinking activates on the interactive streaming path only — never on the # agent / expectJson path (which has no streaming and stays on the plain LangChain client). self._thinking_mode_kwargs: Dict[str, Any] = {} - if self._is_reasoning and bool(config.get('extendedThinking')): + if self._is_reasoning and parse_bool(config.get('extendedThinking')): self._thinking_mode_kwargs = build_anthropic_thinking_kwargs(model_gate, self._modelOutputTokens) self._extended_thinking = bool(self._thinking_mode_kwargs) if self._extended_thinking: diff --git a/nodes/src/nodes/llm_anthropic/services.json b/nodes/src/nodes/llm_anthropic/services.json index 272ed1e52..9cbdabfc6 100644 --- a/nodes/src/nodes/llm_anthropic/services.json +++ b/nodes/src/nodes/llm_anthropic/services.json @@ -344,48 +344,47 @@ "model", "modelTotalTokens", "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "llm.cloud.modelSource" ] }, "anthropic.claude-sonnet-4-6": { "object": "claude-sonnet-4-6", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-4-6": { "object": "claude-opus-4-6", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-haiku-4-5": { "object": "claude-haiku-4-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-sonnet-4-5": { "object": "claude-sonnet-4-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-4-5": { "object": "claude-opus-4-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.profile": { @@ -535,8 +534,8 @@ "object": "claude-opus-4-7", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-3-haiku": { @@ -550,102 +549,103 @@ "object": "claude-fable-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-fable-latest": { "object": "claude-fable-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-haiku-latest": { "object": "claude-haiku-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-4": { "object": "claude-opus-4", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-4-1": { "object": "claude-opus-4-1", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-4-7-fast": { "object": "claude-opus-4-7-fast", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-4-8": { "object": "claude-opus-4-8", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-4-8-fast": { "object": "claude-opus-4-8-fast", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-latest": { "object": "claude-opus-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-sonnet-4": { "object": "claude-sonnet-4", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-sonnet-5": { "object": "claude-sonnet-5", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-sonnet-latest": { "object": "claude-sonnet-latest", "properties": [ "llm.cloud.apikey", - "llm.cloud.modelSource", - "extendedThinking" + "extendedThinking", + "llm.cloud.modelSource" ] }, "anthropic.claude-opus-5": { "object": "claude-opus-5", "properties": [ "llm.cloud.apikey", + "extendedThinking", "llm.cloud.modelSource" ] }, @@ -653,6 +653,7 @@ "object": "claude-opus-5-fast", "properties": [ "llm.cloud.apikey", + "extendedThinking", "llm.cloud.modelSource" ] } diff --git a/packages/ai/src/ai/common/llm_adapter.py b/packages/ai/src/ai/common/llm_adapter.py index bbcb0d651..b765e82de 100644 --- a/packages/ai/src/ai/common/llm_adapter.py +++ b/packages/ai/src/ai/common/llm_adapter.py @@ -221,9 +221,10 @@ def collect(self, user_text: str) -> tuple[str, list[Any]]: class NativeOpenAIResponsesAdapter: """Bridges the OpenAI Responses reasoning stream (create/stream=True) to Events.""" - def __init__(self, chat: Any, history: list[Any] | None = None): + def __init__(self, chat: Any): + # Single-turn: history is not accepted (the request is built from user_text, not history). self.chat = chat - self.history: list[Any] = history if history is not None else [] + self.history: list[Any] = [] self.finish_reason: Optional[str] = None def stream(self, user_text: str) -> Iterator[Event]: diff --git a/packages/ai/src/ai/common/llm_native_stream.py b/packages/ai/src/ai/common/llm_native_stream.py index 0d9823c54..1fae0c196 100644 --- a/packages/ai/src/ai/common/llm_native_stream.py +++ b/packages/ai/src/ai/common/llm_native_stream.py @@ -175,9 +175,10 @@ class NativeAnthropicAdapter: Same payload + raw create(stream=True) path as before; now yields normalized Events. """ - def __init__(self, chat: Any, history: Optional[list] = None): + def __init__(self, chat: Any): + # Single-turn: history is not accepted (the request is built from user_text, not history). self.chat = chat - self.history = history if history is not None else [] + self.history: list = [] self.finish_reason: Optional[str] = None def stream(self, user_text: str): diff --git a/packages/ai/tests/ai/common/test_chat_string_wiring.py b/packages/ai/tests/ai/common/test_chat_string_wiring.py index c75f926eb..c2aae10cb 100644 --- a/packages/ai/tests/ai/common/test_chat_string_wiring.py +++ b/packages/ai/tests/ai/common/test_chat_string_wiring.py @@ -89,3 +89,21 @@ def test_chat_nonstreaming_invokes_via_adapter(): finally: STOP_SEQUENCES_VAR.reset(token) assert llm.kwargs == {'stop': ['\nObservation:']} + + +def test_responses_no_text_falls_back_to_invoke(): + # response.failed with no text → the no-text guard raises → non-streaming invoke fallback. + class _FailEvent: + type = 'response.failed' + + class _Responses: + def create(self, **kwargs): + return iter([_FailEvent()]) + + class _RawClient: + responses = _Responses() + + chat = _Chat(_FakeLLM([_Piece('fallback answer')])) + chat._raw_client = _RawClient() + result = chat._chat_string_responses('q', on_chunk=lambda t: None, emitted={'any': False}) + assert result == 'fallback answer' diff --git a/packages/ai/tests/ai/common/test_native_openai_responses_adapter.py b/packages/ai/tests/ai/common/test_native_openai_responses_adapter.py index 71bcae5cd..773ac07bc 100644 --- a/packages/ai/tests/ai/common/test_native_openai_responses_adapter.py +++ b/packages/ai/tests/ai/common/test_native_openai_responses_adapter.py @@ -72,3 +72,15 @@ def test_incomplete_maps_to_reason(): adapter = NativeOpenAIResponsesAdapter(_Chat(events)) list(adapter.stream('q')) assert adapter.finish_reason == 'max_output_tokens' + + +def test_failed_response_sets_error_finish(): + adapter = NativeOpenAIResponsesAdapter(_Chat([_Ev('response.failed')])) + list(adapter.stream('q')) + assert adapter.finish_reason == 'error' + + +def test_passes_store_false(): + chat = _Chat([_Ev('response.output_text.delta', delta='x')]) + list(NativeOpenAIResponsesAdapter(chat).stream('q')) + assert chat._raw_client.responses.seen['store'] is False diff --git a/tools/sync_models/src/core/patcher.py b/tools/sync_models/src/core/patcher.py index 56673ade5..8870cb33b 100644 --- a/tools/sync_models/src/core/patcher.py +++ b/tools/sync_models/src/core/patcher.py @@ -325,14 +325,14 @@ def _repair_field_objects(fields: Dict[str, Any], profiles: Dict[str, Any] | Non # extendedThinking toggle: present iff the model reasons (capabilities.reasoning). # Only managed for nodes that DEFINE the field (today: llm_anthropic) so it never - # leaks into providers that neither define nor read it. Skips 'custom'/non-model objects. + # leaks into providers that neither define nor read it. 'custom' has no capabilities, + # so it is treated as non-reasoning and the toggle is removed (it would be inert). obj = value.get('object') if ( profiles is not None and _EXTENDED_THINKING in fields and has_apikey and isinstance(obj, str) - and obj != 'custom' and obj in profiles ): reasons = bool(((profiles.get(obj) or {}).get('capabilities') or {}).get('reasoning')) diff --git a/tools/sync_models/test/test_extended_thinking_field.py b/tools/sync_models/test/test_extended_thinking_field.py index 7c489ab3f..ac804f851 100644 --- a/tools/sync_models/test/test_extended_thinking_field.py +++ b/tools/sync_models/test/test_extended_thinking_field.py @@ -42,7 +42,8 @@ def test_toggle_never_added_when_field_undefined(): assert fields['ns.r']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] -def test_repair_skips_custom(): +def test_repair_removes_inert_toggle_from_custom(): + # 'custom' has no capabilities → non-reasoning → the toggle would be inert, so it is removed. fields = { **_FIELD_DEF, 'ns.custom': { @@ -51,7 +52,7 @@ def test_repair_skips_custom(): }, } _repair_field_objects(fields, {'custom': {}}) - assert 'extendedThinking' in fields['ns.custom']['properties'] + assert fields['ns.custom']['properties'] == ['llm.cloud.apikey', 'llm.cloud.modelSource'] def test_repair_without_profiles_leaves_toggle_untouched():