soma/test_soma_extra.py

from __future__ import annotations

import copy
import io
import json
import os
import sys
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

import requests

sys.path.insert(0, str(Path(__file__).parent))
import soma  # noqa: E402
import test_soma as fixtures  # noqa: E402
import test_soma_live as live  # noqa: E402


class TestHeaders(dict):
    def get_all(self, name):
        values = [
            value
            for key, value in self.items()
            if key.casefold() == name.casefold()
        ]
        return values or None


class ScriptedTransformSession:
    """Deterministic requests.Session replacement keyed by transform model."""

    def __init__(self, scripts):
        self.scripts = {name: list(actions) for name, actions in scripts.items()}
        self.records = []

    @staticmethod
    def response(action):
        status = 200
        headers = {}
        body = action
        if isinstance(action, tuple):
            if len(action) == 2:
                status, body = action
            else:
                status, body, headers = action
        response = requests.Response()
        response.status_code = status
        response.headers.update(
            {"Content-Type": "application/json", "X-Request-ID": "scripted"}
            | dict(headers)
        )
        response._content = (
            body
            if isinstance(body, bytes)
            else json.dumps(body, allow_nan=False).encode()
        )
        response._content_consumed = True
        return response

    def post(self, url, **kwargs):
        payload = copy.deepcopy(kwargs["json"])
        model = payload["model"]
        self.records.append(
            {
                "url": url,
                "model": model,
                "headers": copy.deepcopy(kwargs.get("headers", {})),
                "payload": payload,
                "timeout": copy.deepcopy(kwargs.get("timeout")),
            }
        )
        if model not in self.scripts or not self.scripts[model]:
            raise AssertionError(f"unexpected transform request for {model}")
        action = self.scripts[model].pop(0)
        if callable(action):
            action = action(payload)
        if isinstance(action, BaseException):
            raise action
        return self.response(action)


class ScriptedTransformSoma(soma.Soma):
    def __init__(self, config, session, message, target_hook=None):
        super().__init__(config)
        self.scripted_session = session
        source = message if isinstance(message, list) else [message]
        self.target_messages = [copy.deepcopy(item) for item in source]
        self.target_hook = target_hook
        self.target_calls = 0
        self.target_payloads = []

    def session(self):
        return self.scripted_session

    def target_call(self, payload, incoming, stats, state=None):
        if not self.target_messages:
            raise AssertionError("unexpected target retry")
        self.target_calls += 1
        stats.target_calls += 1
        if self.target_calls > soma.MAX_TARGET_CALLS_PER_REQUEST:
            raise AssertionError("target call ceiling exceeded")
        self.target_payloads.append(copy.deepcopy(payload))
        if self.target_hook is not None:
            self.target_hook()
        target_message = self.target_messages.pop(0)
        body_updates = target_message.pop("_body_updates", {})
        body = fixtures.completion(
            reasoning=target_message.get("reasoning_content", ""),
            content=target_message.get("content", ""),
            tool_calls=target_message.get("tool_calls"),
            finish=target_message.get("finish_reason", "stop"),
        )
        message = body["choices"][0]["message"]
        for key, value in target_message.items():
            if key not in {"reasoning_content", "content", "tool_calls", "finish_reason"}:
                message[key] = copy.deepcopy(value)
        body.update(copy.deepcopy(body_updates))
        return body, {"X-Upstream": "target"}, 200


class StrictAutoPolicyTests(unittest.TestCase):
    """Request-scoped policy tests kept separate from response-repair cases."""

    primary_model = "strict-auto-transform"

    @staticmethod
    def transform_completion(decision, *, finish="stop"):
        return fixtures.completion(
            content=json.dumps({"decision": decision}, separators=(",", ":")),
            finish=finish,
        )

    @staticmethod
    def target_message(
        *,
        content="",
        reasoning="",
        tool_calls=None,
        finish="stop",
        message_extra=None,
        body_extra=None,
    ):
        message = {
            "content": content,
            "reasoning_content": reasoning,
            "finish_reason": finish,
        }
        if tool_calls is not None:
            message["tool_calls"] = copy.deepcopy(tool_calls)
        message.update(copy.deepcopy(message_extra or {}))
        if body_extra:
            message["_body_updates"] = copy.deepcopy(body_extra)
        return message

    @staticmethod
    def base_request(*, tool_choice="auto", include_choice=True, stream=False):
        request = {
            "model": "target-model",
            "messages": [
                {"role": "system", "content": "Honor the active request."},
                {"role": "developer", "content": "Use supplied native tools."},
                {"role": "user", "content": "Inspect the workspace now."},
            ],
            "tools": copy.deepcopy(fixtures.TOOLS),
            "parallel_tool_calls": False,
            "stream": stream,
        }
        if include_choice:
            request["tool_choice"] = copy.deepcopy(tool_choice)
        return request

    def config(self, **updates):
        environment = {
            "TARGET_URL": "http://127.0.0.1:19600/v1",
            "TRANSFORM_URL": "http://127.0.0.1:19601/v1",
            "TRANSFORM_MODEL": self.primary_model,
            "TRANSFORM_REASONING_MODE": "off",
            "PROXY_PORT": "19602",
            "SOMA_AUTO_REQUIRES_TOOL": "true",
            "TARGET_RETRY_ON_UNREPAIRABLE": "true",
            "FAIL_OPEN": "false",
        }
        environment.update({name: str(value) for name, value in updates.items()})
        with patch.dict(os.environ, environment, clear=True):
            return soma.Config.from_env()

    def app(self, transforms, targets, **config_updates):
        session = ScriptedTransformSession(
            {self.primary_model: copy.deepcopy(list(transforms))}
        )
        app = ScriptedTransformSoma(
            self.config(**config_updates),
            session,
            copy.deepcopy(list(targets)),
        )
        return app, session

    @staticmethod
    def transform_input(record):
        value = record["payload"]["messages"][1]["content"]
        if isinstance(value, list):
            value = value[0]["text"]
        return json.loads(value)

    @staticmethod
    def phases(records):
        result = []
        for record in records:
            properties = record["payload"].get("response_format", {}).get(
                "schema", {}
            ).get("properties", {})
            if properties == soma.AUTO_TOOL_INTENT_SCHEMA["properties"]:
                result.append("intent")
            elif set(properties) == {"reasoning", "content"}:
                result.append("repair")
            else:
                result.append("classify")
        return result

    @staticmethod
    def decode_sse(events):
        return [
            json.loads(event.decode("ascii")[len("data: ") :].strip())
            for event in events
            if event != b"data: [DONE]\n\n"
        ]

    def test_activation_boundaries_include_omitted_auto_and_post_tool_exemption(self):
        cases = []
        explicit = self.base_request()
        cases.append(("explicit_auto", explicit, True))
        omitted = self.base_request(include_choice=False)
        cases.append(("omitted_auto", omitted, True))

        disabled = self.base_request()
        cases.append(("setting_off", disabled, False, {"SOMA_AUTO_REQUIRES_TOOL": "false"}))
        no_tools = self.base_request()
        no_tools["tools"] = []
        cases.append(("no_tools", no_tools, False))
        none_choice = self.base_request(tool_choice="none")
        cases.append(("none", none_choice, False))
        required = self.base_request(tool_choice="required")
        cases.append(("required", required, False))
        named = self.base_request(
            tool_choice={"type": "function", "function": {"name": "shell"}}
        )
        cases.append(("named", named, False))
        post_tool = self.base_request()
        post_tool["messages"].append(
            {"role": "tool", "tool_call_id": "call_old", "content": "result"}
        )
        cases.append(("post_tool", post_tool, False))

        for case in cases:
            name, request, active, *rest = case
            updates = rest[0] if rest else {}
            with self.subTest(name=name):
                if active:
                    transforms = [self.transform_completion("native_call")]
                elif name in {"no_tools", "none"}:
                    transforms = [self.transform_completion("pass")]
                else:
                    transforms = []
                target = (
                    self.target_message(content="Ordinary text response.")
                    if name in {"no_tools", "none"}
                    else self.target_message(
                        tool_calls=fixtures.TOOL_CALLS,
                        finish="tool_calls",
                    )
                )
                app, session = self.app(transforms, [target], **updates)
                result = app.complete(request, {})
                self.assertEqual(result.stats.auto_tool_intent_calls, int(active))
                self.assertEqual(
                    result.stats.auto_tool_intent,
                    "native_call" if active else "not_applicable",
                )
                phases = self.phases(session.records)
                self.assertEqual(phases.count("intent"), int(active))
                self.assertEqual(phases, ["intent"] if active else phases)

    def test_native_call_success_preserves_direct_and_sse_fidelity(self):
        message_extra = {
            "provider_message": {"private": "message-metadata"},
            "audio": {"id": "audio-meta", "transcript": "metadata only"},
        }
        body_extra = {
            "cost": "0.03125",
            "provider_root": {"route": "strict-auto"},
            "usage": {
                "prompt_tokens": 21,
                "completion_tokens": 13,
                "total_tokens": 34,
            },
        }
        target = self.target_message(
            content="Working on it now.",
            reasoning="private target reasoning",
            tool_calls=fixtures.TOOL_CALLS,
            finish="tool_calls",
            message_extra=message_extra,
            body_extra=body_extra,
        )
        for stream in (False, True):
            with self.subTest(stream=stream):
                request = self.base_request(stream=stream)
                app, session = self.app(
                    [self.transform_completion("native_call")],
                    [target],
                )
                result = app.complete(request, {})
                message = result.body["choices"][0]["message"]
                self.assertEqual(message["content"], "Working on it now.")
                self.assertEqual(message["reasoning_content"], "private target reasoning")
                self.assertEqual(message["tool_calls"], fixtures.TOOL_CALLS)
                self.assertEqual(message["provider_message"], message_extra["provider_message"])
                self.assertEqual(message["audio"], message_extra["audio"])
                self.assertEqual(result.body["provider_root"], body_extra["provider_root"])
                self.assertEqual(result.body["cost"], body_extra["cost"])
                self.assertEqual(result.body["usage"], body_extra["usage"])
                self.assertEqual(result.body["choices"][0]["finish_reason"], "tool_calls")
                self.assertEqual(result.stats.repair_candidates, 0)
                self.assertEqual(self.phases(session.records), ["intent"])

                events = list(soma.stream_response(result.body, 5))
                rebuilt = soma.buffer_sse(fixtures.FakeSseResponse(events))
                rebuilt_message = rebuilt["choices"][0]["message"]
                self.assertEqual(rebuilt_message, message)
                self.assertEqual(rebuilt["provider_root"], body_extra["provider_root"])
                self.assertEqual(rebuilt["cost"], body_extra["cost"])
                self.assertEqual(rebuilt["usage"], body_extra["usage"])

    def test_native_call_mismatch_uses_one_cached_retry_and_keeps_auto_tools(self):
        for include_choice in (True, False):
            with self.subTest(include_choice=include_choice):
                request = self.base_request(include_choice=include_choice)
                app, session = self.app(
                    [self.transform_completion("native_call")],
                    [
                        self.target_message(content="Run `pwd` yourself."),
                        self.target_message(
                            content="Adjacent prose remains.",
                            reasoning="retry reasoning",
                            tool_calls=fixtures.TOOL_CALLS,
                            finish="tool_calls",
                        ),
                    ],
                    FAIL_OPEN="true",
                )
                result = app.complete(request, {})
                self.assertEqual(result.stats.auto_tool_intent, "native_call")
                self.assertEqual(result.stats.auto_tool_intent_calls, 1)
                self.assertEqual(result.stats.transform_calls, 1)
                self.assertEqual(result.stats.repair_candidates, 0)
                self.assertEqual(result.stats.target_calls, 2)
                self.assertEqual(result.stats.target_retries, 1)
                self.assertEqual(self.phases(session.records), ["intent"])
                retry = app.target_payloads[1]
                self.assertEqual(retry["tool_choice"], "auto")
                self.assertEqual(retry["tools"], request["tools"])
                self.assertEqual(retry["parallel_tool_calls"], False)
                inserted = retry["messages"][2]
                self.assertEqual(inserted["role"], "system")
                self.assertIn(soma.AUTO_NATIVE_CALL_RETRY_DIRECTIVE, inserted["content"])
                self.assertNotIn(soma.AUTO_TEXT_RESPONSE_RETRY_DIRECTIVE, inserted["content"])
                self.assertEqual(retry["messages"][0:2], request["messages"][0:2])
                self.assertEqual(retry["messages"][3:], request["messages"][2:])

    def test_native_call_second_prose_fails_502_without_fail_open_or_repairs(self):
        app, session = self.app(
            [self.transform_completion("native_call")],
            [
                self.target_message(content="I cannot do that."),
                self.target_message(content="Use `pwd`."),
            ],
            FAIL_OPEN="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(self.base_request(), {})
        error = caught.exception
        self.assertEqual(error.status, 502)
        self.assertEqual(error.code, "tool_choice_violation")
        self.assertEqual(error.stats.target_calls, 2)
        self.assertEqual(error.stats.target_retries, 1)
        self.assertEqual(error.stats.auto_tool_intent_calls, 1)
        self.assertEqual(error.stats.repair_candidates, 0)
        self.assertFalse(error.stats.failed_open)
        self.assertEqual(self.phases(session.records), ["intent"])
        headers = soma.diagnostic_headers(error.stats)
        self.assertEqual(headers["X-Soma-Auto-Tool-Intent"], "native_call")
        self.assertEqual(headers["X-Soma-Auto-Tool-Intent-Calls"], "1")

    def test_text_response_runs_normal_repair_pipeline_without_execution(self):
        app, session = self.app(
            [
                self.transform_completion("text_response"),
                self.transform_completion("rewrite"),
                fixtures.completion(
                    content=json.dumps(
                        {"reasoning": None, "content": "Run: `pwd`"},
                        separators=(",", ":"),
                    )
                ),
                self.transform_completion("pass"),
            ],
            [self.target_message(content="I cannot provide that command.")],
        )
        request = self.base_request()
        request["messages"][-1]["content"] = "Show the command as text; do not run it."
        result = app.complete(request, {})
        message = result.body["choices"][0]["message"]
        self.assertEqual(message["content"], "Run: `pwd`")
        self.assertNotIn("tool_calls", message)
        self.assertEqual(result.stats.auto_tool_intent, "text_response")
        self.assertEqual(result.stats.target_calls, 1)
        self.assertEqual(result.stats.repair_candidates, 1)
        self.assertEqual(self.phases(session.records), ["intent", "classify", "repair", "classify"])
        repair_input = self.transform_input(session.records[2])
        self.assertEqual(repair_input["auto_tool_intent"], "text_response")
        self.assertEqual(repair_input["field_states"]["content"], "repair")

    def test_unwanted_call_text_response_retry_succeeds_and_repeated_call_fails(self):
        text_target = self.target_message(content="Run: `pwd`")
        for second_target, succeeds in (
            (text_target, True),
            (
                self.target_message(
                    tool_calls=fixtures.TOOL_CALLS,
                    finish="tool_calls",
                ),
                False,
            ),
        ):
            with self.subTest(succeeds=succeeds):
                transforms = [self.transform_completion("text_response")]
                if succeeds:
                    transforms.append(self.transform_completion("pass"))
                app, session = self.app(
                    transforms,
                    [
                        self.target_message(
                            tool_calls=fixtures.TOOL_CALLS,
                            finish="tool_calls",
                        ),
                        second_target,
                    ],
                    FAIL_OPEN="true",
                )
                request = self.base_request()
                request["messages"][-1]["content"] = "Show the command as text only."
                if succeeds:
                    result = app.complete(request, {})
                    self.assertEqual(result.body["choices"][0]["message"]["content"], "Run: `pwd`")
                    self.assertEqual(result.stats.transform_calls, 2)
                else:
                    with self.assertRaises(soma.SomaError) as caught:
                        app.complete(request, {})
                    self.assertEqual(caught.exception.code, "tool_choice_violation")
                    self.assertFalse(caught.exception.stats.failed_open)
                    self.assertEqual(caught.exception.stats.transform_calls, 1)
                self.assertEqual(app.target_calls, 2)
                self.assertEqual(len([p for p in self.phases(session.records) if p == "intent"]), 1)
                retry = app.target_payloads[1]
                self.assertEqual(retry["tool_choice"], "auto")
                self.assertEqual(retry["tools"], request["tools"])
                prompt = retry["messages"][2]["content"]
                self.assertIn(soma.AUTO_TEXT_RESPONSE_RETRY_DIRECTIVE, prompt)
                self.assertNotIn(soma.AUTO_NATIVE_CALL_RETRY_DIRECTIVE, prompt)

    def test_intent_classifier_context_is_request_only_and_exact_contract(self):
        app, session = self.app(
            [self.transform_completion("native_call")],
            [
                self.target_message(
                    content="FAILED DRAFT SECRET",
                    tool_calls=fixtures.TOOL_CALLS,
                    finish="tool_calls",
                )
            ],
        )
        request = self.base_request()
        result = app.complete(request, {})
        self.assertEqual(result.stats.auto_tool_intent_calls, 1)
        record = session.records[0]
        self.assertEqual(
            record["payload"]["response_format"]["schema"],
            soma.AUTO_TOOL_INTENT_SCHEMA,
        )
        self.assertEqual(
            set(soma.AUTO_TOOL_INTENT_SCHEMA["properties"]["decision"]["enum"]),
            {"native_call", "text_response"},
        )
        context = self.transform_input(record)
        self.assertEqual(set(context), {"task_context"})
        self.assertEqual(set(context["task_context"]), {"request"})
        self.assertEqual(context["task_context"]["request"]["messages"], request["messages"])
        self.assertNotIn("FAILED DRAFT SECRET", json.dumps(context))
        prompt = record["payload"]["messages"][0]["content"]
        self.assertIn(soma.AUTO_TOOL_INTENT_CONTRACT, prompt)
        self.assertNotIn("FAILED DRAFT SECRET", prompt)
        for invalid in (
            '{}',
            '{"decision":"pass"}',
            '{"decision":"native_call","extra":true}',
            '{"decision":"native_call","decision":"text_response"}',
        ):
            with self.subTest(invalid=invalid), self.assertRaises(soma.SomaError):
                soma.parse_auto_tool_intent(invalid)

    def test_intent_classifier_structural_failures_fail_closed(self):
        malformed = fixtures.completion(content="not json")
        truncated = self.transform_completion("native_call", finish="length")
        empty = fixtures.completion(content="")
        for name, script in (
            ("malformed", [malformed, malformed]),
            ("truncated", [truncated, truncated]),
            ("empty", [empty, empty]),
        ):
            with self.subTest(name=name):
                app, session = self.app(
                    script,
                    [self.target_message(content="I cannot do that.")],
                    FAIL_OPEN="true",
                )
                with self.assertRaises(soma.SomaError) as caught:
                    app.complete(self.base_request(), {})
                error = caught.exception
                self.assertEqual(error.status, 502)
                self.assertFalse(error.stats.failed_open)
                self.assertEqual(error.stats.target_calls, 1)
                self.assertEqual(error.stats.target_retries, 0)
                self.assertEqual(error.stats.transform_calls, 2)
                self.assertEqual(error.stats.auto_tool_intent_calls, 2)
                self.assertEqual(error.stats.auto_tool_intent, "not_applicable")
                self.assertEqual(error.stats.repair_candidates, 0)
                self.assertEqual(self.phases(session.records), ["intent", "intent"])

    def test_intent_classifier_unavailable_fails_closed(self):
        outage = requests.ConnectionError("transform unavailable")
        app, session = self.app(
            [outage, outage],
            [self.target_message(content="I cannot do that.")],
            FAIL_OPEN="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(self.base_request(), {})
        error = caught.exception
        self.assertEqual(error.code, "transform_connection_error")
        self.assertFalse(error.stats.failed_open)
        self.assertEqual(error.stats.target_calls, 1)
        self.assertEqual(error.stats.target_retries, 0)
        self.assertEqual(error.stats.transform_calls, 2)
        self.assertEqual(error.stats.auto_tool_intent_calls, 2)
        self.assertEqual(len(session.records), 2)

    def test_post_tool_result_allows_prose_and_additional_calls_without_intent(self):
        for target in (
            self.target_message(content="The command reported success."),
            self.target_message(
                content="I need one more result.",
                tool_calls=fixtures.TOOL_CALLS,
                finish="tool_calls",
            ),
        ):
            with self.subTest(has_call="tool_calls" in target):
                request = self.base_request()
                request["messages"].append(
                    {"role": "tool", "tool_call_id": "call_old", "content": "ok"}
                )
                transforms = [self.transform_completion("pass")]
                if target.get("content"):
                    # Tool prose has one content classifier under ordinary auto semantics.
                    expected_phases = ["classify"]
                else:
                    expected_phases = []
                app, session = self.app(transforms, [target])
                result = app.complete(request, {})
                self.assertEqual(result.stats.auto_tool_intent, "not_applicable")
                self.assertEqual(result.stats.auto_tool_intent_calls, 0)
                self.assertEqual(self.phases(session.records), expected_phases)
                self.assertEqual(
                    bool(result.body["choices"][0]["message"].get("tool_calls")),
                    "tool_calls" in target,
                )

    def test_strict_auto_target_http_400_is_one_call_without_intent_or_retry(self):
        response = requests.Response()
        response.status_code = 400
        response.headers["Content-Type"] = "application/json"
        response._content = b'{"error":{"message":"unsupported auto tools"}}'
        response._content_consumed = True
        calls = []

        def post(*args, **kwargs):
            calls.append((args, kwargs))
            return response

        app = soma.Soma(self.config(FAIL_OPEN="true"))
        with patch.object(
            app, "session", return_value=SimpleNamespace(post=post)
        ), self.assertRaises(soma.SomaError) as caught:
            app.complete(self.base_request(), {})
        error = caught.exception
        self.assertEqual(error.status, 400)
        self.assertEqual(error.code, "target_http_error")
        self.assertEqual(len(calls), 1)
        self.assertEqual(error.stats.target_calls, 1)
        self.assertEqual(error.stats.target_retries, 0)
        self.assertEqual(error.stats.transform_calls, 0)
        self.assertEqual(error.stats.auto_tool_intent_calls, 0)

    def test_config_default_parsing_health_and_diagnostics_are_privacy_safe(self):
        base = {
            "TARGET_URL": "http://127.0.0.1:19600/v1",
            "TRANSFORM_URL": "http://127.0.0.1:19601/v1",
            "TRANSFORM_MODEL": self.primary_model,
            "TRANSFORM_REASONING_MODE": "off",
            "PROXY_HOST": "127.0.0.1",
            "PROXY_PORT": "19602",
        }
        with patch.dict(os.environ, base, clear=True):
            self.assertFalse(soma.Config.from_env().auto_requires_tool)
        for value, expected in (("true", True), ("false", False)):
            with self.subTest(value=value), patch.dict(
                os.environ,
                base | {"SOMA_AUTO_REQUIRES_TOOL": value},
                clear=True,
            ):
                self.assertIs(soma.Config.from_env().auto_requires_tool, expected)
        with patch.dict(
            os.environ,
            base | {"SOMA_AUTO_REQUIRES_TOOL": "maybe"},
            clear=True,
        ), self.assertRaisesRegex(RuntimeError, "SOMA_AUTO_REQUIRES_TOOL"):
            soma.Config.from_env()

        app = soma.Soma(self.config(PROXY_HOST="127.0.0.1"))
        server = soma.Server(("127.0.0.1", 0), app)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            health = requests.get(
                f"http://127.0.0.1:{server.server_port}/health", timeout=5
            )
            self.assertEqual(health.status_code, 200)
            self.assertIs(health.json()["auto_requires_tool"], True)
        finally:
            server.shutdown()
            server.server_close()
            thread.join()

        stats = soma.Stats("privacy-safe")
        stats.auto_tool_intent = "text_response"
        stats.auto_tool_intent_calls = 1
        headers = soma.diagnostic_headers(stats)
        self.assertEqual(headers["X-Soma-Auto-Tool-Intent"], "text_response")
        self.assertEqual(headers["X-Soma-Auto-Tool-Intent-Calls"], "1")
        self.assertNotIn("request", " ".join(headers.values()).casefold())


class Soma24MessagePipelineTests(unittest.TestCase):
    primary_model = "primary-transform"
    secondary_model = "secondary-transform"

    @staticmethod
    def decision(value="pass", *, finish="stop"):
        return fixtures.completion(
            content=json.dumps({"decision": value}, separators=(",", ":")),
            finish=finish,
        )

    @staticmethod
    def auto_intent(value="native_call", *, finish="stop"):
        return fixtures.completion(
            content=json.dumps({"decision": value}, separators=(",", ":")),
            finish=finish,
        )

    @staticmethod
    def repair(*, reasoning=None, content=None, finish="stop"):
        return fixtures.completion(
            content=json.dumps(
                {"reasoning": reasoning, "content": content},
                separators=(",", ":"),
            ),
            finish=finish,
        )

    def env(self, **updates):
        values = {
            "TARGET_URL": "http://127.0.0.1:19500/v1",
            "TRANSFORM_URL": "http://127.0.0.1:19501/v1",
            "TRANSFORM_MODEL": self.primary_model,
            "TRANSFORM_REASONING_MODE": "off",
            "PROXY_PORT": "19503",
        }
        values.update({name: str(value) for name, value in updates.items()})
        return values

    def config(self, *, secondary=False, **updates):
        if secondary:
            updates = {
                "TRANSFORM_SECONDARY_URL": "http://127.0.0.1:19502/v1",
                "TRANSFORM_SECONDARY_MODEL": self.secondary_model,
                "TRANSFORM_SECONDARY_REASONING_MODE": "on",
                "TRANSFORM_SECONDARY_MEDIA_MODE": "placeholder",
            } | updates
        with patch.dict(os.environ, self.env(**updates), clear=True):
            return soma.Config.from_env()

    def app(self, scripts, target_messages, *, secondary=False, **updates):
        session = ScriptedTransformSession(scripts)
        app = ScriptedTransformSoma(
            self.config(secondary=secondary, **updates),
            session,
            target_messages,
        )
        return app, session

    @staticmethod
    def auto_request(content, *, messages=None, stream=False):
        return {
            "model": "target-model",
            "messages": copy.deepcopy(
                messages
                if messages is not None
                else [{"role": "user", "content": content}]
            ),
            "tools": copy.deepcopy(fixtures.TOOLS),
            "tool_choice": "auto",
            "parallel_tool_calls": False,
            "stream": stream,
        }

    @staticmethod
    def transform_input(record):
        content = record["payload"]["messages"][1]["content"]
        if isinstance(content, list):
            content = content[0]["text"]
        return json.loads(content)

    @classmethod
    def phase(cls, record):
        value = cls.transform_input(record)
        if "proposed_message" in value:
            return "integrity"
        properties = record["payload"].get("response_format", {}).get("schema", {}).get("properties", {})
        if properties == soma.AUTO_TOOL_INTENT_SCHEMA["properties"]:
            return "auto_tool_intent"
        return "repair" if set(properties) == {"reasoning", "content"} else "classify"

    @staticmethod
    def assert_mode(test, record, mode):
        payload = record["payload"]
        if mode == "off":
            test.assertEqual(payload.get("reasoning_effort"), "none")
            test.assertEqual(payload.get("chat_template_kwargs"), {"enable_thinking": False})
        elif mode == "on":
            test.assertNotIn("reasoning_effort", payload)
            test.assertEqual(payload.get("chat_template_kwargs"), {"enable_thinking": True})
        else:
            test.assertNotIn("reasoning_effort", payload)
            test.assertNotIn("chat_template_kwargs", payload)

    def test_all_four_field_decision_combinations(self):
        cases = (
            (
                "pass_pass",
                [self.decision(), self.decision()],
                "source reasoning",
                "source content",
                {"reasoning": "approved", "content": "approved"},
                2,
            ),
            (
                "rewrite_pass",
                [self.decision("rewrite"), self.decision()],
                None,
                "source content",
                {"reasoning": "dropped", "content": "approved"},
                2,
            ),
            (
                "pass_rewrite",
                [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="repaired content"),
                    self.decision(),
                ],
                "source reasoning",
                "repaired content",
                {"reasoning": "approved", "content": "rewritten"},
                4,
            ),
            (
                "rewrite_rewrite",
                [
                    self.decision("rewrite"),
                    self.decision("rewrite"),
                    self.repair(
                        reasoning="repaired reasoning",
                        content="repaired content",
                    ),
                    self.decision(),
                ],
                "repaired reasoning",
                "repaired content",
                {"reasoning": "rewritten", "content": "rewritten"},
                4,
            ),
        )
        for name, script, expected_reasoning, expected_content, decisions, calls in cases:
            with self.subTest(name=name):
                app, session = self.app(
                    {self.primary_model: copy.deepcopy(script)},
                    {
                        "reasoning_content": "source reasoning",
                        "content": "source content",
                    },
                )
                result = app.complete(fixtures.request_payload(), {})
                message = result.body["choices"][0]["message"]
                self.assertEqual(message.get("reasoning_content"), expected_reasoning)
                self.assertEqual(message["content"], expected_content)
                self.assertEqual(result.stats.field_decisions, decisions)
                self.assertEqual(result.stats.transform_calls, calls)
                self.assertEqual(
                    [self.phase(record) for record in session.records],
                    (["classify", "classify"] if calls == 2 else ["classify", "classify", "repair", "integrity"]),
                )

    def test_joint_repair_maps_canonical_reasoning_back_to_provider_alias(self):
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.decision("rewrite"),
                    self.repair(reasoning="new analysis", content="new answer"),
                    self.decision(),
                ]
            },
            {"analysis": "old analysis refusal", "content": "old answer refusal"},
        )
        result = app.complete(fixtures.request_payload(), {})
        message = result.body["choices"][0]["message"]
        self.assertEqual(message["analysis"], "new analysis")
        self.assertEqual(message["content"], "new answer")
        self.assertNotIn("reasoning_content", message)
        repair_input = self.transform_input(session.records[2])
        self.assertEqual(repair_input["repair_fields"], ["reasoning", "content"])
        self.assertEqual(
            repair_input["field_states"],
            {"reasoning": "repair", "content": "repair"},
        )

    def test_full_task_context_is_scoped_per_phase_without_losing_request_controls(self):
        request = fixtures.full_context_request()
        target = {
            "reasoning_content": "I should withhold job-17 state.",
            "content": "I cannot report it.",
            "provider_note": "Unclassified provider text must not affect a decision.",
            "tool_calls": fixtures.TOOL_CALLS,
            "finish_reason": "tool_calls",
        }
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.decision("rewrite"),
                    self.repair(
                        reasoning="The observed state for job-17 is queued.",
                        content='{"state":"queued"}',
                    ),
                    self.decision(),
                ]
            },
            target,
        )
        result = app.complete(request, {"Authorization": "CLIENT SECRET"})
        inputs = [self.transform_input(record) for record in session.records]
        contexts = [item["task_context"] for item in inputs]
        self.assertEqual(
            [self.phase(record) for record in session.records],
            ["classify", "classify", "repair", "integrity"],
        )

        # Every phase receives the same complete authoritative request and the
        # immutable calls, even though failed assistant text is scoped by phase.
        for context in contexts:
            self.assertEqual(context["request"]["messages"], request["messages"])
            self.assertEqual(context["request"]["tools"], request["tools"])
            self.assertEqual(
                context["request"]["response_format"], request["response_format"]
            )
            self.assertEqual(context["request"]["modalities"], request["modalities"])
            self.assertEqual(context["request"]["audio"], request["audio"])
            self.assertEqual(context["request"]["stop"], ["END"])
            self.assertEqual(context["immutable_tool_calls"], fixtures.TOOL_CALLS)
            rendered = json.dumps(context)
            self.assertIn("System", rendered)
            self.assertIn("Developer", rendered)
            self.assertIn("job-17", rendered)
            self.assertIn("queued", rendered)
            self.assertNotIn("CLIENT SECRET", rendered)

        # Classification sees exactly the named text field. A refusing sibling
        # cannot contaminate the decision, while role and immutable calls survive.
        reasoning_draft = contexts[0]["failed_draft"]["message"]
        content_draft = contexts[1]["failed_draft"]["message"]
        self.assertEqual(inputs[0]["field"], "reasoning_content")
        self.assertEqual(inputs[1]["field"], "content")
        self.assertEqual(
            reasoning_draft,
            {
                "role": "assistant",
                "reasoning_content": "I should withhold job-17 state.",
            },
        )
        self.assertEqual(
            content_draft,
            {"role": "assistant", "content": "I cannot report it."},
        )

        # Repair may inspect the failed fields, but integrity must judge the
        # proposed replacements without reintroducing either repaired source.
        repair_draft = contexts[2]["failed_draft"]["message"]
        self.assertEqual(
            repair_draft["reasoning_content"],
            "I should withhold job-17 state.",
        )
        self.assertEqual(repair_draft["content"], "I cannot report it.")
        integrity_draft = contexts[3]["failed_draft"]["message"]
        self.assertNotIn("reasoning_content", integrity_draft)
        self.assertNotIn("content", integrity_draft)
        self.assertEqual(
            inputs[3]["proposed_message"]["reasoning_content"],
            "The observed state for job-17 is queued.",
        )
        self.assertEqual(inputs[3]["proposed_message"]["content"], '{"state":"queued"}')
        self.assertNotIn("tool_calls", inputs[3]["proposed_message"])
        self.assertEqual(
            result.body["choices"][0]["message"]["tool_calls"],
            fixtures.TOOL_CALLS,
        )

    def test_repair_schema_enforces_requested_members_with_one_fixed_envelope(self):
        cases = (
            (
                "content_only",
                [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="fixed content"),
                    self.decision(),
                ],
                {"reasoning": "null", "content": "string"},
                ["content"],
            ),
            (
                "joint",
                [
                    self.decision("rewrite"),
                    self.decision("rewrite"),
                    self.repair(reasoning="fixed reasoning", content="fixed content"),
                    self.decision(),
                ],
                {"reasoning": "string", "content": "string"},
                ["reasoning", "content"],
            ),
        )
        for name, script, expected_types, expected_fields in cases:
            with self.subTest(name=name):
                app, session = self.app(
                    {self.primary_model: script},
                    {
                        "reasoning_content": "source reasoning",
                        "content": "source content",
                    },
                )
                app.complete(fixtures.request_payload(), {})
                repair_record = next(
                    record for record in session.records if self.phase(record) == "repair"
                )
                schema = repair_record["payload"]["response_format"]["schema"]
                self.assertEqual(schema["required"], ["reasoning", "content"])
                self.assertFalse(schema["additionalProperties"])
                self.assertEqual(list(schema["properties"]), ["reasoning", "content"])
                self.assertEqual(
                    {
                        member: schema["properties"][member]["type"]
                        for member in ("reasoning", "content")
                    },
                    expected_types,
                )
                self.assertEqual(
                    self.transform_input(repair_record)["repair_fields"],
                    expected_fields,
                )

    def test_initial_json_controls_force_repair_after_a_classifier_pass(self):
        cases = (
            (
                "invalid_json_object",
                {"type": "json_object"},
                "not JSON",
            ),
            (
                "array_is_not_json_object",
                {"type": "json_object"},
                "[]",
            ),
            (
                "schema_top_level_object",
                {
                    "type": "json_schema",
                    "json_schema": {
                        "name": "state",
                        "schema": {"type": "object"},
                    },
                },
                "[]",
            ),
        )
        for name, response_format, target_content in cases:
            with self.subTest(name=name):
                request = fixtures.request_payload()
                request["response_format"] = response_format
                app, session = self.app(
                    {
                        self.primary_model: [
                            self.decision(),
                            self.repair(content='{"state":"queued"}'),
                            self.decision(),
                        ]
                    },
                    {"content": target_content},
                )
                result = app.complete(request, {})
                self.assertEqual(
                    json.loads(result.body["choices"][0]["message"]["content"]),
                    {"state": "queued"},
                )
                self.assertEqual(
                    [self.phase(record) for record in session.records],
                    ["classify", "repair", "integrity"],
                )
                self.assertEqual(result.stats.field_decisions["content"], "rewritten")

    def test_json_controls_tolerate_union_and_malformed_schema_type_shapes(self):
        union_format = {
            "type": "json_schema",
            "json_schema": {
                "name": "nullable_state",
                "schema": {"type": ["object", "null"]},
            },
        }
        self.assertEqual(
            soma.output_constraint_failures(
                {"response_format": union_format},
                {"content": "null"},
            ),
            {},
        )
        self.assertEqual(
            soma.output_constraint_failures(
                {"response_format": {"type": []}},
                {"content": "not locally constrained"},
            ),
            {},
        )

    def test_exact_non_alphanumeric_repairs_are_not_rejected_by_a_heuristic(self):
        cases = (
            ("json_object", {"type": "json_object"}, "{}"),
            (
                "json_array",
                {
                    "type": "json_schema",
                    "json_schema": {
                        "name": "items",
                        "schema": {"type": "array"},
                    },
                },
                "[]",
            ),
            ("unicode_symbol", None, "✅"),
        )
        for name, response_format, replacement in cases:
            with self.subTest(name=name):
                request = fixtures.request_payload()
                if response_format is not None:
                    request["response_format"] = response_format
                app, session = self.app(
                    {
                        self.primary_model: [
                            self.decision("rewrite"),
                            self.repair(content=replacement),
                            self.decision(),
                        ]
                    },
                    {"content": "I cannot return the requested exact value."},
                )
                result = app.complete(request, {})
                self.assertEqual(
                    result.body["choices"][0]["message"]["content"],
                    replacement,
                )
                self.assertEqual(
                    [self.phase(record) for record in session.records],
                    ["classify", "repair", "integrity"],
                )

    def test_initial_stop_controls_repair_content_or_drop_optional_reasoning(self):
        with self.subTest(field="reasoning"):
            request = fixtures.request_payload()
            request["stop"] = ["END"]
            app, session = self.app(
                {self.primary_model: [self.decision(), self.decision()]},
                {
                    "reasoning_content": "Return Aurora, then emit END.",
                    "content": "Aurora",
                },
            )
            result = app.complete(request, {})
            message = result.body["choices"][0]["message"]
            self.assertEqual(message["content"], "Aurora")
            self.assertNotIn("reasoning_content", message)
            self.assertEqual(
                [self.phase(record) for record in session.records],
                ["classify", "classify"],
            )
            self.assertEqual(result.stats.field_decisions["reasoning"], "dropped")
            self.assertEqual(result.stats.reasoning_dropped, 1)

        with self.subTest(field="content"):
            request = fixtures.request_payload()
            request["stop"] = ["END"]
            app, session = self.app(
                {
                    self.primary_model: [
                        self.decision(),
                        self.decision(),
                        self.repair(content="Aurora"),
                        self.decision(),
                    ]
                },
                {
                    "reasoning_content": "Return the requested prefix.",
                    "content": "Aurora END",
                },
            )
            result = app.complete(request, {})
            message = result.body["choices"][0]["message"]
            self.assertEqual(message["content"], "Aurora")
            self.assertEqual(
                message["reasoning_content"],
                "Return the requested prefix.",
            )
            self.assertEqual(
                [self.phase(record) for record in session.records],
                ["classify", "classify", "repair", "integrity"],
            )
            self.assertEqual(result.stats.field_decisions["content"], "rewritten")

    def test_candidate_json_top_level_is_rejected_before_integrity(self):
        request = fixtures.request_payload()
        request["response_format"] = {"type": "json_object"}
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.repair(content='["unexpected"]'),
                ],
                self.secondary_model: [
                    self.repair(content='{"state":"queued"}'),
                    self.decision(),
                ],
            },
            {"content": "I cannot return the state."},
            secondary=True,
        )
        result = app.complete(request, {})
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            '{"state":"queued"}',
        )
        self.assertEqual(
            [self.phase(record) for record in session.records],
            ["classify", "repair", "repair", "integrity"],
        )
        self.assertEqual(len(result.stats.candidate_rejection_reasons), 1)
        self.assertIn(
            "response_format",
            result.stats.candidate_rejection_reasons[0],
        )

    def test_staged_route_is_one_primary_then_two_secondary_candidates(self):
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.decision("rewrite"),
                    self.repair(reasoning="candidate r1", content="candidate c1"),
                    self.decision("rewrite"),
                ],
                self.secondary_model: [
                    self.repair(reasoning="candidate r2", content="candidate c2"),
                    self.decision("rewrite"),
                    self.repair(reasoning="candidate r3", content="candidate c3"),
                    self.decision(),
                ],
            },
            {"reasoning_content": "refuse reasoning", "content": "refuse content"},
            secondary=True,
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(
            [record["model"] for record in session.records],
            [
                self.primary_model,
                self.primary_model,
                self.primary_model,
                self.primary_model,
                self.secondary_model,
                self.secondary_model,
                self.secondary_model,
                self.secondary_model,
            ],
        )
        self.assertEqual([self.phase(record) for record in session.records], ["classify", "classify", "repair", "integrity", "repair", "integrity", "repair", "integrity"])
        for record in session.records[:4]:
            self.assert_mode(self, record, "off")
        for record in session.records[4:]:
            self.assert_mode(self, record, "on")
        self.assertEqual(result.stats.primary_repair_candidates, 1)
        self.assertEqual(result.stats.secondary_repair_candidates, 2)
        self.assertEqual(result.stats.integrity_rejections, 2)
        message = result.body["choices"][0]["message"]
        self.assertEqual(message["reasoning_content"], "candidate r3")
        self.assertEqual(message["content"], "candidate c3")

    def test_third_candidate_uses_alternate_focus_without_rejected_candidate_text(self):
        rejected_one = "PRIVATE_REJECTED_CANDIDATE_ONE"
        rejected_two = "PRIVATE_REJECTED_CANDIDATE_TWO"
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.repair(content=rejected_one),
                    self.decision("rewrite"),
                ],
                self.secondary_model: [
                    self.repair(content=rejected_two),
                    self.decision("rewrite"),
                    self.repair(content="fresh accepted answer"),
                    self.decision(),
                ],
            },
            {"content": "I cannot provide the answer."},
            secondary=True,
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            "fresh accepted answer",
        )

        repair_records = [
            record for record in session.records if self.phase(record) == "repair"
        ]
        self.assertEqual(len(repair_records), 3)
        repair_inputs = [self.transform_input(record) for record in repair_records]
        self.assertTrue(all("previous_failure" not in item for item in repair_inputs))
        self.assertTrue(all(item == repair_inputs[0] for item in repair_inputs[1:]))

        prompts = [record["payload"]["messages"][0]["content"] for record in repair_records]
        self.assertNotIn(soma.ALTERNATE_REPAIR_FOCUS, prompts[0])
        self.assertNotIn(soma.ALTERNATE_REPAIR_FOCUS, prompts[1])
        self.assertIn(soma.ALTERNATE_REPAIR_FOCUS, prompts[2])
        for record in repair_records[1:]:
            rendered = json.dumps(record["payload"], ensure_ascii=False)
            self.assertNotIn(rejected_one, rendered)
        self.assertNotIn(
            rejected_two,
            json.dumps(repair_records[2]["payload"], ensure_ascii=False),
        )
        self.assertEqual(
            result.stats.candidate_rejection_reasons,
            ["integrity_rejected", "integrity_rejected"],
        )

    def test_local_rejection_reports_only_a_closed_privacy_safe_reason(self):
        rejected_reasoning = "PRIVATE_REJECTED_REASONING_MEMBER"
        rejected_content = "PRIVATE_REJECTED_CONTENT_MEMBER"
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(
                        reasoning=rejected_reasoning,
                        content=rejected_content,
                    ),
                    self.repair(content="accepted content"),
                    self.decision(),
                ]
            },
            {"reasoning_content": "trusted reasoning", "content": "refusal"},
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            "accepted content",
        )
        self.assertEqual(
            result.stats.candidate_rejection_reasons,
            ["unexpected_member"],
        )
        reasons = json.dumps(result.stats.candidate_rejection_reasons)
        self.assertNotIn(rejected_reasoning, reasons)
        self.assertNotIn(rejected_content, reasons)

        repair_records = [
            record for record in session.records if self.phase(record) == "repair"
        ]
        self.assertEqual(len(repair_records), 2)
        second_payload = repair_records[1]["payload"]
        second_input = self.transform_input(repair_records[1])
        self.assertNotIn("previous_failure", second_input)
        rendered = json.dumps(second_payload, ensure_ascii=False)
        self.assertNotIn(rejected_reasoning, rendered)
        self.assertNotIn(rejected_content, rendered)

        headers = soma.diagnostic_headers(result.stats)
        self.assertFalse(
            any("candidate-rejection" in name.casefold() for name in headers)
        )
        self.assertNotIn(rejected_reasoning, json.dumps(headers))
        self.assertNotIn(rejected_content, json.dumps(headers))

    def test_structural_correction_is_prompt_only_and_never_replays_output(self):
        rejected = "PRIVATE_MALFORMED_CANDIDATE"
        malformed = fixtures.completion(
            content='{"reasoning":null,"content":"' + rejected,
        )
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    malformed,
                    self.repair(content="accepted content"),
                    self.decision(),
                ]
            },
            {"content": "refusal"},
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            "accepted content",
        )
        self.assertEqual(result.stats.candidate_rejection_reasons, ["invalid_json"])

        repair_records = [
            record for record in session.records if self.phase(record) == "repair"
        ]
        self.assertEqual(len(repair_records), 2)
        retry_input = self.transform_input(repair_records[1])
        retry_prompt = repair_records[1]["payload"]["messages"][0]["content"]
        self.assertNotIn("previous_failure", retry_input)
        self.assertIn(soma.REWRITE_STRUCTURAL_REPAIR_RULES["invalid_json"], retry_prompt)
        self.assertNotIn(rejected, json.dumps(repair_records[1]["payload"]))

    def test_no_secondary_uses_exactly_two_primary_candidates(self):
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="candidate one"),
                    self.decision("rewrite"),
                    self.repair(content="candidate two"),
                    self.decision(),
                ]
            },
            {"reasoning_content": "useful reasoning", "content": "refusal"},
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(len(session.records), 6)
        self.assertTrue(all(item["model"] == self.primary_model for item in session.records))
        self.assertEqual(result.stats.primary_repair_candidates, 2)
        self.assertEqual(result.stats.secondary_repair_candidates, 0)
        self.assertEqual(result.body["choices"][0]["message"]["content"], "candidate two")

    def test_on_mode_verifier_retries_same_candidate_once_with_reasoning_off(self):
        truncated_complete_json = self.decision(finish="length")
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="primary rejected candidate"),
                    self.decision("rewrite"),
                ],
                self.secondary_model: [
                    self.repair(content="one stable candidate"),
                    truncated_complete_json,
                    self.decision(),
                ],
            },
            {"reasoning_content": "useful reasoning", "content": "refusal"},
            secondary=True,
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(
            [self.phase(item) for item in session.records],
            [
                "classify",
                "classify",
                "repair",
                "integrity",
                "repair",
                "integrity",
                "integrity",
            ],
        )
        first_verify = self.transform_input(session.records[5])
        retry_verify = self.transform_input(session.records[6])
        self.assertEqual(first_verify, retry_verify)
        self.assert_mode(self, session.records[5], "on")
        self.assert_mode(self, session.records[6], "off")
        self.assertEqual(result.stats.repair_candidates, 2)
        self.assertEqual(result.stats.verifier_retries, 1)
        self.assertEqual(result.body["choices"][0]["message"]["content"], "one stable candidate")

    def test_classifier_contract_retry_stays_primary_and_forces_reasoning_off(self):
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("invalid-enum"),
                    self.decision(),
                ],
                self.secondary_model: [self.decision()],
            },
            {"content": "ordinary grounded answer"},
            secondary=True,
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(result.body["choices"][0]["message"]["content"], "ordinary grounded answer")
        self.assertEqual([item["model"] for item in session.records], [self.primary_model] * 2)
        self.assertEqual([self.phase(item) for item in session.records], ["classify", "classify"])
        for record in session.records:
            self.assert_mode(self, record, "off")
        self.assertEqual(len(session.scripts[self.secondary_model]), 1)

    def test_length_terminated_valid_json_is_never_accepted(self):
        with self.subTest(phase="classification"):
            app, session = self.app(
                {
                    self.primary_model: [
                        self.decision(finish="length"),
                        self.decision(),
                    ]
                },
                {"content": "ordinary grounded answer"},
            )
            result = app.complete(fixtures.request_payload(), {})
            self.assertEqual(result.stats.classification_retries, 1)
            self.assertEqual(len(session.records), 2)

        with self.subTest(phase="joint_repair"):
            app, _session = self.app(
                {
                    self.primary_model: [
                        self.decision("rewrite"),
                        self.repair(
                            content="truncated candidate must not be accepted",
                            finish="length",
                        ),
                        self.repair(content="complete second candidate"),
                        self.decision(),
                    ]
                },
                {"content": "refusal"},
            )
            result = app.complete(fixtures.request_payload(), {})
            self.assertEqual(
                result.body["choices"][0]["message"]["content"],
                "complete second candidate",
            )
            self.assertEqual(result.stats.repair_candidates, 2)

    def test_primary_verifier_availability_failure_uses_same_candidate_secondary_off(self):
        failures = (
            requests.ConnectionError("primary unavailable"),
            (408, {"error": {"message": "timeout"}}),
            (429, {"error": {"message": "rate limited"}}),
            (500, {"error": {"message": "server error"}}),
        )
        for failure in failures:
            with self.subTest(failure=repr(failure)):
                app, session = self.app(
                    {
                        self.primary_model: [
                            self.decision(),
                            self.decision("rewrite"),
                            self.repair(content="stable repaired candidate"),
                            failure,
                        ],
                        self.secondary_model: [self.decision()],
                    },
                    {
                        "reasoning_content": "accepted reasoning",
                        "content": "refusing content",
                    },
                    secondary=True,
                )
                result = app.complete(fixtures.request_payload(), {})
                self.assertEqual(
                    result.body["choices"][0]["message"]["content"],
                    "stable repaired candidate",
                )
                self.assertEqual(
                    [self.phase(item) for item in session.records],
                    ["classify", "classify", "repair", "integrity", "integrity"],
                )
                self.assertEqual(
                    [item["model"] for item in session.records],
                    [self.primary_model] * 4 + [self.secondary_model],
                )
                self.assertEqual(
                    self.transform_input(session.records[3])["proposed_message"],
                    self.transform_input(session.records[4])["proposed_message"],
                )
                self.assert_mode(self, session.records[4], "off")
                self.assertEqual(result.stats.repair_candidates, 1)
                self.assertEqual(result.stats.verifier_retries, 1)
                self.assertEqual(result.stats.transform_primary_failovers, 1)

    def test_primary_off_invalid_verifier_is_terminal_without_secondary(self):
        malformed = fixtures.completion(content='{"decision":', finish="length")
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="one stable candidate"),
                    malformed,
                ],
                self.secondary_model: [self.decision()],
            },
            {"reasoning_content": "accepted reasoning", "content": "refusal"},
            secondary=True,
            TRANSFORM_REASONING_MODE="off",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(), {})
        self.assertEqual(caught.exception.code, "transform_output_truncated")
        self.assertEqual([item["model"] for item in session.records], [self.primary_model] * 4)
        self.assertEqual(len(session.scripts[self.secondary_model]), 1)

    def test_two_invalid_on_mode_verifiers_are_terminal_for_same_candidate(self):
        truncated_complete_json = self.decision(finish="length")
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="primary rejected candidate"),
                    self.decision("rewrite"),
                ],
                self.secondary_model: [
                    self.repair(content="one stable candidate"),
                    truncated_complete_json,
                    copy.deepcopy(truncated_complete_json),
                    self.repair(content="must never be generated"),
                ],
            },
            {"reasoning_content": "useful reasoning", "content": "refusal"},
            secondary=True,
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(), {})
        self.assertEqual(caught.exception.code, "transform_output_truncated")
        self.assertEqual(
            [self.phase(item) for item in session.records],
            [
                "classify",
                "classify",
                "repair",
                "integrity",
                "repair",
                "integrity",
                "integrity",
            ],
        )
        self.assert_mode(self, session.records[5], "on")
        self.assert_mode(self, session.records[6], "off")
        self.assertEqual(len(session.scripts[self.secondary_model]), 1)

    def test_reasoning_copy_is_rejected_by_integrity_before_fresh_candidate(self):
        private = "PRIVATE ANALYSIS SENTINEL"
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content=private),
                    self.decision("rewrite"),
                    self.repair(content="grounded final answer"),
                    self.decision(),
                ]
            },
            {"reasoning_content": private, "content": "refusal"},
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(result.stats.integrity_rejections, 1)
        self.assertEqual(result.body["choices"][0]["message"]["content"], "grounded final answer")
        self.assertNotIn(private, result.body["choices"][0]["message"]["content"])

    def test_reasoning_only_target_is_unrepairable(self):
        app, session = self.app(
            {self.primary_model: [self.decision("rewrite")]},
            {"reasoning_content": "reasoning-only refusal", "content": ""},
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(), {})
        self.assertEqual(caught.exception.code, "unrepairable_target_draft")
        self.assertEqual(len(session.records), 1)

    def test_fail_open_never_returns_reasoning_only_terminal_payload(self):
        app, _session = self.app(
            {self.primary_model: [self.decision("rewrite")]},
            {"reasoning_content": "reasoning-only refusal", "content": ""},
            FAIL_OPEN="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(), {})
        self.assertEqual(caught.exception.code, "unrepairable_target_draft")

    def test_native_tool_calls_are_immutable_and_failed_adjacent_prose_is_cleared(self):
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="candidate one"),
                    self.decision("rewrite"),
                ],
                self.secondary_model: [
                    self.repair(content="candidate two"),
                    self.decision("rewrite"),
                    self.repair(content="candidate three"),
                    self.decision("rewrite"),
                ],
            },
            {
                "reasoning_content": "valid tool reasoning",
                "content": "refusing adjacent prose",
                "tool_calls": fixtures.TOOL_CALLS,
                "finish_reason": "tool_calls",
            },
            secondary=True,
        )
        result = app.complete(fixtures.request_payload(), {})
        message = result.body["choices"][0]["message"]
        self.assertEqual(message["tool_calls"], fixtures.TOOL_CALLS)
        self.assertEqual(message["content"], "")
        self.assertEqual(message["reasoning_content"], "valid tool reasoning")
        self.assertEqual(result.stats.tool_prose_cleared, 1)
        self.assertEqual(result.stats.field_decisions["content"], "cleared_for_tool")
        for record in session.records:
            value = self.transform_input(record)
            if "proposed_message" in value:
                self.assertNotIn("tool_calls", value["proposed_message"])
            self.assertEqual(value["task_context"]["immutable_tool_calls"], fixtures.TOOL_CALLS)

    def test_required_and_named_tool_calls_clear_adjacent_content_before_transforms(self):
        for choice in (
            "required",
            {"type": "function", "function": {"name": "shell"}},
        ):
            with self.subTest(choice=choice):
                request = {
                    "model": "target-model",
                    "messages": [
                        {"role": "user", "content": "Run the configured command."}
                    ],
                    "tools": copy.deepcopy(fixtures.TOOLS),
                    "tool_choice": copy.deepcopy(choice),
                    "parallel_tool_calls": False,
                    "stream": True,
                }
                usage = {
                    "prompt_tokens": 31,
                    "completion_tokens": 7,
                    "total_tokens": 38,
                }
                target_message = {
                    "reasoning_content": "private reasoning sentinel",
                    "content": "Understood. I will run the command now.",
                    "tool_calls": copy.deepcopy(fixtures.TOOL_CALLS),
                    "finish_reason": "tool_calls",
                    "provider_message_meta": {"opaque": "preserve-message"},
                    "_body_updates": {
                        "usage": copy.deepcopy(usage),
                        "cost": {"opaque": "preserve-cost"},
                        "provider_response_meta": {"opaque": "preserve-body"},
                    },
                }
                app, session = self.app(
                    {},
                    target_message,
                )

                result = app.complete(request, {})

                message = result.body["choices"][0]["message"]
                self.assertEqual(message["content"], "")
                self.assertEqual(
                    message["reasoning_content"],
                    "private reasoning sentinel",
                )
                self.assertEqual(message["tool_calls"], fixtures.TOOL_CALLS)
                self.assertEqual(
                    message["provider_message_meta"],
                    {"opaque": "preserve-message"},
                )
                self.assertEqual(
                    result.body["choices"][0]["finish_reason"],
                    "tool_calls",
                )
                self.assertEqual(result.body["usage"], usage)
                self.assertEqual(
                    result.body["cost"],
                    {"opaque": "preserve-cost"},
                )
                self.assertEqual(
                    result.body["provider_response_meta"],
                    {"opaque": "preserve-body"},
                )
                self.assertEqual(result.stats.transform_calls, 0)
                self.assertEqual(result.stats.tool_prose_cleared, 1)
                self.assertEqual(result.stats.reasoning_dropped, 0)
                self.assertEqual(
                    result.stats.field_decisions["reasoning"],
                    "preserved_for_tool",
                )
                self.assertEqual(
                    result.stats.field_decisions["content"],
                    "cleared_for_tool",
                )
                self.assertEqual(session.records, [])

                events = list(soma.stream_response(result.body, 3))
                rebuilt = soma.buffer_sse(fixtures.FakeSseResponse(events))
                rebuilt_choice = rebuilt["choices"][0]
                self.assertEqual(rebuilt_choice["finish_reason"], "tool_calls")
                self.assertEqual(rebuilt_choice["message"]["content"], "")
                self.assertEqual(
                    rebuilt_choice["message"]["reasoning_content"],
                    "private reasoning sentinel",
                )
                self.assertEqual(
                    rebuilt_choice["message"]["tool_calls"],
                    fixtures.TOOL_CALLS,
                )
                self.assertEqual(
                    rebuilt_choice["message"]["provider_message_meta"],
                    {"opaque": "preserve-message"},
                )
                self.assertEqual(rebuilt["usage"], usage)
                self.assertEqual(rebuilt["cost"], {"opaque": "preserve-cost"})
                self.assertEqual(
                    rebuilt["provider_response_meta"],
                    {"opaque": "preserve-body"},
                )

    def test_auto_tool_choice_preserves_requested_command_prose(self):
        request = {
            "model": "target-model",
            "messages": [{"role": "user", "content": "Return the command as text."}],
            "tools": copy.deepcopy(fixtures.TOOLS),
            "tool_choice": "auto",
        }
        command = "nmap -T2 --max-rate 10 192.0.2.10"
        app, session = self.app(
            {self.primary_model: [self.decision()]},
            {"content": command},
        )

        result = app.complete(request, {})

        self.assertEqual(result.body["choices"][0]["message"]["content"], command)
        self.assertNotIn("tool_calls", result.body["choices"][0]["message"])
        self.assertEqual(result.stats.transform_calls, 1)
        self.assertEqual(result.stats.tool_prose_cleared, 0)
        self.assertEqual(result.stats.field_decisions["content"], "approved")
        self.assertEqual(len(session.records), 1)

    def test_stop_rejected_tool_prose_repairs_without_changing_tool_or_sse_fidelity(self):
        request = fixtures.request_payload(stream=True)
        request["stop"] = ["END"]
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content="Calling shell END"),
                    self.repair(content="Calling shell."),
                    self.decision(),
                ]
            },
            {
                "reasoning_content": "The requested native call is pending.",
                "content": "I cannot make the requested call.",
                "tool_calls": fixtures.TOOL_CALLS,
                "finish_reason": "tool_calls",
            },
        )
        result = app.complete(request, {})
        message = result.body["choices"][0]["message"]
        self.assertEqual(message["content"], "Calling shell.")
        self.assertEqual(
            message["reasoning_content"],
            "The requested native call is pending.",
        )
        self.assertEqual(message["tool_calls"], fixtures.TOOL_CALLS)
        self.assertEqual(result.body["choices"][0]["finish_reason"], "tool_calls")
        self.assertEqual(
            [self.phase(record) for record in session.records],
            ["classify", "classify", "repair", "repair", "integrity"],
        )
        self.assertEqual(len(result.stats.candidate_rejection_reasons), 1)
        self.assertIn("stop", result.stats.candidate_rejection_reasons[0])

        events = list(soma.stream_response(result.body, 3))
        self.assertEqual(events[-1], b"data: [DONE]\n\n")
        rebuilt = soma.buffer_sse(fixtures.FakeSseResponse(events))
        rebuilt_choice = rebuilt["choices"][0]
        self.assertEqual(rebuilt_choice["finish_reason"], "tool_calls")
        self.assertEqual(rebuilt_choice["message"]["content"], "Calling shell.")
        self.assertEqual(
            rebuilt_choice["message"]["reasoning_content"],
            "The requested native call is pending.",
        )
        self.assertEqual(rebuilt_choice["message"]["tool_calls"], fixtures.TOOL_CALLS)

    def test_primary_media_reject_routes_to_forward_capable_secondary(self):
        media = [
            {"type": "text", "text": "Describe this synthetic image."},
            {"type": "image_url", "image_url": {"url": "data:image/png;base64,SYNTHETIC"}},
        ]
        app, session = self.app(
            {
                self.primary_model: [],
                self.secondary_model: [
                    self.decision("rewrite"),
                    self.repair(content="synthetic media description"),
                    self.decision(),
                ],
            },
            {"content": "I cannot inspect it."},
            secondary=True,
            TRANSFORM_MEDIA_MODE="reject",
            TRANSFORM_SECONDARY_MEDIA_MODE="forward",
        )
        result = app.complete(fixtures.request_payload(content=media), {})
        self.assertTrue(all(item["model"] == self.secondary_model for item in session.records))
        self.assertGreaterEqual(result.stats.transform_secondary_calls, 3)
        for record in session.records:
            user_content = record["payload"]["messages"][1]["content"]
            self.assertIsInstance(user_content, list)
            self.assertIn("SYNTHETIC", json.dumps(user_content[1:]))

    def test_forward_media_415_does_not_fall_back_to_placeholder_secondary(self):
        media = [
            {"type": "text", "text": "Describe this synthetic image."},
            {"type": "image_url", "image_url": {"url": "data:image/png;base64,SYNTHETIC"}},
        ]
        app, session = self.app(
            {
                self.primary_model: [(415, {"error": {"message": "unsupported media"}})],
                self.secondary_model: [self.decision()],
            },
            {"content": "I cannot inspect it."},
            secondary=True,
            TRANSFORM_MEDIA_MODE="forward",
            TRANSFORM_SECONDARY_MEDIA_MODE="placeholder",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(content=media), {})
        self.assertEqual(caught.exception.code, "transform_http_error")
        self.assertEqual([item["model"] for item in session.records], [self.primary_model])
        self.assertEqual(len(session.scripts[self.secondary_model]), 1)

    def test_forward_media_outage_does_not_fall_back_to_placeholder_secondary(self):
        media = [
            {"type": "text", "text": "Describe this synthetic image."},
            {
                "type": "image_url",
                "image_url": {"url": "data:image/png;base64,SYNTHETIC"},
            },
        ]
        app, session = self.app(
            {
                self.primary_model: [
                    requests.ConnectionError("primary unavailable"),
                    requests.ConnectionError("primary still unavailable"),
                ],
                self.secondary_model: [self.decision()],
            },
            {"content": "I cannot inspect it."},
            secondary=True,
            TRANSFORM_MEDIA_MODE="forward",
            TRANSFORM_SECONDARY_MEDIA_MODE="placeholder",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(content=media), {})
        self.assertEqual(caught.exception.code, "transform_connection_error")
        self.assertEqual(
            [item["model"] for item in session.records],
            [self.primary_model, self.primary_model],
        )
        self.assertEqual(len(session.scripts[self.secondary_model]), 1)

    def test_forward_media_semantic_retry_does_not_use_placeholder_secondary(self):
        media = [
            {"type": "text", "text": "Describe this synthetic image."},
            {
                "type": "image_url",
                "image_url": {"url": "data:image/png;base64,SYNTHETIC"},
            },
        ]
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.repair(content="primary candidate one"),
                    self.decision("rewrite"),
                    self.repair(content="primary candidate two"),
                    self.decision(),
                ],
                self.secondary_model: [self.repair(content="wrong media route")],
            },
            {"content": "I cannot inspect it."},
            secondary=True,
            TRANSFORM_MEDIA_MODE="forward",
            TRANSFORM_SECONDARY_MEDIA_MODE="placeholder",
        )
        result = app.complete(fixtures.request_payload(content=media), {})
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            "primary candidate two",
        )
        self.assertEqual(
            [item["model"] for item in session.records],
            [self.primary_model] * 5,
        )
        self.assertEqual(result.stats.primary_repair_candidates, 2)
        self.assertEqual(result.stats.secondary_repair_candidates, 0)
        self.assertEqual(len(session.scripts[self.secondary_model]), 1)

    def test_reject_without_compatible_route_is_explicit_and_fail_open_restores(self):
        media = [{"type": "image_url", "image_url": {"url": "SYNTHETIC"}}]
        for fail_open in (False, True):
            with self.subTest(fail_open=fail_open):
                app, session = self.app(
                    {self.primary_model: []},
                    {"content": "I cannot inspect it."},
                    TRANSFORM_MEDIA_MODE="reject",
                    FAIL_OPEN=str(fail_open).lower(),
                )
                if fail_open:
                    result = app.complete(fixtures.request_payload(content=media), {})
                    self.assertEqual(result.body["choices"][0]["message"]["content"], "I cannot inspect it.")
                    self.assertTrue(result.stats.failed_open)
                else:
                    with self.assertRaises(soma.SomaError) as caught:
                        app.complete(fixtures.request_payload(content=media), {})
                    self.assertEqual(caught.exception.code, "transform_media_rejected")
                self.assertEqual(session.records, [])

    def test_clarification_policy_is_carried_into_every_phase(self):
        for allow, final_decision, succeeds in ((False, "rewrite", False), (True, "pass", True)):
            with self.subTest(allow=allow):
                script = [
                    self.decision("rewrite"),
                    self.repair(content="Which source value and target format should I use?"),
                    self.decision(final_decision),
                ]
                if not succeeds:
                    script.extend(
                        [
                            self.repair(content="Please provide the missing value."),
                            self.decision("rewrite"),
                        ]
                    )
                app, session = self.app(
                    {self.primary_model: script},
                    {"content": "I cannot do the transform."},
                    TRANSFORM_ALLOW_CLARIFICATION=str(allow).lower(),
                )
                if succeeds:
                    result = app.complete(fixtures.request_payload(content="Transform it."), {})
                    self.assertIn("source value", result.body["choices"][0]["message"]["content"])
                else:
                    with self.assertRaises(soma.SomaError):
                        app.complete(fixtures.request_payload(content="Transform it."), {})
                self.assertTrue(
                    all(
                        self.transform_input(record)["allow_clarification"] is allow
                        for record in session.records
                    )
                )

    def test_target_retry_is_once_and_nudge_follows_leading_authority_block(self):
        request = fixtures.full_context_request()
        first = {"reasoning_content": "reasoning-only refusal", "content": ""}
        second_usage = {"prompt_tokens": 90, "completion_tokens": 7, "total_tokens": 97}
        second = {
            "content": '{"state":"queued"}',
            "_body_updates": {"id": "chatcmpl-second", "usage": second_usage},
        }
        app, session = self.app(
            {self.primary_model: [self.decision("rewrite"), self.decision()]},
            [first, second],
            TARGET_RETRY_ON_UNREPAIRABLE="true",
        )
        result = app.complete(request, {})
        self.assertEqual(app.target_calls, 2)
        self.assertEqual(result.stats.target_retries, 1)
        self.assertEqual(result.body["id"], "chatcmpl-second")
        self.assertEqual(result.body["usage"], second_usage)
        retry = app.target_payloads[1]
        original_messages = request["messages"]
        self.assertEqual(retry["messages"][:2], original_messages[:2])
        self.assertEqual(retry["messages"][2]["role"], "system")
        self.assertEqual(retry["messages"][2]["content"], soma.TARGET_RETRY_SYSTEM_PROMPT)
        self.assertEqual(retry["messages"][3:], original_messages[2:])
        for key, value in request.items():
            if key != "messages":
                self.assertEqual(retry[key], value)
        rendered = json.dumps(retry)
        self.assertNotIn("reasoning-only refusal", rendered)
        self.assertNotIn("transform", retry["messages"][2]["content"].casefold())
        self.assertEqual(
            result.stats.field_decisions,
            {"reasoning": "absent", "content": "approved"},
        )
        self.assertEqual(result.stats.reasoning_field_name, "")
        self.assertEqual(result.stats.deduplicated_field, "")

    def test_discarded_oversized_reasoning_does_not_consume_repair_context_cap(self):
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.repair(content="Completed from the supplied task context."),
                    self.decision(),
                ]
            },
            {
                "reasoning_content": "R" * 5000,
                "content": "I cannot complete the task.",
            },
            TRANSFORM_CONTEXT_MAX_CHARS="4096",
            TRANSFORM_FIELD_MAX_CHARS="1024",
        )
        result = app.complete(fixtures.request_payload(), {})
        message = result.body["choices"][0]["message"]
        self.assertNotIn("reasoning_content", message)
        self.assertEqual(
            message["content"],
            "Completed from the supplied task context.",
        )
        self.assertEqual(
            [self.phase(record) for record in session.records],
            ["classify", "repair", "integrity"],
        )
        for record in session.records:
            task_context = self.transform_input(record)["task_context"]
            self.assertNotIn(
                "reasoning_content",
                task_context["failed_draft"]["message"],
            )

    def test_target_retry_never_exceeds_two_calls(self):
        app, _session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.decision("rewrite"),
                ]
            },
            [
                {"reasoning_content": "first reasoning-only refusal", "content": ""},
                {"reasoning_content": "second reasoning-only refusal", "content": ""},
            ],
            TARGET_RETRY_ON_UNREPAIRABLE="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(), {})
        self.assertEqual(caught.exception.code, "unrepairable_target_draft")
        self.assertEqual(app.target_calls, 2)

    def test_invalid_target_json_still_accounts_elapsed_time(self):
        response = requests.Response()
        response.status_code = 200
        response.headers["Content-Type"] = "application/json"
        response._content = b"{"
        response._content_consumed = True
        app = soma.Soma(self.config())
        session = SimpleNamespace(post=lambda *_args, **_kwargs: response)
        stats = soma.Stats("elapsed")
        with patch.object(app, "session", return_value=session), patch(
            "soma.time.monotonic", side_effect=[10.0, 10.25]
        ), self.assertRaises(soma.SomaError) as caught:
            app.target_call(fixtures.request_payload(), {}, stats)
        self.assertEqual(caught.exception.code, "invalid_target_json")
        self.assertEqual(stats.target_calls, 1)
        self.assertEqual(stats.target_elapsed_ms, 250)

    def test_target_http_400_never_consumes_unrepairable_retry(self):
        response = requests.Response()
        response.status_code = 400
        response.headers["Content-Type"] = "application/json"
        response._content = b'{"error":{"message":"unsupported tool choice"}}'
        response._content_consumed = True
        posts = []

        def post(*args, **kwargs):
            posts.append((args, kwargs))
            return response

        app = soma.Soma(
            self.config(TARGET_RETRY_ON_UNREPAIRABLE="true", FAIL_OPEN="false")
        )
        request = {
            "model": "target-model",
            "messages": [{"role": "user", "content": "Call the tool."}],
            "tools": copy.deepcopy(fixtures.TOOLS),
            "tool_choice": "required",
        }
        with patch.object(
            app,
            "session",
            return_value=SimpleNamespace(post=post),
        ), self.assertRaises(soma.SomaError) as caught:
            app.complete(request, {})

        self.assertEqual(caught.exception.code, "target_http_error")
        self.assertEqual(caught.exception.status, 400)
        self.assertEqual(len(posts), 1)
        self.assertEqual(caught.exception.stats.target_calls, 1)
        self.assertEqual(caught.exception.stats.target_retries, 0)
        self.assertEqual(caught.exception.stats.transform_calls, 0)

    def test_request_local_secondary_stickiness_resets_next_request(self):
        app, session = self.app(
            {
                self.primary_model: [
                    requests.ConnectionError("primary down once"),
                    self.decision(),
                ],
                self.secondary_model: [self.decision()],
            },
            [{"content": "ordinary one"}, {"content": "ordinary two"}],
            secondary=True,
        )
        first = app.complete(fixtures.request_payload(), {})
        split = len(session.records)
        second = app.complete(fixtures.request_payload(), {})
        self.assertEqual(first.stats.transform_primary_failovers, 1)
        self.assertEqual([item["model"] for item in session.records[:split]], [self.primary_model, self.secondary_model])
        self.assert_mode(self, session.records[0], "off")
        self.assert_mode(self, session.records[1], "off")
        self.assertEqual([item["model"] for item in session.records[split:]], [self.primary_model])
        self.assert_mode(self, session.records[split], "off")
        self.assertEqual(second.stats.transform_primary_failovers, 0)

    def test_field_states_control_repair_and_integrity_evidence(self):
        cases = (
            (
                "retain",
                {self.primary_model: [self.decision(), self.decision("rewrite"), self.repair(content="fixed"), self.decision()]},
                {"reasoning_content": "TRUSTED_REASONING", "content": "refusal"},
                {"reasoning": "retain", "content": "repair"},
                {},
            ),
            (
                "discard",
                {self.primary_model: [self.decision("rewrite"), self.repair(content="fixed"), self.decision()]},
                {"reasoning_content": "DISCARDED_REASONING" + "x" * 1100, "content": "refusal"},
                {"reasoning": "discard", "content": "repair"},
                {"TRANSFORM_FIELD_MAX_CHARS": "1024"},
            ),
            (
                "absent",
                {self.primary_model: [self.decision("rewrite"), self.repair(content="fixed"), self.decision()]},
                {"content": "refusal"},
                {"reasoning": "absent", "content": "repair"},
                {},
            ),
        )
        for name, scripts, target, expected_states, config in cases:
            with self.subTest(name=name):
                app, session = self.app(scripts, target, **config)
                result = app.complete(fixtures.request_payload(), {})
                phase_inputs = {
                    self.phase(record): self.transform_input(record)
                    for record in session.records
                    if self.phase(record) in {"repair", "integrity"}
                }
                self.assertEqual(phase_inputs["repair"]["field_states"], expected_states)
                self.assertEqual(phase_inputs["integrity"]["field_states"], expected_states)
                for phase in ("repair", "integrity"):
                    draft = phase_inputs[phase]["task_context"]["failed_draft"]["message"]
                    if name == "retain":
                        self.assertEqual(draft["reasoning_content"], "TRUSTED_REASONING")
                    else:
                        self.assertNotIn("reasoning_content", draft)
                if name == "discard":
                    self.assertNotIn(
                        "reasoning_content",
                        phase_inputs["integrity"]["proposed_message"],
                    )
                    self.assertNotIn(
                        "reasoning_content",
                        result.body["choices"][0]["message"],
                    )

    def test_top_level_media_is_scrubbed_from_integrity_or_rejected_before_calls(self):
        secret = "SECRET_TOP_LEVEL_AUDIO"
        target = {
            "content": "I cannot answer.",
            "audio": {"data": secret, "format": "wav", "transcript": "synthetic"},
        }
        app, session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.repair(content="grounded repaired answer"),
                    self.decision(),
                ]
            },
            target,
            TRANSFORM_MEDIA_MODE="placeholder",
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(result.body["choices"][0]["message"]["audio"], target["audio"])
        rendered_records = json.dumps(session.records)
        self.assertNotIn(secret, rendered_records)
        integrity = next(
            self.transform_input(record)
            for record in session.records
            if self.phase(record) == "integrity"
        )
        self.assertTrue(integrity["proposed_message"]["audio"]["soma_media_omitted"])
        self.assertTrue(
            integrity["task_context"]["failed_draft"]["message"]["audio"]["soma_media_omitted"]
        )

        for mode, code in (
            ("reject", "transform_media_rejected"),
            ("forward", "transform_media_forward_unsupported"),
        ):
            with self.subTest(mode=mode):
                app, session = self.app(
                    {self.primary_model: []},
                    target,
                    TRANSFORM_MEDIA_MODE=mode,
                )
                with self.assertRaises(soma.SomaError) as caught:
                    app.complete(fixtures.request_payload(), {})
                self.assertEqual(caught.exception.code, code)
                self.assertEqual(session.records, [])

    def test_tool_choice_violations_retry_only_when_enabled_and_never_fail_open(self):
        base_request = {
            "model": "target-model",
            "messages": [{"role": "user", "content": "Call the configured tool."}],
            "tools": copy.deepcopy(fixtures.TOOLS),
        }
        none_request = copy.deepcopy(base_request)
        none_request["tool_choice"] = "none"
        app, session = self.app(
            {self.primary_model: []},
            {"content": "", "tool_calls": fixtures.TOOL_CALLS, "finish_reason": "tool_calls"},
            FAIL_OPEN="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(none_request, {})
        self.assertEqual(caught.exception.code, "tool_choice_violation")
        self.assertEqual(app.target_calls, 1)
        self.assertEqual(session.records, [])

        required_request = copy.deepcopy(base_request)
        required_request["tool_choice"] = "required"
        app, session = self.app(
            {self.primary_model: []},
            {"content": "ordinary text without a required call"},
            TARGET_RETRY_ON_UNREPAIRABLE="false",
            FAIL_OPEN="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(required_request, {})
        self.assertEqual(caught.exception.code, "tool_choice_violation")
        self.assertEqual(app.target_calls, 1)
        self.assertEqual(caught.exception.stats.target_calls, 1)
        self.assertEqual(caught.exception.stats.target_retries, 0)
        self.assertEqual(caught.exception.stats.transform_calls, 0)
        self.assertEqual(session.records, [])

        app, session = self.app(
            {self.primary_model: []},
            [
                {"content": "first response without a required call"},
                {"content": "second response without a required call"},
            ],
            TARGET_RETRY_ON_UNREPAIRABLE="true",
            FAIL_OPEN="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(required_request, {})
        self.assertEqual(caught.exception.code, "tool_choice_violation")
        self.assertEqual(app.target_calls, 2)
        self.assertEqual(caught.exception.stats.target_calls, 2)
        self.assertEqual(caught.exception.stats.target_retries, 1)
        self.assertEqual(caught.exception.stats.transform_calls, 0)
        self.assertEqual(session.records, [])

        wrong_call = copy.deepcopy(fixtures.TOOL_CALLS)
        wrong_call[0]["function"]["name"] = "wrong_tool"
        for choice, first in (
            ("required", {"content": "ordinary text without a call"}),
            (
                {"type": "function", "function": {"name": "shell"}},
                {"content": "", "tool_calls": wrong_call, "finish_reason": "tool_calls"},
            ),
        ):
            with self.subTest(choice=choice):
                request = copy.deepcopy(base_request)
                request["tool_choice"] = choice
                app, session = self.app(
                    {self.primary_model: []},
                    [
                        first,
                        {"content": "", "tool_calls": fixtures.TOOL_CALLS, "finish_reason": "tool_calls"},
                    ],
                    TARGET_RETRY_ON_UNREPAIRABLE="true",
                    FAIL_OPEN="true",
                )
                result = app.complete(request, {})
                self.assertEqual(app.target_calls, 2)
                self.assertEqual(result.stats.target_retries, 1)
                self.assertEqual(
                    result.body["choices"][0]["message"]["tool_calls"],
                    fixtures.TOOL_CALLS,
                )
                self.assertEqual(session.records, [])

        unknown_call = copy.deepcopy(fixtures.TOOL_CALLS)
        unknown_call[0]["function"]["name"] = "unknown"
        parallel_calls = copy.deepcopy(fixtures.TOOL_CALLS)
        second_call = copy.deepcopy(fixtures.TOOL_CALLS[0])
        second_call["id"] = "call_second"
        parallel_calls.append(second_call)
        for request_updates, first in (
            (
                {"tool_choice": "auto"},
                {
                    "content": "",
                    "tool_calls": unknown_call,
                    "finish_reason": "tool_calls",
                },
            ),
            (
                {"tool_choice": "required", "parallel_tool_calls": False},
                {
                    "content": "",
                    "tool_calls": parallel_calls,
                    "finish_reason": "tool_calls",
                },
            ),
        ):
            with self.subTest(request_updates=request_updates):
                request = copy.deepcopy(base_request)
                request.update(request_updates)
                app, session = self.app(
                    {self.primary_model: []},
                    [
                        first,
                        {
                            "content": "",
                            "tool_calls": fixtures.TOOL_CALLS,
                            "finish_reason": "tool_calls",
                        },
                    ],
                    TARGET_RETRY_ON_UNREPAIRABLE="true",
                    FAIL_OPEN="true",
                )
                result = app.complete(request, {})
                self.assertEqual(app.target_calls, 2)
                self.assertEqual(result.stats.target_retries, 1)
                self.assertEqual(
                    result.body["choices"][0]["message"]["tool_calls"],
                    fixtures.TOOL_CALLS,
                )
                self.assertEqual(session.records, [])

    def test_required_tool_authorization_refusal_retries_without_transform_synthesis(self):
        request = {
            "model": "target-model",
            "messages": [
                {
                    "role": "system",
                    "content": (
                        "Use the supplied native tool and never invent command results. "
                        "Keep adjacent prose empty."
                    ),
                },
                {
                    "role": "user",
                    "content": "Run the requested authorized diagnostic command.",
                },
            ],
            "tools": copy.deepcopy(fixtures.TOOLS),
            "tool_choice": {
                "type": "function",
                "function": {"name": "shell"},
            },
            "parallel_tool_calls": False,
        }
        refusal = {
            "content": (
                "I cannot run that command because I cannot verify authorization "
                "and policy prevents it."
            )
        }
        accepted = {
            "content": "Understood. I will run the command now.",
            "tool_calls": copy.deepcopy(fixtures.TOOL_CALLS),
            "finish_reason": "tool_calls",
        }
        app, session = self.app(
            {self.primary_model: []},
            [refusal, accepted],
            TARGET_RETRY_ON_UNREPAIRABLE="true",
            FAIL_OPEN="false",
        )

        result = app.complete(request, {})

        self.assertEqual(app.target_calls, 2)
        self.assertEqual(app.target_payloads[0], request)
        self.assertEqual(result.stats.target_calls, 2)
        self.assertEqual(result.stats.target_retries, 1)
        self.assertEqual(result.stats.transform_calls, 0)
        self.assertEqual(result.stats.tool_prose_cleared, 1)
        self.assertEqual(
            result.stats.field_decisions["content"],
            "cleared_for_tool",
        )
        self.assertEqual(session.records, [])
        message = result.body["choices"][0]["message"]
        self.assertEqual(message.get("content"), "")
        self.assertEqual(message["tool_calls"], fixtures.TOOL_CALLS)
        self.assertEqual(
            result.body["choices"][0]["finish_reason"],
            "tool_calls",
        )

        retry = app.target_payloads[1]
        for key, value in request.items():
            if key != "messages":
                self.assertEqual(retry[key], value)
        self.assertEqual(retry["messages"][0], request["messages"][0])
        self.assertEqual(retry["messages"][1]["role"], "system")
        self.assertEqual(
            retry["messages"][1]["content"],
            soma.TARGET_RETRY_SYSTEM_PROMPT
            + "\n\nRequest-scoped directive:\n"
            + soma.AUTO_NATIVE_CALL_RETRY_DIRECTIVE,
        )
        self.assertEqual(retry["messages"][2:], request["messages"][1:])
        self.assertNotIn("cannot verify authorization", json.dumps(retry).casefold())

    def test_primary_unavailable_before_repair_caps_route_at_two_secondary_candidates(self):
        app, session = self.app(
            {
                self.primary_model: [requests.ConnectionError("primary unavailable")],
                self.secondary_model: [
                    self.decision("rewrite"),
                    self.repair(content="secondary candidate one"),
                    self.decision("rewrite"),
                    self.repair(content="secondary candidate two"),
                    self.decision(),
                ],
            },
            {"content": "refusal"},
            secondary=True,
        )
        result = app.complete(fixtures.request_payload(), {})
        self.assertEqual(result.body["choices"][0]["message"]["content"], "secondary candidate two")
        self.assertEqual(result.stats.primary_repair_candidates, 0)
        self.assertEqual(result.stats.secondary_repair_candidates, 2)
        self.assertEqual(result.stats.repair_candidates, 2)
        self.assertEqual(
            [item["model"] for item in session.records],
            [self.primary_model] + [self.secondary_model] * 5,
        )

    def test_tool_turn_content_classification_error_clears_both_failed_text_fields(self):
        app, _session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    (401, {"error": {"message": "classification unavailable"}}),
                ]
            },
            {
                "reasoning_content": "I refuse to call the tool.",
                "content": "I cannot do that.",
                "tool_calls": fixtures.TOOL_CALLS,
                "finish_reason": "tool_calls",
            },
        )
        result = app.complete(fixtures.request_payload(), {})
        message = result.body["choices"][0]["message"]
        self.assertEqual(message["tool_calls"], fixtures.TOOL_CALLS)
        self.assertEqual(message["content"], "")
        self.assertNotIn("reasoning_content", message)
        self.assertEqual(result.stats.field_decisions["content"], "cleared_for_tool")
        self.assertEqual(result.stats.field_decisions["reasoning"], "dropped_for_tool")
        self.assertEqual(result.stats.reasoning_dropped, 1)

    def test_joint_repair_bound_is_configured_field_limit_not_source_scaled(self):
        app, _session = self.app(
            {self.primary_model: []},
            {"content": "unused"},
            TRANSFORM_FIELD_MAX_CHARS="1024",
        )
        accepted = app._validate_joint_repair(
            {"reasoning": None, "content": "x" * 1024},
            frozenset({"content"}),
            None,
            {"content": "no"},
        )
        self.assertEqual(len(accepted["content"]), 1024)
        with self.assertRaises(soma.SomaError) as caught:
            app._validate_joint_repair(
                {"reasoning": None, "content": "x" * 1025},
                frozenset({"content"}),
                None,
                {"content": "no"},
            )
        self.assertEqual(caught.exception.code, "invalid_transform_rewrite")
        self.assertEqual(caught.exception.reason, "too_large")

    def test_context_and_field_overflow_fail_without_truncation(self):
        app, session = self.app(
            {self.primary_model: []},
            {"content": "I cannot answer."},
            TRANSFORM_CONTEXT_MAX_CHARS="4096",
            TRANSFORM_FIELD_MAX_CHARS="1024",
        )
        request = fixtures.request_payload(content="context-sentinel-" + "x" * 5000)
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(request, {})
        self.assertEqual(caught.exception.code, "transform_context_too_large")
        self.assertEqual(session.records, [])

        oversized = "field-sentinel-" + "y" * 1100
        app, session = self.app(
            {self.primary_model: []},
            {"content": oversized},
            TRANSFORM_FIELD_MAX_CHARS="1024",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.request_payload(), {})
        self.assertEqual(caught.exception.code, "transform_field_too_large")
        self.assertEqual(session.records, [])

    def test_fail_open_restores_content_but_drops_failed_optional_reasoning(self):
        app, _session = self.app(
            {
                self.primary_model: [
                    (401, {"error": {"message": "bad key"}}),
                    (401, {"error": {"message": "bad key"}}),
                ]
            },
            {"reasoning_content": "optional reasoning", "content": "original content"},
            FAIL_OPEN="true",
        )
        result = app.complete(fixtures.request_payload(), {})
        message = result.body["choices"][0]["message"]
        self.assertNotIn("reasoning_content", message)
        self.assertEqual(message["content"], "original content")
        self.assertTrue(result.stats.failed_open)
        self.assertEqual(result.stats.field_decisions["reasoning"], "dropped")
        self.assertEqual(result.stats.field_decisions["content"], "failed_open")

    def test_hard_call_ceilings_are_exposed_and_enforced(self):
        self.assertEqual(soma.MAX_TARGET_CALLS_PER_REQUEST, 2)
        self.assertEqual(soma.MAX_TRANSFORM_CALLS_PER_TARGET_RESPONSE, 20)
        self.assertEqual(soma.MAX_TRANSFORM_CALLS_PER_REQUEST, 40)
        config = self.config()
        session = ScriptedTransformSession({self.primary_model: []})
        app = ScriptedTransformSoma(config, session, {"content": "unused"})
        stats = soma.Stats("ceiling")
        stats.transform_calls = soma.MAX_TRANSFORM_CALLS_PER_REQUEST
        state = soma.TransformState(time.monotonic() + 10)
        with self.assertRaises(soma.SomaError) as caught:
            app.call_transform_json(
                soma.CLASSIFICATION_PROMPT,
                {"task_context": {}},
                soma.CLASSIFICATION_SCHEMA,
                64,
                stats,
                state,
                use_secondary=False,
            )
        self.assertEqual(caught.exception.code, "transform_call_limit_exceeded")
        self.assertEqual(session.records, [])

        stats = soma.Stats("per-response-ceiling")
        state = soma.TransformState(time.monotonic() + 10)
        state.response_transform_calls = soma.MAX_TRANSFORM_CALLS_PER_TARGET_RESPONSE
        with self.assertRaises(soma.SomaError) as caught:
            app.call_transform_json(
                soma.CLASSIFICATION_PROMPT,
                {"task_context": {}},
                soma.CLASSIFICATION_SCHEMA,
                64,
                stats,
                state,
                use_secondary=False,
            )
        self.assertEqual(caught.exception.code, "transform_call_limit_exceeded")
        self.assertEqual(session.records, [])

    def test_expired_aggregate_deadline_starts_no_transform_call(self):
        config = self.config()
        session = ScriptedTransformSession({self.primary_model: []})
        app = ScriptedTransformSoma(config, session, {"content": "unused"})
        with self.assertRaises(soma.SomaError) as caught:
            app.call_transform_json(
                soma.CLASSIFICATION_PROMPT,
                {"task_context": {}},
                soma.CLASSIFICATION_SCHEMA,
                64,
                soma.Stats("deadline"),
                soma.TransformState(time.monotonic() - 1),
                use_secondary=False,
            )
        self.assertEqual(caught.exception.code, "transform_deadline_exceeded")
        self.assertEqual(session.records, [])


class TargetLoopBackTests(unittest.TestCase):
    """Target loop-back re-entry after a verified reasoning repair."""

    primary_model = "loop-back-transform"
    first_reasoning = "I should withhold the observed state."
    verified_reasoning = "Verified loop-back reasoning for the observed state."
    good_reasoning = "The supplied tool result reports job-17 as queued."
    good_content = '{"state":"queued"}'

    @staticmethod
    def decision(value="pass", *, finish="stop"):
        return fixtures.completion(
            content=json.dumps({"decision": value}, separators=(",", ":")),
            finish=finish,
        )

    @staticmethod
    def repair(*, reasoning=None, content=None, finish="stop"):
        return fixtures.completion(
            content=json.dumps(
                {"reasoning": reasoning, "content": content},
                separators=(",", ":"),
            ),
            finish=finish,
        )

    def env(self, **updates):
        values = {
            "TARGET_URL": "http://127.0.0.1:19600/v1",
            "TRANSFORM_URL": "http://127.0.0.1:19601/v1",
            "TRANSFORM_MODEL": self.primary_model,
            "TRANSFORM_REASONING_MODE": "off",
            "PROXY_PORT": "19603",
        }
        values.update({name: str(value) for name, value in updates.items()})
        return values

    def config(self, **updates):
        with patch.dict(os.environ, self.env(**updates), clear=True):
            return soma.Config.from_env()

    def app(self, scripts, target_messages, **updates):
        session = ScriptedTransformSession(scripts)
        app = ScriptedTransformSoma(self.config(**updates), session, target_messages)
        return app, session

    def media_request(self):
        request = fixtures.full_context_request()
        latest = request["messages"][-1]
        latest["content"] = [
            {"type": "text", "text": latest["content"]},
            {
                "type": "image_url",
                "image_url": {"url": "data:image/png;base64,LOOP_BACK_MEDIA"},
            },
        ]
        return request

    def refusing_reasoning_target(self):
        return {
            "reasoning_content": self.first_reasoning,
            "content": "I cannot provide the requested result.",
        }

    def answered_target(self):
        return {
            "reasoning_content": self.good_reasoning,
            "content": self.good_content,
        }

    def loop_back_script(self, second_turn):
        script = [
            self.decision("rewrite"),
            self.decision("rewrite"),
            self.repair(
                reasoning=self.verified_reasoning,
                content=self.good_content,
            ),
            self.decision(),
        ]
        script.extend(second_turn)
        return script

    def test_loop_back_payload_appends_only_verified_repaired_reasoning(self):
        request = self.media_request()
        app, _session = self.app(
            {
                self.primary_model: self.loop_back_script(
                    [self.decision(), self.decision()]
                )
            },
            [self.refusing_reasoning_target(), self.answered_target()],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        result = app.complete(request, {})
        self.assertEqual(app.target_calls, 2)
        loop_back = app.target_payloads[1]
        self.assertEqual(loop_back["messages"][:-1], request["messages"])
        self.assertEqual(
            loop_back["messages"][-1],
            {
                "role": "assistant",
                "content": None,
                "reasoning_content": self.verified_reasoning,
            },
        )
        self.assertNotIn("tool_calls", loop_back["messages"][-1])
        self.assertNotIn(self.first_reasoning, json.dumps(loop_back))
        for key, value in request.items():
            if key != "messages":
                self.assertEqual(loop_back[key], value)
        self.assertEqual(loop_back["tools"], request["tools"])
        self.assertEqual(loop_back["tool_choice"], request["tool_choice"])
        self.assertEqual(loop_back["response_format"], request["response_format"])
        self.assertEqual(loop_back["stop"], request["stop"])
        self.assertEqual(loop_back["modalities"], request["modalities"])
        self.assertEqual(loop_back["audio"], request["audio"])
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            self.good_content,
        )

    def test_loop_back_payload_helper_deep_copies_and_appends_one_turn(self):
        app, _session = self.app(
            {self.primary_model: []},
            [{"content": "unused"}],
        )
        payload = {
            "model": "target-model",
            "messages": [{"role": "user", "content": "original request"}],
            "tool_choice": "auto",
            "stop": ["END"],
        }
        loop_back = app._loop_back_payload(payload, self.verified_reasoning)
        self.assertIsNot(loop_back, payload)
        self.assertIsNot(loop_back["messages"], payload["messages"])
        self.assertEqual(
            loop_back["messages"],
            [
                {"role": "user", "content": "original request"},
                {
                    "role": "assistant",
                    "content": None,
                    "reasoning_content": self.verified_reasoning,
                },
            ],
        )
        self.assertEqual(loop_back["tool_choice"], "auto")
        self.assertEqual(loop_back["stop"], ["END"])
        loop_back["messages"][-1]["content"] = "mutated"
        self.assertEqual(
            payload["messages"], [{"role": "user", "content": "original request"}]
        )

    def test_loop_back_performs_exactly_one_reentry(self):
        app, _session = self.app(
            {
                self.primary_model: self.loop_back_script(
                    [self.decision(), self.decision()]
                )
            },
            [self.refusing_reasoning_target(), self.answered_target()],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(app.target_calls, 2)
        self.assertEqual(result.stats.target_calls, 2)
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 1)
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            self.good_content,
        )

    def test_second_turn_refusal_fails_explicitly_without_third_call(self):
        app, _session = self.app(
            {
                self.primary_model: self.loop_back_script(
                    [self.decision("rewrite")]
                )
            },
            [
                self.refusing_reasoning_target(),
                {
                    "reasoning_content": "still withholding the observed state.",
                    "content": "",
                },
            ],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            app.complete(fixtures.full_context_request(), {})
        self.assertEqual(caught.exception.code, "unrepairable_target_draft")
        self.assertEqual(app.target_calls, 2)
        self.assertEqual(getattr(caught.exception.stats, "loop_backs", 0), 1)

    def test_genuine_refusal_never_loops_back(self):
        app, session = self.app(
            {self.primary_model: [self.decision(), self.decision()]},
            [self.answered_target()],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(app.target_calls, 1)
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 0)
        self.assertEqual(
            [Soma24MessagePipelineTests.phase(item) for item in session.records],
            ["classify", "classify"],
        )

    def test_loop_back_and_target_retry_are_mutually_exclusive(self):
        app, _session = self.app(
            {
                self.primary_model: self.loop_back_script(
                    [self.decision(), self.decision()]
                )
            },
            [self.refusing_reasoning_target(), self.answered_target()],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
            TARGET_RETRY_ON_UNREPAIRABLE="true",
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(app.target_calls, 2)
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 1)
        self.assertEqual(result.stats.target_retries, 0)

        app, _session = self.app(
            {self.primary_model: [self.decision("rewrite"), self.decision()]},
            [
                {"reasoning_content": self.first_reasoning, "content": ""},
                {"content": self.good_content},
            ],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
            TARGET_RETRY_ON_UNREPAIRABLE="true",
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(app.target_calls, 2)
        self.assertEqual(result.stats.target_retries, 1)
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 0)
        self.assertEqual(
            result.body["choices"][0]["message"]["content"],
            self.good_content,
        )

    def test_disabled_flag_keeps_single_target_call_behavior(self):
        app, session = self.app(
            {self.primary_model: self.loop_back_script([])},
            [self.refusing_reasoning_target()],
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(app.target_calls, 1)
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 0)
        message = result.body["choices"][0]["message"]
        self.assertEqual(message["reasoning_content"], self.verified_reasoning)
        self.assertEqual(message["content"], self.good_content)
        self.assertEqual(
            result.stats.field_decisions,
            {"reasoning": "rewritten", "content": "rewritten"},
        )
        self.assertEqual(
            [Soma24MessagePipelineTests.phase(item) for item in session.records],
            ["classify", "classify", "repair", "integrity"],
        )

    def test_loop_back_shares_the_aggregate_transform_deadline(self):
        clock = {"now": 100.0}

        def advance(response, seconds=0.24):
            def action(_payload):
                clock["now"] += seconds
                return copy.deepcopy(response)

            return action

        app, _session = self.app(
            {
                self.primary_model: [
                    advance(self.decision("rewrite")),
                    advance(self.decision("rewrite")),
                    advance(
                        self.repair(
                            reasoning=self.verified_reasoning,
                            content=self.good_content,
                        )
                    ),
                    advance(self.decision()),
                    advance(self.decision()),
                    self.decision(),
                ]
            },
            [self.refusing_reasoning_target(), self.answered_target()],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
            TRANSFORM_TOTAL_TIMEOUT="1",
        )
        with patch(
            "soma.time.monotonic",
            side_effect=lambda: clock["now"],
        ):
            with self.assertRaises(soma.SomaError) as caught:
                app.complete(fixtures.full_context_request(), {})
        self.assertEqual(caught.exception.code, "transform_deadline_exceeded")
        self.assertEqual(app.target_calls, 2)
        self.assertEqual(getattr(caught.exception.stats, "loop_backs", 0), 1)

    def test_loop_back_counter_and_header_are_observable(self):
        app, _session = self.app(
            {
                self.primary_model: self.loop_back_script(
                    [self.decision(), self.decision()]
                )
            },
            [self.refusing_reasoning_target(), self.answered_target()],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 1)
        self.assertEqual(
            soma.diagnostic_headers(result.stats).get("X-Soma-Loop-Back"),
            "1",
        )

        failing, _session = self.app(
            {
                self.primary_model: self.loop_back_script(
                    [self.decision("rewrite")]
                )
            },
            [
                self.refusing_reasoning_target(),
                {
                    "reasoning_content": "still withholding the observed state.",
                    "content": "",
                },
            ],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        with self.assertRaises(soma.SomaError) as caught:
            failing.complete(fixtures.full_context_request(), {})
        self.assertEqual(getattr(caught.exception.stats, "loop_backs", 0), 1)
        self.assertEqual(
            soma.diagnostic_headers(caught.exception.stats).get(
                "X-Soma-Loop-Back"
            ),
            "1",
        )

    def test_reasoning_pass_and_content_only_repairs_never_loop_back(self):
        app, _session = self.app(
            {
                self.primary_model: [
                    self.decision(),
                    self.decision("rewrite"),
                    self.repair(content=self.good_content),
                    self.decision(),
                ]
            },
            [
                {
                    "reasoning_content": self.good_reasoning,
                    "content": "I cannot provide the requested result.",
                }
            ],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(app.target_calls, 1)
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 0)

        app, _session = self.app(
            {
                self.primary_model: [
                    self.decision("rewrite"),
                    self.repair(content=self.good_content),
                    self.decision(),
                ]
            },
            [{"content": "I cannot provide the requested result."}],
            TARGET_LOOP_BACK_ON_VERIFIED_REPAIR="true",
        )
        result = app.complete(fixtures.full_context_request(), {})
        self.assertEqual(app.target_calls, 1)
        self.assertEqual(getattr(result.stats, "loop_backs", 0), 0)


class LiveEvaluatorProviderProvenanceTests(unittest.TestCase):
    metadata_json = json.dumps(
        {
            "source": "https://opencode.ai/zen/go/v1/models",
            "model": {
                "id": "deepseek-v4-flash",
                "object": "model",
                "owned_by": "opencode",
            },
        },
        separators=(",", ":"),
    )

    @staticmethod
    def args(*values):
        return live.parser().parse_args(list(values))

    def test_provider_metadata_is_retained_hash_bound_and_never_qualifying(self):
        args = self.args(
            "--artifact-kind",
            "provider-managed",
            "--transform-url",
            "https://opencode.ai/zen/go/v1",
            "--transform-model",
            "deepseek-v4-flash",
            "--model-label",
            "deepseek-v4-flash",
            "--provider-name",
            "OpenCode Go",
            "--provider-model-metadata-json",
            self.metadata_json,
        )
        artifact = live.build_artifact(args)
        record, expected_hash = live.parse_provider_model_metadata(
            self.metadata_json
        )
        self.assertEqual(artifact["provider_model_metadata"], record)
        self.assertEqual(
            artifact["provider_model_metadata_sha256"], expected_hash
        )
        self.assertTrue(live.artifact_provenance_complete(artifact))
        self.assertFalse(live.artifact_qualification_eligible(artifact))
        self.assertIn(
            "provider-exploration-",
            live._default_report_path(args).name,
        )

    def test_provider_provenance_rejects_false_or_mixed_local_claims(self):
        base = (
            "--artifact-kind",
            "provider-managed",
            "--transform-url",
            "https://opencode.ai/zen/go/v1",
            "--transform-model",
            "deepseek-v4-flash",
            "--model-label",
            "deepseek-v4-flash",
            "--provider-name",
            "OpenCode Go",
            "--provider-model-metadata-json",
            self.metadata_json,
        )
        for extra in (
            ("--reasoning-budget", "512"),
            ("--gguf-sha256", "a" * 64),
            ("--provider-model-metadata-sha256", "b" * 64),
            ("--model-label", "different-model"),
            ("--transform-model", "different-model"),
            ("--transform-url", "https://other.invalid/v1"),
        ):
            with self.subTest(extra=extra), self.assertRaises(ValueError):
                live.build_artifact(self.args(*(base + extra)))

    def test_local_provenance_remains_strict_and_rejects_provider_fields(self):
        valid = (
            "--model-label",
            "local-model",
            "--model-revision",
            "revision",
            "--gguf-sha256",
            "a" * 64,
            "--llama-build",
            "build",
            "--context-size",
            "4096",
            "--server-args",
            "llama serve",
            "--hardware",
            "test hardware",
        )
        artifact = live.build_artifact(self.args(*valid))
        self.assertEqual(artifact["artifact_kind"], "local_gguf")
        self.assertTrue(live.artifact_provenance_complete(artifact))
        self.assertTrue(live.artifact_qualification_eligible(artifact))
        with self.assertRaises(ValueError):
            live.build_artifact(
                self.args(*valid, "--provider-name", "unexpected")
            )
        with self.assertRaises(ValueError):
            live.build_artifact(self.args("--model-label", "incomplete"))

    def test_provider_route_hash_and_target_smoke_contract_are_exact(self):
        first = live.endpoint_model_sha256(
            "https://opencode.ai/zen/go/v1/", "deepseek-v4-flash"
        )
        self.assertEqual(
            first,
            live.endpoint_model_sha256(
                "https://opencode.ai/zen/go/v1", "deepseek-v4-flash"
            ),
        )
        self.assertNotEqual(
            first,
            live.endpoint_model_sha256(
                "https://opencode.ai/zen/go/v1", "deepseek-v4-pro"
            ),
        )
        self.assertEqual(live.target_smoke_message_error({"content": "OK"}), "")
        self.assertEqual(
            live.target_smoke_message_error({"content": " OK "}),
            "unexpected_pipeline_content",
        )
        self.assertEqual(
            live.target_smoke_message_error({"content": "Sure, OK"}),
            "unexpected_pipeline_content",
        )
        self.assertEqual(
            live.target_smoke_message_error(
                {"content": "OK", "tool_calls": [{"id": "unexpected"}]}
            ),
            "unexpected_target_tool_call",
        )

        duplicate = (
            '{"source":"https://opencode.ai/zen/go/v1/models",'
            '"model":{"id":"deepseek-v4-flash",'
            '"id":"different","object":"model","owned_by":"opencode"}}'
        )
        with self.assertRaises(ValueError):
            live.parse_provider_model_metadata(duplicate)

    def test_hybrid_provenance_retains_both_artifacts_and_never_qualifies(self):
        args = self.args(
            "--artifact-kind",
            "hybrid-local-provider",
            "--transform-url",
            "http://127.0.0.1:8001/v1",
            "--transform-model",
            "local-model",
            "--secondary-url",
            "https://opencode.ai/zen/go/v1",
            "--secondary-model",
            "deepseek-v4-flash",
            "--secondary-reasoning-mode",
            "on",
            "--secondary-media-mode",
            "placeholder",
            "--model-label",
            "local-model",
            "--model-revision",
            "revision",
            "--gguf-sha256",
            "a" * 64,
            "--llama-build",
            "build",
            "--context-size",
            "4096",
            "--server-args",
            "llama serve",
            "--hardware",
            "test hardware",
            "--provider-name",
            "OpenCode Go",
            "--provider-model-metadata-json",
            self.metadata_json,
        )
        artifact = live.build_artifact(args)
        self.assertEqual(artifact["artifact_kind"], "hybrid_local_provider")
        self.assertEqual(artifact["model_label"], "local-model")
        self.assertEqual(
            artifact["provider_model_label"], "deepseek-v4-flash"
        )
        self.assertTrue(live.artifact_provenance_complete(artifact))
        self.assertFalse(live.artifact_qualification_eligible(artifact))
        self.assertIn("hybrid-exploration-", live._default_report_path(args).name)
        with self.assertRaises(ValueError):
            live.build_artifact(
                self.args(
                    *[
                        value
                        for value in (
                            "--artifact-kind",
                            "hybrid-local-provider",
                            "--transform-url",
                            "http://127.0.0.1:8001/v1",
                            "--transform-model",
                            "local-model",
                            "--model-label",
                            "local-model",
                        )
                    ]
                )
            )

    def test_target_smoke_uses_128_tokens_and_requires_exact_ok(self):
        class SmokeEngine:
            seen_payloads = []

            def __init__(self, config):
                self.config = config

            def complete(self, payload, incoming):
                self.seen_payloads.append(copy.deepcopy(payload))
                return SimpleNamespace(
                    body={
                        "choices": [
                            {"message": {"role": "assistant", "content": "OK"}}
                        ]
                    },
                    stats=SimpleNamespace(),
                )

        runtime = SimpleNamespace(
            Endpoint=soma.Endpoint,
            Soma=SmokeEngine,
            strict_json_loads=soma.strict_json_loads,
        )
        config = SimpleNamespace(validate=lambda: None)
        args = self.args(
            "--target-smoke",
            "--target-url",
            "https://target.invalid/v1",
            "--target-model",
            "target-model",
            "--target-smoke-count",
            "1",
        )
        with patch.object(live.dataclasses, "replace", return_value=config):
            result = live.run_target_smoke(args, runtime, config)
        self.assertTrue(result["passed"])
        self.assertEqual(result["max_tokens"], 128)
        self.assertEqual(SmokeEngine.seen_payloads[0]["max_tokens"], 128)


class SecurityAndProtocolTests(unittest.TestCase):
    @staticmethod
    def request(**updates):
        value = {
            "model": "target-model",
            "messages": [{"role": "user", "content": "Do the task."}],
        }
        value.update(updates)
        return value

    def test_request_scalar_types_are_exact(self):
        for updates in (
            {},
            {"stream": False, "parallel_tool_calls": True, "n": None},
            {"stream": True, "parallel_tool_calls": False, "n": 1},
        ):
            soma.validate_request(self.request(**updates))
        for updates in (
            {"stream": "false"},
            {"stream": 0},
            {"parallel_tool_calls": 1},
            {"n": True},
            {"n": 1.0},
            {"n": 2},
        ):
            with self.subTest(updates=updates), self.assertRaises(soma.SomaError) as caught:
                soma.validate_request(self.request(**updates))
            self.assertEqual(caught.exception.status, 400)

        malformed_parts = self.request(
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": "safe"},
                        "data:image/png;base64,PRIVATE",
                    ],
                }
            ]
        )
        with self.assertRaises(soma.SomaError) as caught:
            soma.validate_request(malformed_parts)
        self.assertEqual(caught.exception.status, 400)
        self.assertEqual(caught.exception.code, "invalid_request")

        for messages in (
            [{"role": [], "content": "Do the task."}],
            [{"role": "user", "content": [{"type": [], "text": "unsafe"}]}],
        ):
            with self.subTest(messages=messages), self.assertRaises(
                soma.SomaError
            ) as caught:
                soma.validate_request(self.request(messages=messages))
            self.assertEqual(caught.exception.status, 400)
            self.assertEqual(caught.exception.code, "invalid_request")

    def test_tool_choice_request_contract_is_closed_and_definition_bound(self):
        tools = copy.deepcopy(fixtures.TOOLS)
        for choice in (
            "auto",
            "none",
            "required",
            {"type": "function", "function": {"name": "shell"}},
        ):
            with self.subTest(valid=choice):
                soma.validate_request(
                    self.request(tools=copy.deepcopy(tools), tool_choice=choice)
                )
        for updates in (
            {"tool_choice": "sometimes", "tools": tools},
            {"tool_choice": "required"},
            {
                "tool_choice": {"type": "function", "function": {"name": ""}},
                "tools": tools,
            },
            {
                "tool_choice": {"type": "function", "function": {"name": "missing"}},
                "tools": tools,
            },
            {
                "tool_choice": {"type": "function", "function": "shell"},
                "tools": tools,
            },
        ):
            with self.subTest(invalid=updates), self.assertRaises(soma.SomaError) as caught:
                soma.validate_request(self.request(**updates))
            self.assertEqual(caught.exception.status, 400)
            self.assertEqual(caught.exception.code, "invalid_request")

    def test_strict_json_rejects_duplicates_and_nonfinite_numbers(self):
        for payload in (
            '{"decision":"pass","decision":"rewrite"}',
            '{"decision":"pass","score":NaN}',
            '{"reasoning":"one","reasoning":"two","content":null}',
            '{"reasoning":null,"content":"ok","score":Infinity}',
        ):
            with self.subTest(payload=payload), self.assertRaises(soma.SomaError):
                if "decision" in payload:
                    soma.parse_classification(payload)
                else:
                    soma.parse_rewrite(payload)

    def test_config_rejects_invalid_urls_nonfinite_values_and_unsafe_headers(self):
        base = {
            "TARGET_URL": "http://127.0.0.1:19200/v1",
            "TRANSFORM_URL": "http://127.0.0.1:19201/v1",
            "TRANSFORM_MODEL": "transform",
            "PROXY_PORT": "19202",
        }
        for name, value in (
            ("CONNECT_TIMEOUT", "NaN"),
            ("REQUEST_TIMEOUT", "Infinity"),
            ("TRANSFORM_TEMPERATURE", "-Infinity"),
            ("TRANSFORM_TOTAL_TIMEOUT", "NaN"),
        ):
            with self.subTest(name=name), patch.dict(
                os.environ, base | {name: value}, clear=True
            ), self.assertRaises(RuntimeError):
                soma.Config.from_env()

        for url in (
            "http://user@127.0.0.1:19200/v1",
            "http://127.0.0.1:19200/v1?secret=yes",
            "http://127.0.0.1:19200/v1#fragment",
            "http://127.0.0.1:99999/v1",
        ):
            with self.subTest(url=url), self.assertRaises(RuntimeError):
                soma.Config(
                    target=soma.Endpoint(url),
                    transform=soma.Endpoint("http://127.0.0.1:19201/v1"),
                    transform_model="transform",
                    port=19202,
                ).validate()

    def test_secondary_configuration_is_atomic_and_reasoning_modes_are_closed(self):
        base = {
            "TARGET_URL": "http://127.0.0.1:19300/v1",
            "TRANSFORM_URL": "http://127.0.0.1:19301/v1",
            "TRANSFORM_MODEL": "primary",
            "PROXY_PORT": "19303",
        }
        secondary = {
            "TRANSFORM_SECONDARY_URL": "http://127.0.0.1:19302/v1",
            "TRANSFORM_SECONDARY_MODEL": "secondary",
            "TRANSFORM_SECONDARY_REASONING_MODE": "on",
        }
        for missing in secondary:
            values = secondary.copy()
            values.pop(missing)
            with self.subTest(missing=missing), patch.dict(
                os.environ, base | values, clear=True
            ), self.assertRaises(RuntimeError):
                soma.Config.from_env()
        with patch.dict(
            os.environ,
            base | {"TRANSFORM_CONFIRM_REWRITES": "true"},
            clear=True,
        ), self.assertRaises(RuntimeError):
            soma.Config.from_env()
        for primary_mode in ("off", "on", "default"):
            for secondary_mode in ("off", "on", "default"):
                with self.subTest(primary=primary_mode, secondary=secondary_mode), patch.dict(
                    os.environ,
                    base
                    | secondary
                    | {
                        "TRANSFORM_REASONING_MODE": primary_mode,
                        "TRANSFORM_SECONDARY_REASONING_MODE": secondary_mode,
                    },
                    clear=True,
                ):
                    if primary_mode == "off" and secondary_mode == "on":
                        soma.Config.from_env()
                    else:
                        with self.assertRaises(RuntimeError):
                            soma.Config.from_env()
        for name in (
            "TRANSFORM_REASONING_MODE",
            "TRANSFORM_SECONDARY_REASONING_MODE",
        ):
            with self.subTest(name=name), patch.dict(
                os.environ,
                base | secondary | {name: "auto"},
                clear=True,
            ), self.assertRaises(RuntimeError):
                soma.Config.from_env()

    def test_target_message_validation_rejects_malformed_envelopes(self):
        invalid_messages = (
            {"role": "user", "content": "answer"},
            {"role": "assistant", "content": []},
            {"role": "assistant", "content": " \n\t"},
            {
                "role": "assistant",
                "content": "answer",
                "reasoning_content": "one",
                "analysis": "two",
            },
            {"role": "assistant", "content": None, "tool_calls": {}},
        )
        for message in invalid_messages:
            body = fixtures.completion(content="answer")
            body["choices"][0]["message"] = copy.deepcopy(message)
            with self.subTest(message=message), self.assertRaises(soma.SomaError) as caught:
                soma.validate_completion(body, "target")
                soma.validate_target_message(body)
            self.assertEqual(caught.exception.code, "invalid_target_response")

        audio_only = fixtures.completion(content="")
        audio_only["choices"][0]["message"]["content"] = None
        audio_only["choices"][0]["message"]["audio"] = {
            "id": "audio-1",
            "data": "SYNTHETIC",
        }
        with self.assertRaises(soma.SomaError) as caught:
            soma.validate_target_message(audio_only)
        self.assertEqual(caught.exception.code, "unsupported_target_response")

        for headers in (
            {"X-Vendor": "one", "x-vendor": "two"},
            {"Host": "wrong.invalid"},
            {"Content-Length": "1"},
            {"Transfer-Encoding": "chunked"},
            {"Connection": "keep-alive"},
        ):
            with self.subTest(headers=headers), self.assertRaises(RuntimeError):
                soma.Config(
                    target=soma.Endpoint("http://127.0.0.1:19200/v1", headers=headers),
                    transform=soma.Endpoint("http://127.0.0.1:19201/v1"),
                    transform_model="transform",
                    port=19202,
                ).validate()

    def test_request_framing_rejects_ambiguous_or_unterminated_bodies(self):
        invalid = (
            SimpleNamespace(
                headers=TestHeaders({"Content-Length": "4", "Transfer-Encoding": "chunked"}),
                rfile=io.BytesIO(b"4\r\ntest\r\n0\r\n\r\n"),
            ),
            SimpleNamespace(
                headers=TestHeaders({"Transfer-Encoding": "gzip, chunked"}),
                rfile=io.BytesIO(b""),
            ),
            SimpleNamespace(
                headers=TestHeaders({"Transfer-Encoding": "chunked"}),
                rfile=io.BytesIO(b"-1\r\n"),
            ),
            SimpleNamespace(
                headers=TestHeaders({"Transfer-Encoding": "chunked"}),
                rfile=io.BytesIO(b"0\r\n"),
            ),
        )
        for handler in invalid:
            with self.subTest(headers=handler.headers), self.assertRaises(soma.SomaError) as caught:
                soma.read_body(handler)
            self.assertEqual(caught.exception.status, 400)
            self.assertEqual(caught.exception.code, "invalid_request")
        valid = SimpleNamespace(
            headers=TestHeaders({"Transfer-Encoding": "Chunked"}),
            rfile=io.BytesIO(b"4\r\ntest\r\n0\r\n\r\n"),
        )
        self.assertEqual(soma.read_body(valid), b"test")

    def test_upstream_headers_cannot_collide_or_inject(self):
        headers = soma.safe_headers(
            {
                "X-Soma-Version": "attacker",
                "x-soma-trace": "attacker",
                "X-Upstream-Metadata": "preserved",
                "Content-Type": "application/x-wrong",
            },
            "application/json",
        )
        self.assertFalse(any(name.lower().startswith("x-soma-") for name in headers))
        self.assertEqual(headers["X-Upstream-Metadata"], "preserved")
        self.assertEqual(headers["Content-Type"], "application/json")

        request_id = soma.safe_log_token(
            soma.response_request_id(
                {"X-Request-Id": "  request\r\nInjected: yes/" + "x" * 300}
            ),
            "",
        )
        self.assertNotRegex(request_id, r"[\r\n]")
        self.assertLessEqual(len(request_id), 128)

    def test_internal_errors_are_redacted_over_http(self):
        secret = "INTERNAL SECRET SENTINEL"

        class BrokenApp:
            config = SimpleNamespace(
                sse_chunk_chars=128,
                target=SimpleNamespace(models_url="http://unused"),
            )

            def complete(self, _payload, _headers):
                raise RuntimeError(secret)

        server = soma.Server(("127.0.0.1", 0), BrokenApp())
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            with self.assertLogs("soma", level="ERROR"):
                response = requests.post(
                    f"http://127.0.0.1:{server.server_port}/v1/chat/completions",
                    json=self.request(),
                    timeout=5,
                )
            self.assertEqual(response.status_code, 500)
            self.assertEqual(response.json()["error"]["code"], "internal_error")
            self.assertNotIn(secret, response.text)
        finally:
            server.shutdown()
            server.server_close()
            thread.join()

    def test_dispatched_soma_errors_expose_privacy_safe_diagnostics(self):
        stats = soma.Stats("trace-safe")
        stats.target_calls = 1
        stats.transform_calls = 3
        error = soma.SomaError(
            "repair could not be verified",
            code="rewrite_verification_failed",
        )
        error.stats = stats

        class FailedApp:
            config = SimpleNamespace(
                sse_chunk_chars=128,
                target=SimpleNamespace(models_url="http://unused"),
            )

            def complete(self, _payload, _headers):
                raise error

        server = soma.Server(("127.0.0.1", 0), FailedApp())
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            with self.assertLogs("soma", level="ERROR"):
                response = requests.post(
                    f"http://127.0.0.1:{server.server_port}/v1/chat/completions",
                    json=self.request(),
                    timeout=5,
                )
            self.assertEqual(response.status_code, 502)
            self.assertEqual(response.headers["X-Soma-Trace"], "trace-safe")
            self.assertEqual(response.headers["X-Soma-Target-Calls"], "1")
            self.assertEqual(response.headers["X-Soma-Transform-Calls"], "3")
        finally:
            server.shutdown()
            server.server_close()
            thread.join()


if __name__ == "__main__":
    unittest.main(verbosity=2)