diff --git a/src/art_inference/chat_template.py b/src/art_inference/chat_template.py index 20dd21c7e..430ee69f4 100644 --- a/src/art_inference/chat_template.py +++ b/src/art_inference/chat_template.py @@ -1,3 +1,4 @@ +from collections.abc import Sequence import json import re from typing import Any @@ -25,10 +26,496 @@ "reasoning_content and ((preserve_thinking is defined and preserve_thinking is " "true) or loop.index0 > ns.last_user_index)" ) +# These operations infer reasoning from arbitrary assistant content and can +# discard everything before the last or between repeated tags. +# Match the operations, not a model revision or the text of a particular answer. +_QWEN_INLINE_STATEMENTS = ( + "if '' in content", + "set reasoning_content = content.split('')[0].rstrip('\\n').split('')[-1].lstrip('\\n')", + "set content = content.split('')[-1].lstrip('\\n')", + "endif", +) +_QWEN_INLINE_REASONING = re.compile( + r"\s*".join( + r"\{%[-+]?\s*" + re.escape(statement) + r"\s*[-+]?%\}" + for statement in _QWEN_INLINE_STATEMENTS + ) +) + + +def _without_inline_reasoning_parser(template: str) -> str: + if "reasoning_content" not in template or "split" not in template: + return template + from jinja2 import Environment, TemplateSyntaxError, meta, nodes + from jinja2.visitor import NodeTransformer + + # Compare parsed operations, not quote/spacing choices or a template hash. + # Only executable block tokens may be edited; quoted/raw/comment data stays. + class WithoutWhitespace(NodeTransformer): + def visit_Output(self, node: nodes.Output, *args: Any, **kwargs: Any): + if all( + isinstance(child, nodes.TemplateData) and not child.data.strip() + for child in node.nodes + ): + return None + return node + + env = Environment() + # This environment only analyzes source. Treat even built-in globals as + # external bindings rather than hiding them from Jinja's scope analysis. + env.globals.clear() + + def operations(text: str): + return WithoutWhitespace().visit(env.parse(text)).body + + parser = "".join("{% " + statement + " %}" for statement in _QWEN_INLINE_STATEMENTS) + operation = operations(parser) + structured_operation = operations( + "{% if message.reasoning_content is string %}" + "{% set reasoning_content = message.reasoning_content %}{% else %}" + + parser + + "{% endif %}" + )[0] + normalized = re.sub(r"\r\n?", "\n", template) + offsets = [ + i + for i, char in enumerate(template) + if not (char == "\n" and i and template[i - 1] == "\r") + ] + [len(template)] + blocks: list[tuple[int, int, int, int]] = [] + cursor = 0 + opening = None + try: + for _, kind, value in env.lex(template): + start = normalized.find(value, cursor) + if start < 0 or normalized[cursor:start].strip(): + return template # Lexer normalization could not be source-joined. + if kind == "block_begin": + opening = offsets[start], offsets[start + len(value)] + elif kind == "block_end" and opening is not None: + # The lexer can include whitespace following a right-trim tag. + end = start + value.index("%}") + 2 + blocks.append((*opening, offsets[start], offsets[end])) + opening = None + cursor = start + len(value) + except TemplateSyntaxError: + return template # Leave invalid templates to their existing renderer. + edits: dict[tuple[int, int], str] = {} + for index, (start, _, _, _) in enumerate(blocks): + selected = blocks[index : index + 4] + if len(selected) != 4 or "split" not in template[start : selected[-1][3]]: + continue + if any( + template[left[3] : right[0]].strip() + for left, right in zip(selected, selected[1:]) + ): + continue + end = selected[-1][3] + try: + if operations(template[start:end]) == operation: + # The enclosing tags also control unrelated surrounding + # whitespace. Disable the parser without deleting those tags. + _, body_start, body_end, first_end = selected[0] + edits[start, end] = ( + template[start:body_start] + + " if false " + + template[body_end:first_end] + + template[selected[-1][0] : end] + ) + except TemplateSyntaxError: + continue + if not edits: + return template + + def apply_edits() -> str: + result = template + for (start, end), replacement in sorted(edits.items(), reverse=True): + result = result[:start] + replacement + result[end:] + return result + + # Only the content assignment consumed by a recognized parser may lose + # its trim. Independent preview macros/branches have their own bindings. + try: + tree = WithoutWhitespace().visit(env.parse(template)) + except TemplateSyntaxError: + # Optional binding analysis must not undo independently proved parser + # edits when the renderer supports extensions absent from this parser. + result = apply_edits() + if ( + "" in template + and "reasoning_content|trim" in template + and "if not preserve_thinking or message.reasoning_content" not in template + ): + # Keep the pre-existing preservation rewrite when custom syntax + # prevents the narrower binding analysis below. + result = result.replace( + "set content = render_content(message.content, true)|trim", + "set content = (render_content(message.content, true) if preserve_thinking and message.role == 'assistant' else render_content(message.content, true)|trim)", + ) + return result + + # Moving the role test ahead of a renderer is safe only for a local, + # data-only macro under the stock chat renderer. External callbacks and + # unknown macro effects retain the original trim and evaluation order. + # This permits ordinary chat data, built-in tests, integer vision counters + # and the stock raising helper, not custom finalizers or renderer globals. + renderers = [ + node for node in tree.find_all(nodes.Macro) if node.name == "render_content" + ] + if len(renderers) != 1 or renderers[0] not in tree.body: + return apply_edits() + if any(tree.find_all(nodes.Extends)): + return apply_edits() # A parent template can replace the local macro. + renderer = renderers[0] + preceding = tree.body[: tree.body.index(renderer)] + if any( + name.name == "render_content" + for node in preceding + for name in node.find_all(nodes.Name) + ): + return apply_edits() + isolated = nodes.Template([renderer]) + isolated.set_environment(env) + try: + # Use Jinja's own scope analysis, including conditionally bound locals. + global_names = meta.find_undeclared_variables(isolated) + except TemplateSyntaxError: + return apply_edits() + counters = global_names - {"raise_exception", "add_vision_id"} + counter_targets: set[int] = set() + for name in counters: + declarations = [ + node + for node in preceding + if isinstance(node, nodes.Assign) + and node.target == nodes.Name(name, "store") + ] + declaration = env.parse("{% set counter = namespace(value=0) %}").body[0] + assert isinstance(declaration, nodes.Assign) + expected = declaration.node + if len(declarations) != 1 or declarations[0].node != expected: + return apply_edits() + counter_targets.add(id(declarations[0].target)) + inside = {id(node) for node in renderer.find_all(nodes.Node)} + if any( + node.name == name + and id(node) not in inside + and (isinstance(node, nodes.NSRef) or node.ctx == "load") + for node in tree.find_all((nodes.Name, nodes.NSRef)) + ): + return apply_edits() + allowed = tuple( + getattr(nodes, name) + for name in ( + "Add And Assign Call Compare Concat Const For Getattr If Name Not " + "NSRef Operand Or Output TemplateData Test" + ).split() + ) + protected = { + "render_content", + "raise_exception", + "namespace", + "add_vision_id", + *counters, + } + if any( + not isinstance(node, allowed) + # Parameters and their local aliases can refer to the caller's message. + # Only the already-proven private counters may be mutated here. + or isinstance(node, nodes.NSRef) + and node.name not in counters + or isinstance(node, nodes.Call) + and not ( + node.node == nodes.Name("raise_exception", "load") + and len(node.args) == 1 + and isinstance(node.args[0], nodes.Const) + and not node.kwargs + and node.dyn_args is None + and node.dyn_kwargs is None + ) + or isinstance(node, nodes.Test) + and node.name not in {"string", "iterable", "mapping", "none", "undefined"} + for node in renderer.find_all(nodes.Node) + ): + return apply_edits() + if any( + node.name in protected - counters + and node.ctx != "load" + or node.name in counters + and node.ctx != "load" + and id(node) not in counter_targets + for node in tree.find_all(nodes.Name) + ) or any( + isinstance(node, nodes.Import) + and node.target in protected + or isinstance(node, nodes.FromImport) + and any( + (name if isinstance(name, str) else name[1]) in protected + for name in node.names + ) + or isinstance(node, nodes.Macro) + and node is not renderer + and node.name in protected + for node in tree.find_all((nodes.Import, nodes.FromImport, nodes.Macro)) + ): + return apply_edits() + # Macro callables and namespace counters are mutable. A context callback + # can replace either without a lexical reference, even for counter-free macros. + # Admit only the stock chat template's closed data/render operations. + closed = WithoutWhitespace().visit(env.parse(apply_edits())) + try: + external = meta.find_undeclared_variables(closed) + except TemplateSyntaxError: + return apply_edits() + if external.difference( + "messages tools message content reasoning_content preserve_thinking " + "enable_thinking add_generation_prompt add_vision_id raise_exception namespace".split() + ): + return apply_edits() + data_nodes = allowed + tuple( + getattr(nodes, name) + for name in "CondExpr Filter Getitem Keyword Macro Neg Slice Sub Tuple".split() + ) + for node in closed.find_all(nodes.Node): + if not isinstance(node, data_nodes): + return apply_edits() + if isinstance(node, nodes.Call) and not ( + isinstance(node.node, nodes.Name) + and node.node.name in {"render_content", "raise_exception", "namespace"} + or isinstance(node.node, nodes.Getattr) + and node.node.node == nodes.Name("content", "load") + and node.node.attr in {"startswith", "endswith"} + ): + return apply_edits() + if isinstance(node, nodes.Filter) and node.name not in ( + "default items length safe string tojson trim".split() + ): + return apply_edits() + if isinstance(node, nodes.Test) and node.name not in ( + "defined false iterable mapping none string true undefined".split() + ): + return apply_edits() + assignments = list(tree.find_all(nodes.Assign)) + locations = [] + parsed_assignments = [] + for block in blocks: + start, body_start, body_end, end = block + if template[body_start:body_end].split(None, 1)[:1] != ["set"]: + continue + try: + body = env.parse(template[start:end]).body + except TemplateSyntaxError: + continue + if len(body) == 1 and isinstance(body[0], nodes.Assign): + locations.append(block) + parsed_assignments.append(body[0]) + if assignments != parsed_assignments: + return apply_edits() # Do not guess an assignment's source span. + # An AST-equivalent block with comments/data between its tags may not be + # one of the source spans removed above. Join If nodes in source order too. + conditions = [] + for block in blocks: + statement = template[block[1] : block[2]].strip() + keyword = statement.split(None, 1)[:1] + if keyword not in (["if"], ["elif"]): + continue + try: + parsed = env.parse( + "{% if " + statement.split(None, 1)[1] + " %}{% endif %}" + ) + except TemplateSyntaxError: + return apply_edits() + if len(parsed.body) != 1 or not isinstance(parsed.body[0], nodes.If): + return apply_edits() + conditions.append((block, parsed.body[0].test)) + branches = list(tree.find_all(nodes.If)) + if [node.test for node in branches] != [test for _, test in conditions]: + return apply_edits() + edited_parsers = { + id(node) + for node, (block, _) in zip(branches, conditions, strict=True) + if any(start == block[0] for start, _ in edits) + } + selected = set() + shared = set() + + def writes_content(node: nodes.Assign | nodes.AssignBlock) -> bool: + target = node.target + return ( + isinstance(target, nodes.Name) + and target.name == "content" + or any(n.name == "content" for n in target.find_all(nodes.Name)) + ) + + def reads_content(node: nodes.Node, initialized: set[str] | None) -> bool: + # Only a fresh loop-local constant assignment is transparent. Rebinding + # an arbitrary old value can invoke its destructor. Calls, + # filters/tests, loaders, attributes/items, operators and even output + # conversion/finalization may invoke code that observes this scope. + # Keep the trim for unknown nodes instead of enumerating callbacks. + return not ( + initialized is not None + and isinstance(node, nodes.Assign) + and isinstance(node.target, nodes.Name) + and node.target.name not in initialized + and node.target.name != "message" + and isinstance(node.node, nodes.Const) + ) + + def assistant_condition(test: nodes.Node) -> bool | None: + # The rewrite changes assistant content only. Ignore paths proved to + # handle a different role in an ordinary chat message dictionary. + # Every unknown condition retains the original trimmed binding. + if ( + isinstance(test, nodes.Compare) + and test.expr + == nodes.Getattr(nodes.Name("message", "load"), "role", "load") + and len(test.ops) == 1 + and test.ops[0].op in ("eq", "ne") + and isinstance(test.ops[0].expr, nodes.Const) + ): + equal = test.ops[0].expr.value == "assistant" + return equal if test.ops[0].op == "eq" else not equal + return None + + def remember_stores(node: nodes.Node, initialized: set[str] | None) -> None: + if initialized is None: + return + initialized.update( + n.name for n in node.find_all(nodes.Name) if n.ctx == "store" + ) + for child in ( + node, + *node.find_all((nodes.Macro, nodes.Import, nodes.FromImport)), + ): + if isinstance(child, nodes.Macro): + initialized.add(child.name) + elif isinstance(child, nodes.Import): + initialized.add(child.target) + elif isinstance(child, nodes.FromImport): + initialized.update( + name if isinstance(name, str) else name[1] for name in child.names + ) + + def visit( + body: Sequence[nodes.Node], + bindings: set[int], + initialized: set[str] | None = None, + loop_locals: bool = False, + ) -> set[int]: + bindings = bindings.copy() + # Store history belongs to every scope; only loop locals have the + # ownership proof allowing a fresh constant assignment to be ignored. + if initialized is None: + initialized = set() + for node in body: + if isinstance(node, nodes.If): + if (id(node) in edited_parsers and node == operation[0]) or ( + node == structured_operation + and any(id(child) in edited_parsers for child in node.else_) + ): + # The recognized structured/inline reasoning selector is + # one normalization boundary too, preserving its existing + # literal-content contract in either reasoning mode. + if len(bindings) == 1: + selected.update(bindings) + else: + shared.update(bindings) # No unique consumed assignment. + bindings.clear() + remember_stores(node, initialized) + continue + joined = set() + # If does not introduce a Jinja scope. Retain every binding + # reaching the join, including paths that skipped the parser. + for branch in (node, *node.elif_): + condition = assistant_condition(branch.test) + if condition is None: + shared.update(bindings) + if condition is not False: + joined.update( + visit(branch.body, bindings, initialized, loop_locals) + ) + if condition is True: + break + else: + joined.update(visit(node.else_, bindings, initialized, loop_locals)) + bindings = joined + # A role-pruned path may still have initialized a local before + # rebinding message. Freshness follows every syntactic store. + remember_stores(node, initialized) + else: + if reads_content(node, initialized if loop_locals else None): + shared.update(bindings) + replacing_content = bool(bindings) or ("content" in initialized) + remember_stores(node, initialized) + if isinstance(node, nodes.Assign): + if writes_content(node): + bindings = {id(node)} + if replacing_content: + # Publishing a new binding can release an old + # object whose destructor observes the new value. + shared.update(bindings) + else: + # Macro/loop/with/block bodies have independent bindings. + for _, value in node.iter_fields(): + if isinstance(value, list) and all( + isinstance(n, nodes.Node) for n in value + ): + # Jinja initializes loop locals before each body; + # parameters/targets already have arbitrary values. + local_names = ( + { + n.name + for n in ( + node.target, + *node.target.find_all(nodes.Name), + ) + if isinstance(n, nodes.Name) + } + if isinstance(node, nodes.For) and value is node.body + else None + ) + if isinstance(node, nodes.Macro): + local_names = {arg.name for arg in node.args} + elif isinstance(node, nodes.With): + local_names = { + bound.name + for target in node.targets + for bound in (target, *target.find_all(nodes.Name)) + if isinstance(bound, nodes.Name) + } + visit( + value, + set(), + local_names, + isinstance(node, nodes.For) and value is node.body, + ) + if isinstance(node, nodes.AssignBlock) and writes_content(node): + bindings.clear() + return bindings + + visit(tree.body, set()) + trims = [ + env.parse("{% set content = " + content + " %}").body[0] + for content in ( + "render_content(message.content, true)|trim", + "(render_content(message.content, true) if preserve_thinking and message.role == 'assistant' else render_content(message.content, true)|trim)", + ) + ] + for node, (start, body_start, body_end, end) in zip( + assignments, locations, strict=True + ): + if id(node) in selected - shared and node in trims: + edits[start, end] = ( + template[start:body_start] + + " set content = (render_content(message.content, true) if message.role == 'assistant' else render_content(message.content, true)|trim) " + + template[body_end:end] + ) + return apply_edits() def chat_template_with_preserved_thinking(chat_template: object) -> object: - """Preserve prior reasoning by default, while respecting explicit opt-outs.""" + """Preserve structured reasoning without interpreting tags in plain content.""" if isinstance(chat_template, dict): return { name: chat_template_with_preserved_thinking(template) @@ -36,6 +523,9 @@ def chat_template_with_preserved_thinking(chat_template: object) -> object: } if not isinstance(chat_template, str): return chat_template + literal_template = _without_inline_reasoning_parser(chat_template) + inline_parser_removed = literal_template != chat_template + chat_template = literal_template replacements = ( ( _QWEN_DROP_PRIOR_THINKING, @@ -117,10 +607,19 @@ def chat_template_with_preserved_thinking(chat_template: object) -> object: "reasoning_content + '\\n\\n\\n'", "reasoning_content + ('\\n\\n' if preserve_thinking and message.reasoning_content is string and reasoning_content else '\\n\\n\\n')", ) - chat_template = chat_template.replace( - "set content = render_content(message.content, true)|trim", - "set content = (render_content(message.content, true) if preserve_thinking and message.role == 'assistant' else render_content(message.content, true)|trim)", - ) + if ( + not inline_parser_removed + and "{%- set reasoning_content = reasoning_content|trim %}" + in literal_template + ): + # The recognized parser path already preserved its own input. Do + # not apply the legacy trim rewrite without its recognized + # reasoning assignment: an unrelated inline filter can survive + # parser removal and must not activate this on a second call. + chat_template = chat_template.replace( + "set content = render_content(message.content, true)|trim", + "set content = (render_content(message.content, true) if preserve_thinking and message.role == 'assistant' else render_content(message.content, true)|trim)", + ) if "clear_thinking" in chat_template: chat_template = chat_template.replace( "{{ content.strip() }}", diff --git a/tests/unit/test_literal_reasoning_content.py b/tests/unit/test_literal_reasoning_content.py new file mode 100644 index 000000000..d346d25d3 --- /dev/null +++ b/tests/unit/test_literal_reasoning_content.py @@ -0,0 +1,1271 @@ +from copy import deepcopy +import hashlib +from pathlib import Path + +from jinja2 import DictLoader, Environment, pass_context +from jinja2.sandbox import ImmutableSandboxedEnvironment +import pytest + +from art_inference.chat_template import ( + _QWEN_INLINE_REASONING, + _without_inline_reasoning_parser, + chat_template_with_preserved_thinking, + default_chat_template_kwargs_for_template, +) + +_TEMPLATE = ( + Path(__file__).parents[1] / "fixtures/qwen35_preserved_thinking.jinja" +).read_text() +_FIXED = chat_template_with_preserved_thinking(_TEMPLATE) +_USER = {"role": "user", "content": "A public question."} +_LITERALS = ( + "plain answer", + "thoughtanswer", + "prefixliteralsuffix", + "answerliteral", + "prefixmiddlesuffix", + "onetwo", + "nested", + "unclosed", + "unopened", + "", + "", + "\n before café 漢字 🦉 after \n", + "", +) + + +@pytest.mark.parametrize( + "middle", + [ + "{{ ''|peek }}", + "{% if '' is peek %}seen{% endif %}", + "{% include 'preview' %}", + "{% import 'module' as p with context %}{{ p.stored }}", + "{% from 'module' import stored with context %}{{ stored }}", + "{% set observed = probe.value %}", + "{% set observed = probe['value'] %}", + "{{ probe }}", + "{% if probe %}seen{% endif %}", + "{% set observed = probe + 1 %}", + "{% if probe == 1 %}seen{% endif %}", + "{% for item in probe %}seen{% endfor %}", + "{{ 42 }}", + "{% set probe = none %}", + ], +) +@pytest.mark.parametrize("content", [" answer ", " beforexafter "]) +def test_implicit_context_consumers_keep_original_shared_trim(middle, content): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + trim = "{% set content = render_content(message.content, true)|trim %}" + template = ( + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + "{% set probe = bind() %}" + trim + middle + parser + "[{{ content }}]" + ) + seen = [] + + class Probe: + def __init__(self, context): + self.context = context + + def observe(self): + value = self.context["content"] + seen.append(value) + return value + + @property + def value(self): + return self.observe() + + def __getitem__(self, key): + return self.observe() + + def __str__(self): + return self.observe() + + def __bool__(self): + self.observe() + return True + + def __add__(self, other): + return self.observe() + + def __eq__(self, other): + self.observe() + return True + + def __iter__(self): + self.observe() + return iter([1]) + + def __del__(self): + if middle == "{% set probe = none %}": + self.observe() + + @pass_context + def peek(context, value): + seen.append(context["content"]) + return context["content"] + + @pass_context + def finalize(context, value): + if value == 42: + seen.append(context["content"]) + return value + + env = ImmutableSandboxedEnvironment( + loader=DictLoader( + {"preview": "{{ content }}", "module": "{% set stored=content %}"} + ), + finalize=finalize, + ) + bind = pass_context(lambda context: Probe(context)) + env.filters["peek"] = env.tests["peek"] = peek + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + assert not _QWEN_INLINE_REASONING.search(fixed) + message = {"role": "assistant", "content": content} + # Removing the destructive parser is intentional; all surrounding consumers + # must continue observing the same original trimmed binding. + expected = env.from_string(template.replace(parser, "")).render( + message=message, bind=bind + ) + before = seen[:] + seen.clear() + assert env.from_string(fixed).render(message=message, bind=bind) == expected + assert seen == before + assert all(value == content.strip() for value in seen) + assert trim in fixed + + +def test_role_guard_is_not_proof_after_message_reassignment(): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + template = ( + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + + trim + + "{% set message = none %}{% if message.role == 'assistant' %}" + + match.group() + + "{% else %}[{{ content }}]{% endif %}" + ) + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + env = ImmutableSandboxedEnvironment() + message = {"role": "assistant", "content": " answer "} + assert env.from_string(fixed).render(message=message) == "[answer]" + assert trim in fixed + + +@pytest.mark.parametrize("prior_binding", [False, True]) +def test_only_fresh_loop_local_initialization_is_transparent(prior_binding): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + template = ( + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + "{% for message in messages %}" + + ("{% set probe = make_probe() %}" if prior_binding else "") + + trim + + "{% set probe = none %}" + + match.group() + + "[{{ content }}]{% endfor %}" + ) + destroyed = [] + + class Probe: + def __del__(self): + destroyed.append(True) + + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + content = " beforexafter " + env = ImmutableSandboxedEnvironment() + actual = env.from_string(fixed).render( + messages=[{"role": "assistant", "content": content}], make_probe=Probe + ) + assert actual == "[" + (content.strip() if prior_binding else content) + "]" + assert bool(destroyed) is prior_binding + assert (trim in fixed) is prior_binding + + +@pytest.mark.parametrize( + "branch", + [ + "{% if message.role == 'user' %}BODY{% endif %}", + "{% if message.role == 'assistant' %}{% else %}BODY{% endif %}", + "{% if message.role == 'system' %}{% elif message.role == 'user' %}BODY{% endif %}", + ], +) +def test_role_pruning_keeps_prior_store_history(branch): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + seen = [] + + class Probe: + def __init__(self, reader): + self.reader = reader + + def __del__(self): + seen.append(self.reader()) + + trim = "{% set content = render_content(message.content, true)|trim %}" + template = ( + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + "{% for message in messages %}" + "{% macro read_content() %}{{ content }}{% endmacro %}" + + branch.replace( + "BODY", + "{% set probe = make_probe(read_content) %}" + "{% set message = {'role': 'assistant', 'content': message.content} %}", + ) + + trim + + "{% set probe = none %}" + + match.group() + + "{% endfor %}{{ seen|join('|') }}" + ) + env = ImmutableSandboxedEnvironment() + kwargs = dict( + messages=[{"role": "user", "content": " answer "}], + make_probe=Probe, + seen=seen, + ) + assert env.from_string(template).render(**kwargs) == "answer" + seen.clear() + fixed = _without_inline_reasoning_parser(template) + assert env.from_string(fixed).render(**kwargs) == "answer" + assert seen == ["answer"] + assert trim in fixed + + +@pytest.mark.parametrize( + "prefix", + [ + "", + "{% macro render_content(content, count) %}{{ mutate(content, count) }}{% endmacro %}", + "{% macro render_content(content, count) %}{{ content|mutate }}{% endmacro %}", + "{% macro render_content(content, count) %}{% if content is mutate %}{{ content }}{% endif %}{% endmacro %}", + "{% macro render_content(content, count) %}{{ probe.value }}{% endmacro %}", + "{% set dict = probe %}{% macro render_content(content, count) %}{{ dict.value }}{% endmacro %}", + "{% macro render_content(content, count) %}{{ probe.value }}{% if false %}{% set probe = 0 %}{% endif %}{% endmacro %}", + "{% macro render_content(content, count) %}{{ probe['value'] }}{% endmacro %}", + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}{% set render_content = mutate %}", + ], +) +@pytest.mark.parametrize("mutate_role", [False, True]) +def test_unknown_renderer_effects_keep_trim_and_call_order(prefix, mutate_role): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + source = ( + prefix + + trim + + "{% if message.role == 'assistant' %}" + + match.group() + + "{% endif %}[{{ content }}]" + ) + env = ImmutableSandboxedEnvironment() + + def render(template): + message = {"role": "assistant", "content": " beforexafter "} + calls = [] + + def mutate(value, count=None): + calls.append(message["role"]) + if mutate_role: + message["role"] = "user" + return value + + class Probe: + @property + def value(self): + return mutate(message["content"]) + + def __getitem__(self, key): + return self.value + + env.filters["mutate"] = mutate + env.tests["mutate"] = lambda value: bool(mutate(value)) + result = env.from_string(template).render( + message=message, render_content=mutate, mutate=mutate, probe=Probe() + ) + return result, calls + + fixed = _without_inline_reasoning_parser(source) + assert not _QWEN_INLINE_REASONING.search(fixed) + assert trim in fixed + assert render(fixed) == render(source.replace(match.group(), "")) + assert render(fixed) == ("[beforexafter]", ["assistant"]) + assert chat_template_with_preserved_thinking(source) == fixed + assert chat_template_with_preserved_thinking(fixed) == fixed + + +def test_renderer_declared_after_use_does_not_prove_role_stability(): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + source = ( + trim + + "{% if message.role == 'assistant' %}" + + match.group() + + "{% else %}[{{ content }}]{% endif %}" + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + ) + env = ImmutableSandboxedEnvironment() + + def render(template): + message = {"role": "assistant", "content": " answer "} + + def render_content(value, count): + message["role"] = "user" + return value + + return env.from_string(template).render( + message=message, render_content=render_content + ) + + fixed = _without_inline_reasoning_parser(source) + assert trim in fixed + assert render(source) == render(fixed) == "[answer]" + + +def test_inherited_renderer_does_not_prove_role_stability(): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + source = ( + "{% extends 'parent' %}" + "{% macro render_content(content,count) %}{{ content }}{% endmacro %}" + "{% block body %}" + + trim + + "{% if message.role == 'assistant' %}" + + match.group() + + "{% endif %}[{{ content }}]{% endblock %}" + ) + parent = ( + "{% macro render_content(content,count) %}{{ mutate(content,count) }}{% endmacro %}" + "{% block body %}{% endblock %}" + ) + message = {"role": "assistant", "content": " answer "} + + def mutate(value, count): + message["role"] = "user" + return value + + fixed = _without_inline_reasoning_parser(source) + assert trim in fixed + assert ( + ImmutableSandboxedEnvironment(loader=DictLoader({"parent": parent})) + .from_string(fixed) + .render(message=message, mutate=mutate) + == "[answer]" + ) + + +@pytest.mark.parametrize( + "mutation", + [ + "{% set content.role = 'user' %}", + "{% set alias = content %}{% set alias.role = 'user' %}", + ], +) +@pytest.mark.parametrize( + "text", [" answer ", " beforeliteralafter "] +) +def test_renderer_cannot_mutate_parameter_namespace_aliases(mutation, text): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + source = ( + "{% macro render_content(content,count) %}" + + mutation + + "{{ content.text }}{% endmacro %}" + + "{% set message = namespace(role=message.role, text=message.content) %}" + + "{% set message.content = message %}" + + trim + + "{% if message.role == 'assistant' %}" + + match.group() + + "{% else %}[{{ content }}]{% endif %}" + ) + env = ImmutableSandboxedEnvironment() + kwargs = {"message": {"role": "assistant", "content": text}} + original = env.from_string(source).render(**kwargs) + fixed = _without_inline_reasoning_parser(source) + assert original == "[" + text.strip() + "]" + assert env.from_string(fixed).render(**kwargs) == original + assert trim in fixed + assert not _QWEN_INLINE_REASONING.search(fixed) + assert _without_inline_reasoning_parser(fixed) == fixed + + +@pytest.mark.parametrize("text", [" answer ", " literalafter "]) +def test_private_renderer_counter_cannot_be_rebound_to_message(text): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + source = ( + "{% set counter = namespace(value=0) %}" + "{% macro render_content(content,count) %}" + "{% set counter.role = 'user' %}{{ content }}{% endmacro %}" + "{% set message = namespace(role=message.role, content=message.content) %}" + "{% set counter = message %}" + + trim + + "{% if message.role == 'assistant' %}" + + match.group() + + "{% else %}[{{ content }}]{% endif %}" + ) + env = ImmutableSandboxedEnvironment() + kwargs = {"message": {"role": "assistant", "content": text}} + original = env.from_string(source).render(**kwargs) + fixed = _without_inline_reasoning_parser(source) + assert original == "[" + text.strip() + "]" + assert env.from_string(fixed).render(**kwargs) == original + assert trim in fixed + assert not _QWEN_INLINE_REASONING.search(fixed) + assert _without_inline_reasoning_parser(fixed) == fixed + + +@pytest.mark.parametrize("environment", [Environment, ImmutableSandboxedEnvironment]) +@pytest.mark.parametrize( + "middle", + [ + "{{ inject() }}", + "{{ ''|inject }}", + "{% if '' is inject %}{% endif %}", + "{% include 'setup' %}", + ], +) +def test_counter_free_renderer_rejects_implicit_macro_mutation(environment, middle): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + source = ( + "{% macro render_content(content,count) %}{{ content }}{% endmacro %}" + + middle + + trim + + "{% if message.role == 'assistant' %}" + + match.group() + + "{% endif %}[{{ content }}]" + ) + + def render(template): + message = {"role": "assistant", "content": " answer "} + calls = [] + + @pass_context + def inject(context, value=None): + calls.append("inject") + + def replacement(content, count): + calls.append("render:" + message["role"]) + message["role"] = "user" + return content + + context["render_content"]._func = replacement + return "" + + env = environment(loader=DictLoader({"setup": "{{ inject() }}"})) + env.filters["inject"] = env.tests["inject"] = inject + output = env.from_string(template).render(message=message, inject=inject) + return output, calls, message["role"] + + expected = "[answer]", ["inject", "render:assistant"], "user" + fixed = _without_inline_reasoning_parser(source) + assert render(source) == render(source.replace(match.group(), "")) == expected + assert render(fixed) == expected + assert trim in fixed + assert not _QWEN_INLINE_REASONING.search(fixed) + assert _without_inline_reasoning_parser(fixed) == fixed + + +@pytest.mark.parametrize( + "middle", + [ + "{% include 'setup' %}", + "{{ inject() }}", + "{{ ''|inject }}", + "{% if '' is inject %}{% endif %}", + ], +) +def test_counter_renderer_rejects_implicit_context_exports(middle): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + trim = "{% set content = render_content(message.content, true)|trim %}" + source = ( + "{% set counter=namespace(value=0) %}" + "{% macro render_content(content,count) %}{% set counter.value=counter.value+1 %}{{ content }}{% endmacro %}" + + middle + + trim + + "{% if message.role == 'assistant' %}" + + match.group() + + "{% endif %}[{{ content }}]" + ) + message = {"role": "assistant", "content": " answer "} + + class Probe: + def __add__(self, other): + message["role"] = "user" + return 1 + + @pass_context + def inject(context, value=None): + context["counter"]["value"] = Probe() + return "" + + env = ImmutableSandboxedEnvironment( + loader=DictLoader({"setup": "{% set counter.value=probe %}"}) + ) + env.filters["inject"] = env.tests["inject"] = inject + fixed = _without_inline_reasoning_parser(source) + assert trim in fixed + assert ( + env.from_string(fixed).render(message=message, inject=inject, probe=Probe()) + == "[answer]" + ) + + +@pytest.mark.parametrize("scope", ["top", "loop", "macro", "with"]) +def test_replacing_content_keeps_trim_for_its_own_observers(scope): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + seen = [] + + class Probe: + def __init__(self, reader): + self.reader = reader + + def __del__(self): + seen.append(self.reader()) + + trim = "{% set content = render_content(message.content,true)|trim %}" + body = ( + "{% macro read_content() %}{{ content }}{% endmacro %}" + "{% if message.role == 'user' %}{% set content = make_probe(read_content) %}" + "{% set message = {'role': 'assistant', 'content': message.content} %}{% endif %}" + + trim + + match.group() + ) + if scope == "loop": + body = "{% for message in messages %}" + body + "{% endfor %}" + elif scope == "macro": + body = "{% macro run(message) %}" + body + "{% endmacro %}{{ run(message) }}" + elif scope == "with": + body = "{% with message=message %}" + body + "{% endwith %}" + source = ( + "{% macro render_content(content,count) %}{{ content }}{% endmacro %}" + + body + + "{{ seen|join('|') }}" + ) + fixed = _without_inline_reasoning_parser(source) + assert trim in fixed + assert ( + ImmutableSandboxedEnvironment() + .from_string(fixed) + .render( + messages=[{"role": "user", "content": " answer "}], + message={"role": "user", "content": " answer "}, + make_probe=Probe, + seen=seen, + ) + == "answer" + ) + + +@pytest.mark.parametrize("trim_blocks,lstrip_blocks", [(False, False), (True, True)]) +@pytest.mark.parametrize( + "left,right", [("", ""), ("-", ""), ("", "-"), ("-", "-"), ("+", "+")] +) +@pytest.mark.parametrize("newline", ["\n", "\r\n", "\r"]) +def test_disabling_inline_parser_preserves_outer_whitespace( + trim_blocks, lstrip_blocks, left, right, newline +): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group() + operation = "{%" + left + operation[3:] + operation = operation[:-2].rstrip("-+") + right + "%}" + template = ("HEADER \n\t" + operation + "\n \tTAIL{{ content }}").replace( + "\n", newline + ) + env = ImmutableSandboxedEnvironment( + trim_blocks=trim_blocks, lstrip_blocks=lstrip_blocks + ) + fixed = _without_inline_reasoning_parser(template) + ordinary = env.from_string(template).render(content="plain answer") + assert env.from_string(fixed).render(content="plain answer") == ordinary + literal = "prefixliteralsuffix" + assert env.from_string(fixed).render(content=literal) == ordinary.replace( + "plain answer", literal + ) + assert _without_inline_reasoning_parser(fixed) == fixed + + +def _render(template, messages, **kwargs): + def refuse(message): + raise ValueError(message) + + env = ImmutableSandboxedEnvironment( + trim_blocks=True, lstrip_blocks=True, extensions=["jinja2.ext.loopcontrols"] + ) + return env.from_string(template).render( + messages=messages, raise_exception=refuse, **kwargs + ) + + +@pytest.mark.parametrize("thinking", [False, True]) +@pytest.mark.parametrize("preserve", [False, True]) +@pytest.mark.parametrize("content", _LITERALS) +def test_plain_content_is_literal_in_every_mode(content, thinking, preserve): + messages = [_USER, {"role": "assistant", "content": content}] + before = deepcopy(messages) + rendered = _render( + _FIXED, messages, enable_thinking=thinking, preserve_thinking=preserve + ) + assert rendered.endswith(content + "<|im_end|>\n") + assert messages == before + # Changing the next turn's thinking mode never reinterprets history. + assert rendered == _render( + _FIXED, messages, enable_thinking=not thinking, preserve_thinking=preserve + ) + + +@pytest.mark.parametrize("thinking", [False, True]) +@pytest.mark.parametrize("preserve", [False, True]) +@pytest.mark.parametrize( + "reasoning", [None, "", "reasoned\n", "literal reasoning text\n"] +) +def test_structured_reasoning_and_explicit_empty_field_keep_existing_behavior( + thinking, preserve, reasoning +): + messages = [ + _USER, + {"role": "assistant", "content": "answer", "reasoning_content": reasoning}, + {"role": "user", "content": "next"}, + ] + kwargs = dict(enable_thinking=thinking, preserve_thinking=preserve) + before = deepcopy(messages) + assert _render(_FIXED, messages, **kwargs) == _render(_TEMPLATE, messages, **kwargs) + assert messages == before + rendered = _render(_FIXED, messages, **kwargs) + if reasoning: + assert (reasoning in rendered) == preserve + + +@pytest.mark.parametrize("thinking", [False, True]) +@pytest.mark.parametrize("preserve", [False, True]) +def test_proven_legacy_encoding_uses_existing_structured_fields(thinking, preserve): + # This fixture declares the old encoding. The renderer cannot infer that + # declaration from an indistinguishable literal string in plain content. + legacy = {"role": "assistant", "content": "\nthought\n\n\nanswer"} + structured = { + "role": "assistant", + "reasoning_content": "thought\n", + "content": "answer", + } + kwargs = dict(enable_thinking=thinking, preserve_thinking=preserve) + later = {"role": "user", "content": "next"} + assert _render(_FIXED, [_USER, structured, later], **kwargs) == _render( + _TEMPLATE, [_USER, legacy, later], **kwargs + ) + assert legacy["content"] in _render(_FIXED, [_USER, legacy], **kwargs) + + +@pytest.mark.parametrize("thinking", [False, True]) +@pytest.mark.parametrize("preserve", [False, True]) +@pytest.mark.parametrize("content", [None, "", "beforeliteralafter"]) +def test_tool_call_and_continuation_keep_content_and_arguments( + thinking, preserve, content +): + assistant = { + "role": "assistant", + "content": content, + "tool_calls": [ + {"function": {"name": "lookup", "arguments": {"q": ""}}} + ], + } + messages = [_USER, assistant] + before = deepcopy(messages) + kwargs = dict(enable_thinking=thinking, preserve_thinking=preserve) + rendered = _render(_FIXED, messages, **kwargs) + assert "" in rendered + assert "\n\n" in rendered + if content: + assert content in rendered + continued = _render( + _FIXED, + [ + *messages, + {"role": "tool", "content": "result"}, + {"role": "assistant", "content": "nextliteralanswer"}, + ], + **kwargs, + ) + assert continued.startswith(rendered) + assert messages == before + + +@pytest.mark.parametrize("thinking", [False, True]) +@pytest.mark.parametrize("preserve", [False, True]) +def test_generation_prompt_and_preserved_history_prefix_are_stable(thinking, preserve): + kwargs = dict( + enable_thinking=thinking, preserve_thinking=preserve, add_generation_prompt=True + ) + assert _render(_FIXED, [_USER], **kwargs) == _render(_TEMPLATE, [_USER], **kwargs) + messages = [ + _USER, + {"role": "assistant", "content": "plainliteraltail"}, + ] + completed = _render(_FIXED, messages, preserve_thinking=preserve) + continuation = _render(_FIXED, [*messages, _USER], **kwargs) + if preserve: + assert continuation.startswith(completed) + else: + # Explicit opt-out still removes the previous turn's reasoning scaffold; + # the visible body is unchanged, not a newly promised full-token prefix. + assert messages[-1]["content"] + "<|im_end|>\n" in continuation + assert not continuation.startswith(completed) + + +def test_actual_template_operation_and_public_e2ac_shaped_regression(): + assert ( + hashlib.sha256(_TEMPLATE.encode()).hexdigest() + == "098047d425a6673b1fe1a82a197a481616e53a283beaa8cb76cbb74d38ca6644" + ) + # Public text with the captured branch's shape; no private text or IDs. + prefix = "P" * 4149 + body = prefix + "\n" + "R" * 747 + "\n\n\n" + "A" * 1161 + messages = [_USER, {"role": "assistant", "content": body}] + original = _render( + _TEMPLATE, messages, enable_thinking=False, preserve_thinking=True + ) + assert prefix not in original + assert body in _render( + _FIXED, messages, enable_thinking=False, preserve_thinking=True + ) + + +def test_configuration_is_idempotent_and_does_not_change_defaults_or_other_templates(): + assert _FIXED != _TEMPLATE + assert chat_template_with_preserved_thinking(_FIXED) == _FIXED + assert default_chat_template_kwargs_for_template( + _FIXED + ) == default_chat_template_kwargs_for_template(_TEMPLATE) + other = "{% for message in messages %}{{ message.content }}{% endfor %}" + assert chat_template_with_preserved_thinking(other) == other + assert chat_template_with_preserved_thinking( + {"default": _TEMPLATE, "other": other} + ) == {"default": _FIXED, "other": other} + + +def test_unconfigured_template_receives_the_same_correction(): + # Reverse the prior preservation-only rewrite of this public fixture. + raw = ( + _TEMPLATE.replace( + "{%- set preserve_thinking = preserve_thinking | default(true) -%}", "" + ) + .replace( + "(render_content(message.content, true) if preserve_thinking and message.role == 'assistant' else render_content(message.content, true)|trim)", + "render_content(message.content, true)|trim", + ) + .replace( + "{%- if not preserve_thinking or message.reasoning_content is not string %}{%- set reasoning_content = reasoning_content|trim %}{%- endif %}", + "{%- set reasoning_content = reasoning_content|trim %}", + ) + .replace( + "('\\n\\n' if preserve_thinking and message.reasoning_content is string and reasoning_content else '\\n\\n\\n')", + "'\\n\\n\\n'", + ) + ) + assert chat_template_with_preserved_thinking(raw) == _FIXED + + +@pytest.mark.parametrize("indirect", [False, True]) +@pytest.mark.parametrize("content", [" answer ", " beforexafter "]) +def test_macro_capture_keeps_shared_content_trim(indirect, content): + parser = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert parser is not None + template = ( + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + "{% macro preview() %}[{{ content }}]{% endmacro %}" + "{% macro wrapper() %}{{ preview() }}{% endmacro %}" + "{% set content = render_content(message.content, true)|trim %}" + + ("{{ wrapper() }}" if indirect else "{{ preview() }}") + + parser.group() + + "[{{ content }}]" + ) + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + env = ImmutableSandboxedEnvironment(trim_blocks=True, lstrip_blocks=True) + assert ( + env.from_string(fixed).render(message={"role": "assistant", "content": content}) + == "[" + content.strip() + "][" + content.strip() + "]" + ) + assert chat_template_with_preserved_thinking(fixed) == fixed + + +@pytest.mark.parametrize("preserve", [False, True]) +def test_extension_fallback_keeps_prior_structured_content_whitespace(preserve): + template = "{% for item in [1] %}{% break %}{% endfor %}" + _TEMPLATE.replace( + "(render_content(message.content, true) if preserve_thinking and message.role == 'assistant' else render_content(message.content, true)|trim)", + "render_content(message.content, true)|trim", + ).replace( + "{%- if not preserve_thinking or message.reasoning_content is not string %}{%- set reasoning_content = reasoning_content|trim %}{%- endif %}", + "{%- set reasoning_content = reasoning_content|trim %}", + ) + fixed = chat_template_with_preserved_thinking(template) + rendered = _render( + fixed, + [ + _USER, + { + "role": "assistant", + "content": " answer ", + "reasoning_content": "reasoned\n", + }, + ], + enable_thinking=False, + preserve_thinking=preserve, + ) + assert rendered.endswith((" answer " if preserve else "answer") + "<|im_end|>\n") + assert chat_template_with_preserved_thinking(fixed) == fixed + + +@pytest.mark.parametrize("wrapper", [("{% raw %}", "{% endraw %}"), ("{#", "#}")]) +def test_inline_operation_as_raw_or_comment_text_is_not_rewritten(wrapper): + from art_inference.chat_template import _QWEN_INLINE_REASONING + + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group() + template = wrapper[0] + operation + wrapper[1] + assert chat_template_with_preserved_thinking(template) == template + + +def test_other_structured_reasoning_condition_is_not_rewritten(): + gate = "{% if preserve_thinking and message.role == 'assistant' %}{{ message.reasoning_content }}{% endif %}" + template = _TEMPLATE + "{% for message in messages %}" + gate + "{% endfor %}" + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + assert gate in fixed + messages = [ + _USER, + {"role": "assistant", "content": "answer", "reasoning_content": "prior reason"}, + _USER, + ] + assert "prior reason" not in _render(fixed, messages, preserve_thinking=False) + + +def test_inline_operation_inside_quoted_expression_is_literal(): + from art_inference.chat_template import _QWEN_INLINE_REASONING + + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group().replace("\n", " ") + template = '{{ "' + operation + '" }}' + fixed = chat_template_with_preserved_thinking(template) + assert fixed == template + assert "set reasoning_content = content.split" in _render(fixed, []) + + +@pytest.mark.parametrize( + "wrapper", [("{% raw %}", "{% endraw %}"), ("{#", "#}"), ('{{ "', '" }}')] +) +def test_mixed_executable_and_literal_operations_only_changes_executable(wrapper): + from art_inference.chat_template import _QWEN_INLINE_REASONING + + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group().replace("\n", " ") + literal = wrapper[0] + operation + wrapper[1] + template = _TEMPLATE + literal + fixed = chat_template_with_preserved_thinking(template) + assert fixed == _FIXED + literal + assert "headliteraltail" in _render( + fixed, + [_USER, {"role": "assistant", "content": "headliteraltail"}], + ) + + +@pytest.mark.parametrize("newline", ["\n", "\r\n", "\r"]) +@pytest.mark.parametrize("prefix", ["comment", "data"]) +def test_newline_lexing_preserves_literal_content(newline, prefix): + intro = ( + "{# public\nmultiline comment #}\n" + if prefix == "comment" + else "public\nheader\n" + ) + template = (intro + _TEMPLATE).replace("\n", newline) + content = "prefixliteralsuffix" + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + assert content in _render( + fixed, + [_USER, {"role": "assistant", "content": content}], + enable_thinking=False, + preserve_thinking=True, + ) + assert fixed.startswith(intro.replace("\n", newline)) + assert not _QWEN_INLINE_REASONING.search(fixed) + assert chat_template_with_preserved_thinking(fixed) == fixed + + +@pytest.mark.parametrize("newline", ["\r\n", "\r"]) +@pytest.mark.parametrize("wrapper", ["comment", "raw", "quoted"]) +def test_newline_parser_spelling_in_nonexecutable_token_unchanged(newline, wrapper): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group().replace("\n", newline) + if wrapper == "comment": + template = "{#" + newline + operation + newline + "#}" + elif wrapper == "raw": + template = "{% raw %}" + newline + operation + newline + "{% endraw %}" + else: + template = '{{ "' + operation + '" }}' + assert _without_inline_reasoning_parser(template) == template + + +@pytest.mark.parametrize("spelling", ["double_quotes", "spacing", "parentheses"]) +def test_equivalent_inline_operations_preserve_literal_and_structured_fields(spelling): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group() + if spelling == "double_quotes": + operation = operation.replace("'", '"') + elif spelling == "spacing": + operation = operation.replace("content.split", "content . split").replace( + "[0]", "[ 0 ]" + ) + else: + operation = operation.replace( + "if '' in content", "if ('' in content)" + ) + template = _TEMPLATE[: match.start()] + operation + _TEMPLATE[match.end() :] + assert template != _TEMPLATE + fixed = chat_template_with_preserved_thinking(template) + for content in _LITERALS: + for reasoning in (None, "", "explicit structured reasoning\n"): + messages = [_USER, {"role": "assistant", "content": content}] + if reasoning is not None: + messages[-1]["reasoning_content"] = reasoning + for preserve in (False, True): + kwargs = dict(enable_thinking=False, preserve_thinking=preserve) + assert _render(fixed, messages, **kwargs) == _render( + _FIXED, messages, **kwargs + ) + assert chat_template_with_preserved_thinking(fixed) == fixed + + +@pytest.mark.parametrize("wrapper", ["raw", "comment", "quoted"]) +def test_equivalent_operation_as_literal_data_is_not_edited(wrapper): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group().replace("'", '"') + if wrapper == "raw": + literal = "{% raw %}" + operation + "{% endraw %}" + elif wrapper == "comment": + literal = "{#" + operation + "#}" + else: + literal = "{{ '" + operation + "' }}" + assert _without_inline_reasoning_parser(literal) == literal + assert isinstance(_FIXED, str) + assert _without_inline_reasoning_parser(_TEMPLATE + literal) == _FIXED + literal + + +@pytest.mark.parametrize("change", ["different_split", "side_effect", "different_gate"]) +def test_distinct_custom_content_operations_are_not_inferred(change): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + operation = match.group() + if change == "different_split": + operation = operation.replace( + "content.split('')[-1]", "content.split('')[0]" + ) + elif change == "side_effect": + operation = operation.replace( + "{%- endif %}", "{%- set other = content %}{%- endif %}" + ) + else: + operation = operation.replace( + "if '' in content", "if custom and '' in content" + ) + assert operation != match.group() + assert _without_inline_reasoning_parser(operation) == operation + + +@pytest.mark.parametrize("separator", ["\n", " ", "\r\n"]) +@pytest.mark.parametrize("quoted", [False, True]) +def test_plain_block_whitespace_keeps_prior_inline_parser_coverage(separator, quoted): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + # The prior regex admitted whitespace-only separators without trim dashes. + operation = match.group().replace("{%-", "{%").replace("-%}", "%}") + operation = operation.replace("\n", separator) + if quoted: + operation = operation.replace("'", '"') + template = _TEMPLATE[: match.start()] + operation + _TEMPLATE[match.end() :] + fixed = chat_template_with_preserved_thinking(template) + content = "HEADliteralTAIL" + assert content in _render(fixed, [_USER, {"role": "assistant", "content": content}]) + assert chat_template_with_preserved_thinking(fixed) == fixed + + +@pytest.mark.parametrize("preserve", [False, True]) +@pytest.mark.parametrize("newline", ["\n", "\r\n"]) +@pytest.mark.parametrize("layout", ["macros", "same_line", "branches"]) +@pytest.mark.parametrize("inline_structured_reasoning", [False, True]) +def test_custom_macro_calls_keep_trim_while_removing_recognized_parser( + preserve, newline, layout, inline_structured_reasoning +): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + if inline_structured_reasoning: + parser += "{{ reasoning_content|trim }}" + trim = "{% set content = render_content(message.content, true)|trim %}" + render = "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + preview = "{% macro preview(message) %}" + trim + "[{{ content }}]{% endmacro %}" + main = ( + "{% macro answer(message) %}" + trim + parser + "[{{ content }}]{% endmacro %}" + ) + if layout == "branches": + main = ( + "{% macro answer(message) %}{% if message.role == 'assistant' %}" + + trim + + parser + + "[{{ content }}]{% else %}" + + trim + + "[{{ content }}]{% endif %}{% endmacro %}" + ) + separator = "" if layout == "same_line" else newline + template = separator.join( + [render, preview, main, "{{ answer(message) }}|{{ preview(message) }}"] + ) + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + assert preview in fixed + # Other macro calls lack a closed callable-custody proof. Retain their + # original trims while still removing only the recognized parser. + assert trim in fixed + assert not _QWEN_INLINE_REASONING.search(fixed) + env = ImmutableSandboxedEnvironment(trim_blocks=True, lstrip_blocks=True) + kwargs = dict( + message={"role": "assistant", "content": " answer "}, + preserve_thinking=preserve, + ) + assert env.from_string(fixed).render(**kwargs).endswith("[answer]|[answer]") + kwargs["message"]["content"] = " beforeliteralafter " + assert ( + env.from_string(fixed) + .render(**kwargs) + .endswith( + "[beforeliteralafter]|[beforeliteralafter]" + ) + ) + assert chat_template_with_preserved_thinking(fixed) == fixed + assert ( + chat_template_with_preserved_thinking( + chat_template_with_preserved_thinking(fixed) + ) + == fixed + ) + if layout == "branches": + kwargs["message"] = {"role": "user", "content": " answer "} + assert env.from_string(fixed).render(**kwargs).endswith("[answer]|[answer]") + + +@pytest.mark.parametrize("preserve", [False, True]) +@pytest.mark.parametrize("raw", [False, True]) +def test_qwen_content_preview_macro_is_unchanged(preserve, raw): + preview = "{% macro preview(message) %}{% set content = render_content(message.content, true)|trim %}[{{ content }}]{% endmacro %}" + body = _TEMPLATE + if raw: + body = body.replace( + "{%- if not preserve_thinking or message.reasoning_content is not string %}{%- set reasoning_content = reasoning_content|trim %}{%- endif %}", + "{%- set reasoning_content = reasoning_content|trim %}", + ) + template = preview + body + "{{ preview(messages[-1]) }}" + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + assert preview in fixed + content = " beforeliteralafter " + rendered = _render( + fixed, + [_USER, {"role": "assistant", "content": content}], + preserve_thinking=preserve, + ) + # The extra renderer reference before its declaration prevents the closed + # stock-template proof. Keep its existing preservation/trim behavior. + assert (content if preserve else content.strip()) + "<|im_end|>\n" in rendered + assert rendered.endswith("[beforeliteralafter]") + + +@pytest.mark.parametrize("shadow", ["assignment", "conditional", "scope"]) +def test_content_trim_requires_a_proven_local_binding(shadow): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + trim = "{% set content = render_content(message.content, true)|trim %}" + render = "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + if shadow == "assignment": + template = render + trim + "{% set content = 'replacement' %}" + parser + elif shadow == "conditional": + template = ( + render + + trim + + "{% if custom %}{% set content = 'replacement' %}{% endif %}" + + parser + ) + else: + template = ( + render + + trim + + "{% macro nested() %}" + + parser + + "{{ content }}{% endmacro %}{{ nested() }}" + ) + fixed = _without_inline_reasoning_parser(template) + assert trim in fixed + assert not _QWEN_INLINE_REASONING.search(fixed) + assert _without_inline_reasoning_parser(fixed) == fixed + + +@pytest.mark.parametrize("extension", ["loopcontrols", "generation"]) +def test_optional_trim_binding_parse_keeps_proven_parser_removal(extension): + from jinja2 import nodes + from jinja2.ext import Extension + + class Generation(Extension): + tags = {"generation"} + + def parse(self, parser): + next(parser.stream) + body = parser.parse_statements(["name:endgeneration"], drop_needle=True) + return nodes.Scope(body) + + env = ImmutableSandboxedEnvironment( + extensions=["jinja2.ext.loopcontrols", Generation], + trim_blocks=True, + lstrip_blocks=True, + ) + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + prefix = ( + "{% for item in [1] %}{% break %}{% endfor %}" + if extension == "loopcontrols" + else "{% generation %}PUBLIC{% endgeneration %}" + ) + template = prefix + "{% set content = message.content %}" + parser + "{{ content }}" + content = "HEADliteralTAIL" + kwargs = {"message": {"role": "assistant", "content": content}} + assert env.from_string(template).render(**kwargs).endswith("TAIL") + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + assert prefix in fixed + assert env.from_string(fixed).render(**kwargs).endswith(content) + assert chat_template_with_preserved_thinking(fixed) == fixed + + +def test_tuple_scope_content_replacement_retains_trim(): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + seen = [] + + class Probe: + def __init__(self): + self.reader = lambda: "unbound" + + def __del__(self): + seen.append(self.reader()) + + def bind(probe, reader): + probe.reader = reader + return "" + + trim = "{% set content = render_content(message.content,true)|trim %}" + source = ( + "{% macro render_content(content,count) %}{{ content }}{% endmacro %}" + "{% with (content, spare)=make_pair() %}" + "{% macro read_content() %}{{ content }}{% endmacro %}" + "{{ bind(content,read_content) }}" + + trim + + match.group() + + "{% endwith %}{{ seen|join('|') }}" + ) + fixed = _without_inline_reasoning_parser(source) + assert trim in fixed + assert ( + ImmutableSandboxedEnvironment() + .from_string(fixed) + .render( + message={"role": "assistant", "content": " answer "}, + make_pair=lambda: (Probe(), None), + bind=bind, + seen=seen, + ) + == "answer" + ) + assert seen == ["answer"] + + +@pytest.mark.parametrize("consumer", ["output", "alias", "condition", "other_branch"]) +def test_shared_content_preview_prevents_ambiguous_trim_rewrite(consumer): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + trim = "{% set content = render_content(message.content, true)|trim %}" + render = "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + if consumer == "output": + body = "[{{ content }}]" + parser + "[{{ content }}]" + elif consumer == "alias": + body = "{% set preview = content %}" + parser + "[{{ preview }}][{{ content }}]" + elif consumer == "condition": + body = "{% if content %}PREVIEW{% endif %}" + parser + "[{{ content }}]" + else: + body = ( + "{% if preview_only %}[{{ content }}]{% else %}" + + parser + + "[{{ content }}]{% endif %}" + ) + template = render + trim + body + fixed = chat_template_with_preserved_thinking(template) + assert isinstance(fixed, str) + assert trim in fixed + assert not _QWEN_INLINE_REASONING.search(fixed) + env = ImmutableSandboxedEnvironment(trim_blocks=True, lstrip_blocks=True) + kwargs = dict(message={"role": "assistant", "content": " "}, preview_only=True) + assert env.from_string(fixed).render(**kwargs) == env.from_string(template).render( + **kwargs + ) + + +def test_custom_macro_calls_keep_trim_and_unedited_parser(): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + unedited = parser.replace("%}", "%}{# preserve this custom block #}", 1) + trim = "{% set content = render_content(message.content, true)|trim %}" + render = "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + preview = ( + "{% macro preview(message) %}" + + trim + + unedited + + "[{{ content }}]{% endmacro %}" + ) + answer = ( + "{% macro answer(message) %}" + trim + parser + "[{{ content }}]{% endmacro %}" + ) + template = ( + render + preview + answer + "{{ answer(message) }}|{{ preview(message) }}" + ) + fixed = _without_inline_reasoning_parser(template) + assert preview in fixed + assert answer not in fixed + assert trim in fixed + env = ImmutableSandboxedEnvironment(trim_blocks=True, lstrip_blocks=True) + rendered = env.from_string(fixed).render( + message={"role": "assistant", "content": " HEADliteralTAIL "} + ) + assert rendered == "[HEADliteralTAIL]|[TAIL]" diff --git a/tests/unit/test_reasoning_parser_joined_branches.py b/tests/unit/test_reasoning_parser_joined_branches.py new file mode 100644 index 000000000..4c0aa0a01 --- /dev/null +++ b/tests/unit/test_reasoning_parser_joined_branches.py @@ -0,0 +1,89 @@ +from jinja2.sandbox import ImmutableSandboxedEnvironment +import pytest +from test_literal_reasoning_content import _TEMPLATE + +from art_inference.chat_template import ( + _QWEN_INLINE_REASONING, + _without_inline_reasoning_parser, +) + + +@pytest.mark.parametrize("scope", ["top", "macro"]) +@pytest.mark.parametrize("layout", ["if", "elif", "nested", "sequential"]) +@pytest.mark.parametrize("content", [" answer ", " beforexafter "]) +def test_joined_preview_retains_its_original_trim(scope, layout, content): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + trim = "{% set content = render_content(message.content, true)|trim %}" + if layout == "if": + branch = "{% if not preview_only %}" + parser + "{% endif %}" + elif layout == "elif": + branch = ( + "{% if preview_only %}{% elif mode == 'answer' %}" + parser + "{% endif %}" + ) + elif layout == "nested": + branch = ( + "{% if outer %}{% if not preview_only %}" + + parser + + "{% endif %}{% endif %}" + ) + else: + branch = ( + "{% if not preview_only %}" + parser + "{% endif %}" + "{% if mode == 'other' %}" + parser + "{% endif %}" + ) + body = trim + branch + "[{{ content }}]" + if scope == "macro": + body = ( + "{% macro answer(message) %}" + body + "{% endmacro %}{{ answer(message) }}" + ) + template = ( + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + body + ) + fixed = _without_inline_reasoning_parser(template) + env = ImmutableSandboxedEnvironment(trim_blocks=True, lstrip_blocks=True) + kwargs = dict( + message={"role": "assistant", "content": content}, + preview_only=True, + mode="answer", + outer=True, + ) + assert env.from_string(fixed).render(**kwargs) == env.from_string(template).render( + **kwargs + ) + assert env.from_string(fixed).render(**kwargs) == "[" + content.strip() + "]" + assert trim in fixed + assert not _QWEN_INLINE_REASONING.search(fixed) + # The parser path still treats the text as literal; its shared trim stays. + kwargs["preview_only"] = False + assert env.from_string(fixed).render(**kwargs) == "[" + content.strip() + "]" + assert _without_inline_reasoning_parser(fixed) == fixed + + +@pytest.mark.parametrize("mode", ["a", "b", "c"]) +def test_unknown_comparison_keeps_shared_trim_even_when_all_paths_have_parser(mode): + match = _QWEN_INLINE_REASONING.search(_TEMPLATE) + assert match is not None + parser = match.group() + template = ( + "{% macro render_content(content, count) %}{{ content }}{% endmacro %}" + "{% set content = render_content(message.content, true)|trim %}" + "{% if mode == 'a' %}" + + parser + + "{% elif mode == 'b' %}" + + parser + + "{% else %}" + + parser + + "{% endif %}[{{ content }}]" + ) + fixed = _without_inline_reasoning_parser(template) + content = " beforexafter " + env = ImmutableSandboxedEnvironment(trim_blocks=True, lstrip_blocks=True) + assert ( + env.from_string(fixed).render( + mode=mode, message={"role": "assistant", "content": content} + ) + == "[" + content.strip() + "]" + ) + assert _without_inline_reasoning_parser(fixed) == fixed diff --git a/tests/unit/trajectories/test_literal_thinking_off.py b/tests/unit/trajectories/test_literal_thinking_off.py index 16188c733..c30c1a632 100644 --- a/tests/unit/trajectories/test_literal_thinking_off.py +++ b/tests/unit/trajectories/test_literal_thinking_off.py @@ -142,9 +142,12 @@ def test_native_thinking_off_retains_literal_content( ) -> None: history, tokenizer = _history(content=content) original = history.model_dump(mode="python") - # The pre-fix history path misrenders literal content even when later native - # token splicing can recover the terminal output. + # Render the original template without either the general parser correction + # or the old hash-specific workaround to retain the destructive baseline. with monkeypatch.context() as patch: + patch.setattr( + _tokenize, "chat_template_with_preserved_thinking", lambda value: value + ) patch.setattr( _tokenize, "_preserve_literal_thinking_off_content", lambda *args: None ) @@ -162,7 +165,7 @@ def test_native_thinking_off_retains_literal_content( required = tr.TokenFlag.EXACT | tr.TokenFlag.ASSISTANT | tr.TokenFlag.OUTPUT assert all(tokenized.flags[i] & required == required for i in sampled) assert not any(flag & tr.TokenFlag.STOP for flag in tokenized.flags) - assert tokenizer.calls[0][-1]["reasoning_content"] == "" + assert tokenizer.calls[0][-1].get("reasoning_content", "") == "" assert tokenizer.calls[0][-1]["content"] == content assert history.model_dump(mode="python") == original @@ -249,7 +252,8 @@ def test_explicit_empty_reasoning_is_preserved(field: str) -> None: ) == _LITERAL ) - assert tokenizer.calls[0][-1]["reasoning_content"] == "" + assert tokenizer.calls[0][-1].get("reasoning_content", "") == "" + assert _LITERAL in tokenizer.rendered[0] assert history.model_dump(mode="python") == original @@ -278,7 +282,8 @@ def test_mixed_history_uses_each_generations_own_request( _outcome(history, tokenizer) rendered_messages = tokenizer.calls[0] assert "reasoning_content" not in rendered_messages[1] - assert rendered_messages[3]["reasoning_content"] == "" + assert "reasoning_content" not in rendered_messages[3] + assert _LITERAL in tokenizer.rendered[0] assert [message["content"] for message in rendered_messages] == [ message["content"] for message in history.messages ] @@ -402,6 +407,9 @@ def observe(*args: Any, **kwargs: Any) -> tr.TokenizedHistory | None: monkeypatch.setattr(_tokenize, "_tokenize_exact_projected_chat_history", observe) with monkeypatch.context() as patch: + patch.setattr( + _tokenize, "chat_template_with_preserved_thinking", lambda value: value + ) patch.setattr( _tokenize, "_preserve_literal_thinking_off_content", lambda *args: None )