diff --git a/docs/user-guide/cli-reference.md b/docs/user-guide/cli-reference.md
index a4406b015bb..0b67de37ae4 100644
--- a/docs/user-guide/cli-reference.md
+++ b/docs/user-guide/cli-reference.md
@@ -316,8 +316,8 @@ contract, session behavior, and model-family selection.
 | `--tito-model` | enum | `default` | TITO model family. Named families load their registered fixed template; `default` is best-effort with a checkpoint-native or custom template. |
 | `--max-seq-len` | int | – | Total tokens per session, including prompts, completions, and environment responses. Registered with the agentic wrapper. |
 | `--session-server-ip` | str | router IP | Session-server bind address. |
-| `--session-server-port` | int | auto | Starting port for standalone session-server instances. |
-| `--session-server-workers` | int | `32` | Number of instances; Miles uses consecutive ports starting at `--session-server-port`. |
+| `--session-server-port` | int | auto | First port for standalone session-server instances. When unset, each worker port is auto-allocated. |
+| `--session-server-workers` | int | `32` | Number of instances; an explicit `--session-server-port` anchors a consecutive range. |
 | `--session-sample-picker-path` | `<module>.<fn>` | `drop_retries` | v2 only: selects leaf samples before post-processing. |
 | `--session-sample-postprocessor-path` | `<module>.<fn>` | `default_postprocess` | v2 only: finalizes loss masks and rewards. |
 
diff --git a/miles/ray/rollout/router_manager.py b/miles/ray/rollout/router_manager.py
index 9bdde5094d9..c36f519a901 100644
--- a/miles/ray/rollout/router_manager.py
+++ b/miles/ray/rollout/router_manager.py
@@ -83,12 +83,19 @@ def start_router(args, *, has_pd_disaggregation: bool = False, force_new: bool =
 
 
 def _resolve_session_server_ports(start: int | None, workers: int) -> list[int]:
-    """Return the requested number of consecutive ports from the configured or auto-selected start."""
+    """Return consecutive ports from an explicit start, or auto-allocate each worker port."""
     if workers < 1:
         raise ValueError("--session-server-workers must be at least 1.")
     # TODO(#1837): Refactor IP/port allocation; keep this naive for now.
     if start is None:
-        start = find_available_port(random.randint(5000, 6000))
+        search_start = random.randint(5000, 6000)
+        ports = []
+        while len(ports) < workers:
+            port = find_available_port(search_start)
+            if port not in ports:
+                ports.append(port)
+            search_start = port + 1
+        return ports
     return list(range(start, start + workers))
 
 
diff --git a/miles/utils/arguments.py b/miles/utils/arguments.py
index b1a170f3744..782854b461d 100644
--- a/miles/utils/arguments.py
+++ b/miles/utils/arguments.py
@@ -2554,13 +2554,13 @@ def add_session_arguments(parser):
                 "--session-server-port",
                 type=int,
                 default=None,
-                help="Starting port for standalone session servers. Auto-allocated if not set.",
+                help="First port for standalone session servers. When unset, each worker port is auto-allocated.",
             )
             parser.add_argument(
                 "--session-server-workers",
                 type=int,
                 default=32,
-                help="Number of standalone session servers to launch on consecutive ports.",
+                help="Number of standalone session servers to launch. An explicit start uses consecutive ports.",
             )
             parser.add_argument(
                 "--tito-model",
diff --git a/miles/utils/test_utils/anthropic_session_verify_agent.py b/miles/utils/test_utils/anthropic_session_verify_agent.py
new file mode 100644
index 00000000000..0865e131046
--- /dev/null
+++ b/miles/utils/test_utils/anthropic_session_verify_agent.py
@@ -0,0 +1,386 @@
+"""Six-turn Anthropic Messages agent for per-model session verification."""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Callable
+
+import httpx
+
+from miles.rollout.base_types import GenerateFnInput, GenerateFnOutput
+from miles.rollout.generate_hub.agentic_tool_call import generate as _base_generate
+from miles.utils.chat_template_utils.message_matcher_hub import loose_tool_call_message_matches
+from miles.utils.chat_template_utils.tito_tokenizer import TITOTokenizerType
+from miles.utils.test_utils.session_verify_agent import (
+    INITIAL_SYSTEM_PROMPT,
+    INITIAL_USER_PROMPT,
+    MOCK_TOOL_RESULTS,
+    SYSTEM_REMINDER_TEXT,
+    TOOLS,
+    USER_FOLLOWUP_TEXT,
+    _journal_verifier_assertions,
+    _verify_tito_samples,
+    fixed_template_append_roles,
+)
+from miles.utils.test_utils.session_verify_agent import generate as _session_verify_generate
+
+logger = logging.getLogger(__name__)
+
+_ANTHROPIC_TOOLS = [
+    {
+        "name": tool["function"]["name"],
+        "description": tool["function"]["description"],
+        "input_schema": tool["function"]["parameters"],
+    }
+    for tool in TOOLS
+]
+_TOOL_PROMPTS = (
+    INITIAL_USER_PROMPT,
+    USER_FOLLOWUP_TEXT,
+    "Finally, check the weather in London.",
+)
+_REQUEST_COUNT = len(_TOOL_PROMPTS) * 2
+_MAX_TOKENS_PER_TURN = 1024
+_MAX_ANTHROPIC_INCOMPLETE_TURN_RETRIES = 8
+_MINIMAX_TITO_MODELS = frozenset(
+    {
+        TITOTokenizerType.MINIMAX_M25.value,
+        TITOTokenizerType.MINIMAX_M27.value,
+    }
+)
+_TOOL_RECOVERY_TEXT = "The weather service is temporarily unavailable."
+_TOOL_RESULT_EVENTS = (
+    "anthropic_tool_result_string",
+    "anthropic_tool_result_list",
+    "anthropic_tool_result_string",
+)
+_INTERMEDIATE_SYSTEM_EXPECTATIONS = frozenset({"required", "forbidden"})
+
+
+def _assert_supported_tito_model(tito_model: str) -> None:
+    assert (
+        tito_model not in _MINIMAX_TITO_MODELS
+    ), f"Anthropic session verification does not support tito_model={tito_model!r}"
+
+
+def _verify_intermediate_system(tito_model: str, *, route_supports: bool) -> bool:
+    """Require both the live route decision and the TITO append capability."""
+    return route_supports and "system" in fixed_template_append_roles(tito_model)
+
+
+def _assert_intermediate_system_expectation(expectation: str, *, actual: bool) -> None:
+    assert (
+        expectation in _INTERMEDIATE_SYSTEM_EXPECTATIONS
+    ), f"invalid Anthropic intermediate-system expectation: {expectation!r}"
+    expected = expectation == "required"
+    assert actual is expected, (
+        "Anthropic intermediate-system capability did not match the per-model E2E contract: "
+        f"expected={expectation}, actual={actual}"
+    )
+
+
+def _build_payload(request_kwargs: dict, metadata: dict, messages: list[dict], *, tool_choice: dict) -> dict:
+    payload = {
+        "model": metadata["anthropic_model"],
+        "max_tokens": min(request_kwargs["max_tokens"], _MAX_TOKENS_PER_TURN),
+        "system": INITIAL_SYSTEM_PROMPT,
+        "messages": list(messages),
+        "tools": _ANTHROPIC_TOOLS,
+        "tool_choice": tool_choice,
+        "stream": False,
+    }
+    for key in ("temperature", "top_p", "top_k"):
+        if request_kwargs.get(key) is not None:
+            payload[key] = request_kwargs[key]
+    if request_kwargs.get("stop") is not None:
+        stop = request_kwargs["stop"]
+        payload["stop_sequences"] = [stop] if isinstance(stop, str) else stop
+    return payload
+
+
+def _build_tool_result(tool_use: dict, turn_index: int) -> dict:
+    result_text = MOCK_TOOL_RESULTS[turn_index % len(MOCK_TOOL_RESULTS)]
+    content = [{"type": "text", "text": result_text}] if turn_index == 1 else result_text
+    if turn_index == 2:
+        content = _TOOL_RECOVERY_TEXT
+    return {
+        "type": "tool_result",
+        "tool_use_id": tool_use["id"],
+        "content": content,
+    }
+
+
+async def _post_complete(
+    client,
+    url: str,
+    payload: dict,
+    *,
+    label: str,
+    assert_response: Callable[[dict], object],
+) -> dict:
+    for attempt in range(_MAX_ANTHROPIC_INCOMPLETE_TURN_RETRIES + 1):
+        response = await client.post(url, json=payload)
+        assert response.status_code == 200, f"{label} failed ({response.status_code}): {response.text}"
+        body = response.json()
+        try:
+            assert_response(body)
+        except AssertionError:
+            if attempt == _MAX_ANTHROPIC_INCOMPLETE_TURN_RETRIES:
+                raise
+        else:
+            return body
+    raise AssertionError(f"{label} did not return a complete response")
+
+
+def _expected_driver_events(*, include_system: bool) -> list[str]:
+    events = []
+    for turn_index, result_event in enumerate(_TOOL_RESULT_EVENTS):
+        if turn_index == 2 and include_system:
+            events.append("anthropic_system")
+        events.extend(("anthropic_tool_use", result_event, "anthropic_text"))
+    return events
+
+
+@_journal_verifier_assertions("anthropic_session_verify_agent.run_agent")
+async def run_agent(base_url, prompt, request_kwargs, metadata, **kwargs):
+    """Run three tool-use/result/text cycles and verify their canonical records."""
+    tito_model = metadata["tito_model"]
+    _assert_supported_tito_model(tito_model)
+    intermediate_system_expectation = metadata["anthropic_intermediate_system_expectation"]
+    messages = []
+    events = []
+    tool_uses_per_turn = []
+    tool_use_count = 0
+
+    async with httpx.AsyncClient(timeout=180) as client:
+        server_url = base_url.rsplit("/sessions/", 1)[0]
+        health_response = await client.get(f"{server_url}/health")
+        assert health_response.status_code == 200, health_response.text
+        route_supports = health_response.json().get("anthropic_intermediate_system_supported")
+        assert type(route_supports) is bool, "session health did not report Anthropic intermediate-system capability"
+        include_system = _verify_intermediate_system(tito_model, route_supports=route_supports)
+        _assert_intermediate_system_expectation(intermediate_system_expectation, actual=include_system)
+
+        for turn_index, user_prompt in enumerate(_TOOL_PROMPTS):
+            if turn_index == 2 and include_system:
+                messages.append({"role": "system", "content": SYSTEM_REMINDER_TEXT})
+                events.append("anthropic_system")
+            user_content = [{"type": "text", "text": user_prompt}] if turn_index == 2 else user_prompt
+            messages.append({"role": "user", "content": user_content})
+
+            payload = _build_payload(
+                request_kwargs,
+                metadata,
+                messages,
+                tool_choice={"type": "tool", "name": "get_weather"},
+            )
+            tool_body = await _post_complete(
+                client,
+                f"{base_url}/v1/messages",
+                payload,
+                label=f"Anthropic tool turn {turn_index + 1}",
+                assert_response=_assert_anthropic_tool_response,
+            )
+            tool_uses = _assert_anthropic_tool_response(tool_body)
+            tool_uses_per_turn.append(tool_uses)
+            [tool_use] = tool_uses
+            tool_use_count += 1
+            events.append("anthropic_tool_use")
+
+            messages.append({"role": "assistant", "content": tool_body["content"]})
+            tool_result = _build_tool_result(tool_use, turn_index)
+            messages.append({"role": "user", "content": [tool_result]})
+            events.append(_TOOL_RESULT_EVENTS[turn_index])
+
+            payload = _build_payload(request_kwargs, metadata, messages, tool_choice={"type": "none"})
+            text_body = await _post_complete(
+                client,
+                f"{base_url}/v1/messages",
+                payload,
+                label=f"Anthropic text turn {turn_index + 1}",
+                assert_response=_assert_anthropic_text_response,
+            )
+            _assert_anthropic_text_response(text_body)
+            messages.append({"role": "assistant", "content": text_body["content"]})
+            events.append("anthropic_text")
+
+        session_response = await client.get(base_url)
+        assert session_response.status_code == 200, session_response.text
+        _assert_canonical_records(
+            session_response.json(),
+            tool_uses_per_turn,
+            include_system=include_system,
+        )
+
+    return {
+        "endpoint": "anthropic",
+        "driver_events": events,
+        "request_count": _REQUEST_COUNT,
+        "tool_use_count": tool_use_count,
+        "tool_result_count": tool_use_count,
+        "text_turn_count": len(_TOOL_PROMPTS),
+        "tool_result_string_count": 2,
+        "tool_result_list_count": 1,
+        "intermediate_system_used": include_system,
+    }
+
+
+def _assert_anthropic_tool_response(body: dict) -> list[dict]:
+    assert body["type"] == "message"
+    assert body["role"] == "assistant"
+    assert body["stop_reason"] == "tool_use"
+    tool_uses = [block for block in body["content"] if block["type"] == "tool_use"]
+    assert len(tool_uses) == 1, f"Anthropic response must contain exactly one tool_use block: {body!r}"
+    assert tool_uses[0]["name"] == "get_weather"
+    assert isinstance(tool_uses[0]["input"], dict)
+    return tool_uses
+
+
+def _assert_anthropic_text_response(body: dict) -> None:
+    assert body["type"] == "message"
+    assert body["role"] == "assistant"
+    assert body["stop_reason"] == "end_turn"
+    assert not any(block["type"] == "tool_use" for block in body["content"])
+    assert any(block["type"] == "text" and block["text"] for block in body["content"])
+
+
+def _assert_canonical_tool_calls(record: dict, tool_uses: list[dict]) -> None:
+    tool_calls = record["response"]["choices"][0]["message"]["tool_calls"]
+    calls_by_id = {tool_call["id"]: tool_call for tool_call in tool_calls}
+    assert set(calls_by_id) == {tool_use["id"] for tool_use in tool_uses}
+    for tool_use in tool_uses:
+        tool_call = calls_by_id[tool_use["id"]]
+        assert tool_call["function"]["name"] == tool_use["name"]
+        assert json.loads(tool_call["function"]["arguments"]) == tool_use["input"]
+
+
+def _assert_canonical_records(snapshot: dict, tool_uses_per_turn: list[list[dict]], *, include_system: bool) -> None:
+    records = snapshot["records"]
+    assert len(records) == _REQUEST_COUNT
+    for record in records:
+        assert record["path"] == "/v1/chat/completions"
+        assert record["request"]["input_ids"]
+    max_trim_tokens = snapshot["metadata"]["max_trim_tokens"]
+    for previous, current in zip(records, records[1:], strict=False):
+        previous_choice = previous["response"]["choices"][0]
+        completion_ids = [item[1] for item in previous_choice["meta_info"]["output_token_logprobs"]]
+        previous_ids = previous["request"]["input_ids"] + completion_ids
+        current_ids = current["request"]["input_ids"]
+        check_len = max(0, len(previous_ids) - max_trim_tokens)
+        assert current_ids[:check_len] == previous_ids[:check_len]
+    for record_index, current in enumerate(records[1:], start=1):
+        replayed_assistants = [message for message in current["request"]["messages"] if message["role"] == "assistant"]
+        stored_assistants = [record["response"]["choices"][0]["message"] for record in records[:record_index]]
+        assert len(replayed_assistants) == len(stored_assistants)
+        assert all(
+            loose_tool_call_message_matches(stored, replayed)
+            for stored, replayed in zip(stored_assistants, replayed_assistants, strict=True)
+        )
+
+    history_roles = []
+    for turn_index, tool_uses in enumerate(tool_uses_per_turn):
+        if turn_index == 2 and include_system:
+            history_roles.append("system")
+        history_roles.append("user")
+        tool_record = records[turn_index * 2]
+        assert [message["role"] for message in tool_record["request"]["messages"]] == [
+            "system",
+            *history_roles,
+        ]
+        _assert_canonical_tool_calls(tool_record, tool_uses)
+
+        history_roles.extend(["assistant", *(["tool"] * len(tool_uses))])
+        text_record = records[turn_index * 2 + 1]
+        assert [message["role"] for message in text_record["request"]["messages"]] == [
+            "system",
+            *history_roles,
+        ]
+        text_message = text_record["response"]["choices"][0]["message"]
+        assert text_message["content"]
+        assert not text_message.get("tool_calls")
+        history_roles.append("assistant")
+
+    tree = snapshot["metadata"]["tree"]
+    nodes_by_id = {node["id"]: node for node in tree["nodes"]}
+    nodes_by_response_id = {node["response_id"]: node for node in tree["nodes"]}
+    assert len(nodes_by_id) == len(tree["nodes"])
+    assert len(nodes_by_response_id) == len(tree["nodes"])
+    active_path_node_ids = [nodes_by_response_id[record["response"]["id"]]["id"] for record in records]
+    assert [nodes_by_id[node_id]["parent"] for node_id in active_path_node_ids] == [
+        None,
+        *active_path_node_ids[:-1],
+    ]
+    assert {
+        "node_id": active_path_node_ids[-1],
+        "path_node_ids": active_path_node_ids,
+    } in tree["leaves"]
+    last_choice = records[-1]["response"]["choices"][0]
+    last_completion_ids = [item[1] for item in last_choice["meta_info"]["output_token_logprobs"]]
+    assert snapshot["metadata"]["accumulated_token_ids"] == records[-1]["request"]["input_ids"] + last_completion_ids
+
+
+@_journal_verifier_assertions("anthropic_session_verify_agent.generate")
+async def generate(input: GenerateFnInput) -> GenerateFnOutput:
+    """Run the Anthropic agent, check hard TITO mismatches, and write metrics."""
+    tito_model = input.args.tito_model
+    _assert_supported_tito_model(tito_model)
+    intermediate_system_expectation = input.args.anthropic_intermediate_system_expectation
+    input.sample.metadata["anthropic_model"] = input.args.hf_checkpoint
+    input.sample.metadata["tito_model"] = tito_model
+    input.sample.metadata["anthropic_intermediate_system_expectation"] = intermediate_system_expectation
+    output = await _base_generate(input)
+
+    samples = output.samples if isinstance(output.samples, list) else [output.samples]
+    events_per_sample = [sample.metadata.get("driver_events", []) for sample in samples]
+    allowed_roles = list(fixed_template_append_roles(tito_model))
+    _verify_tito_samples(samples, events_per_sample, allowed_roles=allowed_roles)
+    if len(samples) != 1:
+        raise AssertionError(f"Anthropic per-model e2e: expected one linear sample, got {len(samples)}")
+    include_system = samples[0].metadata.get("intermediate_system_used")
+    if type(include_system) is not bool:
+        raise AssertionError("Anthropic per-model e2e: missing intermediate-system capability result")
+    _assert_intermediate_system_expectation(intermediate_system_expectation, actual=include_system)
+    if include_system and "system" not in fixed_template_append_roles(tito_model):
+        raise AssertionError(f"Anthropic per-model e2e: {tito_model!r} used an unsupported intermediate system")
+    expected_events = _expected_driver_events(include_system=include_system)
+    expected_counters = {
+        "request_count": _REQUEST_COUNT,
+        "tool_use_count": len(_TOOL_PROMPTS),
+        "tool_result_count": len(_TOOL_PROMPTS),
+        "text_turn_count": len(_TOOL_PROMPTS),
+        "tool_result_string_count": 2,
+        "tool_result_list_count": 1,
+    }
+    for i, sample in enumerate(samples):
+        if sample.metadata.get("endpoint") != "anthropic":
+            raise AssertionError(f"Anthropic per-model e2e: sample {i} did not retain agent metadata")
+        if events_per_sample[i] != expected_events:
+            raise AssertionError(
+                f"Anthropic per-model e2e: sample {i} events={events_per_sample[i]!r}, "
+                f"expected {expected_events!r}"
+            )
+        for key, expected in expected_counters.items():
+            if sample.metadata.get(key) != expected:
+                raise AssertionError(
+                    f"Anthropic per-model e2e: sample {i} {key}={sample.metadata.get(key)!r}, expected {expected}"
+                )
+        path_node_ids = sample.metadata.get("leaf", {}).get("path_node_ids")
+        if not isinstance(path_node_ids, list) or len(path_node_ids) != _REQUEST_COUNT:
+            raise AssertionError(f"Anthropic per-model e2e: sample {i} did not retain the linear six-turn leaf")
+
+    logger.info("Anthropic endpoint verified: samples=%d, requests_per_sample=%d", len(samples), _REQUEST_COUNT)
+    return output
+
+
+def _add_arguments(parser):
+    _session_verify_generate.add_arguments(parser)
+    parser.add_argument(
+        "--anthropic-intermediate-system-expectation",
+        required=True,
+        choices=sorted(_INTERMEDIATE_SYSTEM_EXPECTATIONS),
+        help=("Require or forbid the intermediate system turn in the Anthropic " "session-verification trajectory."),
+    )
+
+
+generate.add_arguments = _add_arguments
diff --git a/miles/utils/test_utils/session_verify_agent.py b/miles/utils/test_utils/session_verify_agent.py
index 4f422cdb417..4cdee8eb419 100644
--- a/miles/utils/test_utils/session_verify_agent.py
+++ b/miles/utils/test_utils/session_verify_agent.py
@@ -12,6 +12,7 @@
 import logging
 import os
 from enum import Enum
+from functools import wraps
 
 try:
     from enum import StrEnum
@@ -28,6 +29,48 @@
 logger = logging.getLogger(__name__)
 
 
+def _append_session_verify_record(entry: dict) -> bool:
+    metrics_path = os.environ.get("MILES_SESSION_VERIFY_METRICS_PATH")
+    if not metrics_path:
+        return False
+    payload = (json.dumps(entry) + "\n").encode()
+    fd = os.open(
+        metrics_path,
+        os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_CLOEXEC", 0),
+        0o600,
+    )
+    try:
+        written = os.write(fd, payload)
+        if written != len(payload):
+            raise OSError(f"short metrics sidecar write: expected {len(payload)} bytes, wrote {written}")
+    finally:
+        os.close(fd)
+    return True
+
+
+def _journal_verifier_assertions(stage: str):
+    def decorate(func):
+        @wraps(func)
+        async def wrapped(*args, **kwargs):
+            try:
+                return await func(*args, **kwargs)
+            except AssertionError as exc:
+                _append_session_verify_record(
+                    {
+                        "verification_error": {
+                            "stage": stage,
+                            "type": type(exc).__name__,
+                            "message": str(exc)[:1000],
+                        }
+                    }
+                )
+                raise
+
+        return wrapped
+
+    return decorate
+
+
 class DriverAction(Enum):
     TOOL_RESULT = "tool_result"
     USER_FOLLOWUP = "user_followup"
@@ -80,7 +123,7 @@ class ToolCallFailureMode(StrEnum):
 
 # Mismatch tiers reported by the session-server's per-sample comparator
 # (sessions.py:83).  Any occurrence of these "hard" types in a sample's
-# tito_session_mismatch indicates a TITO bug and fails the sample.  The
+# tito_session_mismatch indicates a TITO bug and fails the verifier run.  The
 # soft `assistant_text` tier is excluded — it is aggregated across samples
 # and gated by a ratio threshold instead.
 _FORBIDDEN_MISMATCH_TYPES: frozenset[str] = frozenset(
@@ -91,6 +134,8 @@ class ToolCallFailureMode(StrEnum):
 # (pytest via ``run_session_verify``).  Smaller-context models with a 4K
 # response budget should drop to 2 to avoid context overflow.
 DEFAULT_CYCLES = 3
+_MAX_INCOMPLETE_TURN_RETRIES = 2
+_RETRY_SEED_STRIDE = 1_000_000
 
 
 def fixed_template_append_roles(tito_model: TITOTokenizerType | str) -> tuple[str, ...]:
@@ -203,6 +248,21 @@ async def _chat(client, base_url, messages, request_kwargs, *, label):
     return resp.json()
 
 
+async def _chat_complete(client, base_url, messages, request_kwargs, *, label):
+    for retry in range(_MAX_INCOMPLETE_TURN_RETRIES + 1):
+        attempt_kwargs = request_kwargs
+        if retry and request_kwargs.get("seed") is not None:
+            attempt_kwargs = {
+                **request_kwargs,
+                "seed": request_kwargs["seed"] + retry * _RETRY_SEED_STRIDE,
+            }
+        response = await _chat(client, base_url, messages, attempt_kwargs, label=label)
+        if response["choices"][0].get("finish_reason") != "length":
+            return response
+    raise AssertionError(f"{label} exhausted {_MAX_INCOMPLETE_TURN_RETRIES} retries after finish_reason='length'")
+
+
+@_journal_verifier_assertions("session_verify_agent.run_agent")
 async def run_agent(base_url, prompt, request_kwargs, metadata, **kwargs):
     """Custom-agent entry point.  Returns ``{"driver_events": [...], **counters}``.
 
@@ -248,7 +308,7 @@ async def run_agent(base_url, prompt, request_kwargs, metadata, **kwargs):
 
     async with httpx.AsyncClient(timeout=180) as client:
         # Initial completion — no driver action yet.
-        resp = await _chat(client, base_url, messages, rk, label="Initial")
+        resp = await _chat_complete(client, base_url, messages, rk, label="Initial")
         assistant = resp["choices"][0]["message"]
         messages.append(assistant)
         events.append("initial")
@@ -356,7 +416,7 @@ async def run_agent(base_url, prompt, request_kwargs, metadata, **kwargs):
             else:
                 raise AssertionError(f"Unknown DriverAction {action!r}")
 
-            resp = await _chat(client, base_url, messages, rk, label=label)
+            resp = await _chat_complete(client, base_url, messages, rk, label=label)
             assistant = resp["choices"][0]["message"]
             messages.append(assistant)
             counters["tool_call_count"] += len(assistant.get("tool_calls") or [])
@@ -366,6 +426,65 @@ async def run_agent(base_url, prompt, request_kwargs, metadata, **kwargs):
     return {"driver_events": events, **counters}
 
 
+def _verify_tito_samples(samples, events_per_sample, *, allowed_roles) -> None:
+    """Record every sample and defer hard failures to the run-level gate.
+
+    The rollout loop treats custom-generate exceptions as retryable sample
+    failures.  With a metrics sidecar, hard TITO mismatches must therefore be
+    persisted instead of raised here and discarded with the sample.  Direct
+    callers without a sidecar retain the immediate-failure behavior.
+    """
+    metrics_path = os.environ.get("MILES_SESSION_VERIFY_METRICS_PATH")
+    for i, sample in enumerate(samples):
+        mismatches = sample.metadata.get("tito_session_mismatch")
+        if mismatches is None:
+            raise AssertionError(
+                f"Session multi-role e2e: sample {i} has no tito_session_mismatch "
+                f"in metadata.  The session-server's compute_session_mismatch raised "
+                f"TokenizationError (sessions.py:83 swallows it) — this always "
+                f"indicates a TITO subclass / setup bug, not a real PASS."
+            )
+        forbidden = [m for m in mismatches if m.get("type") in _FORBIDDEN_MISMATCH_TYPES]
+        assistant_mismatches = [m for m in mismatches if m.get("type") == "assistant_text"]
+        if metrics_path:
+            had_assistant_mismatch = bool(assistant_mismatches)
+            assistant_example = None
+            if assistant_mismatches:
+                first = assistant_mismatches[0]
+                assistant_example = {
+                    "segment_index": first.get("segment_index"),
+                    "expected_text": (first.get("expected_text") or "")[:300],
+                    "actual_text": (first.get("actual_text") or "")[:300],
+                }
+            hard_example = None
+            if forbidden:
+                first = forbidden[0]
+                hard_example = {
+                    "type": first.get("type"),
+                    "segment_index": first.get("segment_index"),
+                    "detail": str(first.get("detail") or "")[:500],
+                }
+            entry = {
+                "sample_index": i,
+                "driver_events": events_per_sample[i],
+                "had_assistant_mismatch": had_assistant_mismatch,
+                "total_mismatches": len(mismatches),
+                "assistant_mismatch_count": len(assistant_mismatches),
+                "assistant_mismatch_example": assistant_example,
+                "hard_mismatch_count": len(forbidden),
+                "hard_mismatch_types": sorted({m.get("type") for m in forbidden}),
+                "hard_mismatch_example": hard_example,
+            }
+            _append_session_verify_record(entry)
+        elif forbidden:
+            raise AssertionError(
+                f"Session multi-role e2e: sample {i} has forbidden mismatches "
+                f"{forbidden}. allowed_roles={allowed_roles}.  These types must be 0 "
+                f"for any TITO-correct setup."
+            )
+
+
+@_journal_verifier_assertions("session_verify_agent.generate")
 async def generate(input: GenerateFnInput) -> GenerateFnOutput:
     """Custom-generate wrapper that asserts driver-action coverage.
 
@@ -389,6 +508,10 @@ async def generate(input: GenerateFnInput) -> GenerateFnOutput:
     events_per_sample = [s.metadata.get("driver_events", []) for s in samples]
     metrics_path = os.environ.get("MILES_SESSION_VERIFY_METRICS_PATH")
 
+    if not samples:
+        raise AssertionError("Session multi-role e2e: generate returned no samples")
+    _verify_tito_samples(samples, events_per_sample, allowed_roles=allowed_roles)
+
     required_per_sample = ["rollback"]
     if "user" in allowed_roles:
         required_per_sample.append("append_user")
@@ -411,48 +534,6 @@ async def generate(input: GenerateFnInput) -> GenerateFnOutput:
             f"the model may not be tool-calling.  events_per_sample={events_per_sample}"
         )
 
-    for i, sample in enumerate(samples):
-        mismatches = sample.metadata.get("tito_session_mismatch")
-        if mismatches is None:
-            raise AssertionError(
-                f"Session multi-role e2e: sample {i} has no tito_session_mismatch "
-                f"in metadata.  The session-server's compute_session_mismatch raised "
-                f"TokenizationError (sessions.py:83 swallows it) — this always "
-                f"indicates a TITO subclass / setup bug, not a real PASS."
-            )
-        forbidden = [m for m in mismatches if m.get("type") in _FORBIDDEN_MISMATCH_TYPES]
-        if forbidden:
-            raise AssertionError(
-                f"Session multi-role e2e: sample {i} has forbidden mismatches "
-                f"{forbidden}. allowed_roles={allowed_roles}.  These types must be 0 "
-                f"for any TITO-correct setup."
-            )
-        if metrics_path:
-            assistant_mismatches = [m for m in mismatches if m.get("type") == "assistant_text"]
-            had_assistant_mismatch = bool(assistant_mismatches)
-            example = None
-            if assistant_mismatches:
-                first = assistant_mismatches[0]
-                example = {
-                    "segment_index": first.get("segment_index"),
-                    "expected_text": (first.get("expected_text") or "")[:300],
-                    "actual_text": (first.get("actual_text") or "")[:300],
-                }
-            with open(metrics_path, "a") as f:
-                f.write(
-                    json.dumps(
-                        {
-                            "sample_index": i,
-                            "driver_events": events_per_sample[i],
-                            "had_assistant_mismatch": had_assistant_mismatch,
-                            "total_mismatches": len(mismatches),
-                            "assistant_mismatch_count": len(assistant_mismatches),
-                            "assistant_mismatch_example": example,
-                        }
-                    )
-                    + "\n"
-                )
-
     logger.info(
         "Multi-role coverage verified: per_sample=%s, samples=%d, events=%s",
         required_per_sample,
diff --git a/miles/utils/test_utils/session_verify_runner.py b/miles/utils/test_utils/session_verify_runner.py
index 6411cdc996b..402f34fcac6 100644
--- a/miles/utils/test_utils/session_verify_runner.py
+++ b/miles/utils/test_utils/session_verify_runner.py
@@ -28,16 +28,19 @@
 import os
 import shutil
 import tempfile
-from typing import Any
+from typing import Any, Literal
 
 import miles.utils.external_utils.command_utils as U
 from miles.utils.chat_template_utils import resolve_reasoning_and_tool_call_parser
+from miles.utils.tracking_utils.ci_history import RECORD_DIR_ENV
 
 logger = logging.getLogger(__name__)
 
+SessionWireFormat = Literal["openai", "anthropic"]
+
 # Soft cap on how many samples may report any assistant_text mismatch.  Hard
 # mismatch types (special_token_count / special_token_type / non_assistant_text)
-# are asserted per-sample inside the agent wrapper — those must be 0.
+# are journaled per sample and rejected by the run-level gate — those must be 0.
 ASSISTANT_TEXT_MISMATCH_RATIO_THRESHOLD = 0.2
 
 PROMPT_DATA_PATH = "/root/datasets/session_multi_role_verify.jsonl"
@@ -66,6 +69,7 @@
     "custom_generate_function_path": "miles.utils.test_utils.session_verify_agent.generate",
     "custom_agent_function_path": "miles.utils.test_utils.session_verify_agent.run_agent",
     "use_session_server": "v2",
+    "session_message_matcher": "strict",
     "debug_rollout_only": True,
     "ci_test": True,
     "colocate": True,
@@ -153,6 +157,7 @@ def namespace_to_train_args(ns: argparse.Namespace) -> str:
         f"--session-verify-cycles {ns.session_verify_cycles}",
         f"--tool-call-failure-mode {ns.tool_call_failure_mode}",
         f"--tito-model {ns.tito_model}",
+        f"--session-message-matcher {ns.session_message_matcher}",
         f"--rollout-num-gpus-per-engine {ns.rollout_num_gpus_per_engine}",
         f"--sglang-reasoning-parser {ns.sglang_reasoning_parser}",
         f"--rm-type {ns.rm_type}",
@@ -180,6 +185,8 @@ def namespace_to_train_args(ns: argparse.Namespace) -> str:
                 "--sglang-speculative-num-draft-tokens 3",
             ]
         )
+    if getattr(ns, "anthropic_intermediate_system_expectation", None) is not None:
+        parts.append("--anthropic-intermediate-system-expectation " f"{ns.anthropic_intermediate_system_expectation}")
     if ns.use_session_server:
         # Preserve an explicit version string ("v2"); a bare True stays the bare flag.
         if isinstance(ns.use_session_server, str):
@@ -195,12 +202,25 @@ def namespace_to_train_args(ns: argparse.Namespace) -> str:
     return " ".join(parts) + " "
 
 
-def run_session_verify(args: argparse.Namespace) -> None:
-    """Boot ``miles`` rollout pipeline and run the multi-role driver.
+def _session_verify_env(
+    args: argparse.Namespace, metrics_path: str, *, wire_format: SessionWireFormat
+) -> dict[str, str]:
+    env = {
+        "MILES_TITO_MODEL": args.tito_model,
+        "MILES_SESSION_VERIFY_METRICS_PATH": metrics_path,
+    }
+    if wire_format == "anthropic":
+        # This run still uses the local metrics file, but must not overwrite
+        # another endpoint's CI-history series under the same v2 metric key.
+        env[RECORD_DIR_ENV] = ""
+    return env
+
+
+def run_session_verify(args: argparse.Namespace, *, wire_format: SessionWireFormat = "openai") -> None:
+    """Boot ``miles`` rollout pipeline and run the session-verification driver.
 
     Returns nothing on success; raises ``AssertionError`` on TITO mismatch
-    (HTTP 500 from server-side prefix check) or coverage shortfall (raised by
-    ``session_verify_agent.generate``).
+    (HTTP 500 from server-side prefix check) or coverage shortfall raised by the selected generate wrapper.
 
     ``args`` MUST be a fully-shaped Namespace carrying miles-canonical field
     names plus the session-verify-specific fields (``session_verify_cycles``,
@@ -218,6 +238,9 @@ def run_session_verify(args: argparse.Namespace) -> None:
     - ``args.hf_checkpoint`` is replaced with the local download path so the
       composed train_args points at the downloaded model, not the HF id.
     """
+    if wire_format not in ("openai", "anthropic"):
+        raise ValueError(f"unsupported session verification wire format: {wire_format}")
+
     args.sglang_reasoning_parser, args.sglang_tool_call_parser = resolve_reasoning_and_tool_call_parser(
         args.tito_model, args.sglang_reasoning_parser, args.sglang_tool_call_parser
     )
@@ -228,29 +251,32 @@ def run_session_verify(args: argparse.Namespace) -> None:
     train_args = namespace_to_train_args(args)
 
     # Per-sample token-seq metrics file: rollout workers append one JSONL line
-    # per sample inside session_verify_agent.generate; we aggregate after
+    # per sample inside the selected generate wrapper; we aggregate after
     # execute_train returns to apply the assistant_text soft threshold.
     metrics_fd, metrics_path = tempfile.mkstemp(prefix="session_verify_metrics_", suffix=".jsonl")
     os.close(metrics_fd)
 
-    preserved_metrics_path = None
     try:
         U.execute_train(
             train_args=train_args,
             num_gpus_per_node=args.actor_num_gpus_per_node,
             megatron_model_type=None,
-            extra_env_vars={
-                "MILES_TITO_MODEL": args.tito_model,
-                "MILES_SESSION_VERIFY_METRICS_PATH": metrics_path,
-            },
+            extra_env_vars=_session_verify_env(args, metrics_path, wire_format=wire_format),
+        )
+        assert_session_verify_metrics(
+            metrics_path,
+            assistant_text_threshold=args.assistant_text_threshold,
+            require_append_tool=wire_format == "openai",
         )
+    except Exception:
+        preserved_metrics_path = metrics_path + ".failed"
         try:
-            assert_session_verify_metrics(metrics_path, assistant_text_threshold=args.assistant_text_threshold)
-        except AssertionError:
-            preserved_metrics_path = metrics_path + ".failed"
             shutil.copy(metrics_path, preserved_metrics_path)
+        except Exception:
+            logger.exception("Failed to preserve per-sample mismatch payloads at %s", preserved_metrics_path)
+        else:
             logger.error("Preserved per-sample mismatch payloads at %s for post-mortem", preserved_metrics_path)
-            raise
+        raise
     finally:
         try:
             os.unlink(metrics_path)
@@ -258,30 +284,77 @@ def run_session_verify(args: argparse.Namespace) -> None:
             pass
 
 
-def assert_session_verify_metrics(metrics_path: str, *, assistant_text_threshold: float) -> None:
+def assert_session_verify_metrics(
+    metrics_path: str, *, assistant_text_threshold: float, require_append_tool: bool = True
+) -> None:
     """Read per-sample JSONL metrics and assert cross-sample verifier gates.
 
-    Forbidden mismatch types (special_*, non_assistant_text) are caught
-    per-sample in the agent wrapper and would have already raised by now.
-    Here we only cross-check the soft assistant_text rate against the
-    caller-provided threshold (per-model: some upstream sglang reasoning
-    parsers — notably ``nemotron_3`` — leave a trailing ``\\n`` in
-    ``reasoning_content`` that breaks the canonical roundtrip until the
-    parser is patched, so those families ride at threshold=1.0).
+    Forbidden mismatch types (special_*, non_assistant_text) are recorded by
+    the agent wrapper and hard-failed here so the rollout loop cannot discard
+    their assertion as a retryable sample failure.  The assistant_text tier
+    remains soft and is checked only against the caller-provided ratio
+    threshold.
     """
     samples_with_mismatch = 0
     total_samples = 0
     has_append_tool = False
+    samples_with_hard_mismatch = 0
+    hard_mismatch_count = 0
+    hard_mismatch_types = set()
+    hard_mismatch_example = None
+    verification_error_count = 0
+    verification_error_stages = set()
+    verification_error_example = None
     with open(metrics_path) as f:
         for line in f:
             line = line.strip()
             if not line:
                 continue
             entry = json.loads(line)
+            if "verification_error" in entry:
+                verification_error_count += 1
+                error = entry["verification_error"]
+                if isinstance(error, dict):
+                    stage = error.get("stage")
+                    if stage:
+                        verification_error_stages.add(stage)
+                    if verification_error_example is None:
+                        verification_error_example = {
+                            "stage": stage,
+                            "type": error.get("type"),
+                            "message": str(error.get("message") or "")[:1000],
+                        }
+                elif verification_error_example is None:
+                    verification_error_example = str(error)[:1000]
+                continue
             total_samples += 1
             has_append_tool = has_append_tool or "append_tool" in entry.get("driver_events", [])
             if entry.get("had_assistant_mismatch"):
                 samples_with_mismatch += 1
+            entry_hard_types = entry.get("hard_mismatch_types", [])
+            entry_hard_count = entry.get("hard_mismatch_count", len(entry_hard_types))
+            if entry_hard_count or entry_hard_types:
+                samples_with_hard_mismatch += 1
+                hard_mismatch_count += entry_hard_count
+                hard_mismatch_types.update(entry_hard_types)
+                if hard_mismatch_example is None:
+                    entry_example = entry.get("hard_mismatch_example")
+                    if isinstance(entry_example, dict):
+                        hard_mismatch_example = {
+                            "type": entry_example.get("type"),
+                            "segment_index": entry_example.get("segment_index"),
+                            "detail": str(entry_example.get("detail") or "")[:500],
+                        }
+                    elif entry_example is not None:
+                        hard_mismatch_example = str(entry_example)[:500]
+
+    if verification_error_count:
+        raise AssertionError(
+            "Session multi-role e2e: verifier assertions failed in "
+            f"{verification_error_count} attempted trajectories "
+            f"(completed_samples={total_samples}, stages={sorted(verification_error_stages)}, "
+            f"first={verification_error_example})."
+        )
 
     if total_samples == 0:
         raise AssertionError(
@@ -290,21 +363,32 @@ def assert_session_verify_metrics(metrics_path: str, *, assistant_text_threshold
             "run before any sample completed.  Check rollout logs."
         )
 
-    if not has_append_tool:
-        raise AssertionError(
-            "Session multi-role e2e: no sample produced an append_tool action — "
-            "the model may not be tool-calling.  Check sampling temperature, "
-            "the tool spec, or parser configuration."
-        )
-
     ratio = samples_with_mismatch / total_samples
     logger.info(
-        "Token-seq metric summary: samples=%d, with_assistant_text_mismatch=%d, ratio=%.3f, threshold=%.3f",
+        "Token-seq metric summary: samples=%d, with_hard_mismatch=%d, "
+        "with_assistant_text_mismatch=%d, ratio=%.3f, threshold=%.3f",
         total_samples,
+        samples_with_hard_mismatch,
         samples_with_mismatch,
         ratio,
         assistant_text_threshold,
     )
+    if samples_with_hard_mismatch:
+        raise AssertionError(
+            f"Session multi-role e2e: hard TITO mismatches found in "
+            f"{samples_with_hard_mismatch}/{total_samples} attempted samples "
+            f"({hard_mismatch_count} mismatches, types={sorted(hard_mismatch_types)}, "
+            f"first={hard_mismatch_example}).  These types must be 0 for any "
+            "TITO-correct setup."
+        )
+
+    if require_append_tool and not has_append_tool:
+        raise AssertionError(
+            "Session multi-role e2e: no sample produced an append_tool action — "
+            "the model may not be tool-calling.  Check sampling temperature, "
+            "the tool spec, or parser configuration."
+        )
+
     if ratio > assistant_text_threshold:
         raise AssertionError(
             f"Session multi-role e2e: assistant_text mismatch ratio "
diff --git a/tests/e2e/sglang/test_session_server_multi_role/_common.py b/tests/e2e/sglang/test_session_server_multi_role/_common.py
index a89f2424449..416a87afd9e 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/_common.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/_common.py
@@ -1,8 +1,9 @@
 """Shared types and runner for multi-role session-server TITO e2e tests.
 
 Each test file in this directory owns a single ``ModelConfig`` and drives it
-through ``run_both_versions(cfg)``. The runner applies the model-specific
-GPU topology centrally.
+through ``run_both_versions(cfg)``. Each model runs OpenAI on session v1 and
+v2, then Anthropic Messages on v2 unless that model explicitly disables the
+unsupported endpoint. The runner applies the model-specific GPU topology centrally.
 """
 
 import argparse
@@ -16,7 +17,16 @@
 )
 
 SessionServerVersion = Literal["v1", "v2"]
+SessionEndpoint = Literal["openai", "anthropic"]
+AnthropicIntermediateSystemExpectation = Literal["required", "forbidden"]
 _SESSION_SERVER_VERSIONS: tuple[SessionServerVersion, ...] = ("v1", "v2")
+_SESSION_RUNS: tuple[tuple[SessionServerVersion, SessionEndpoint], ...] = (
+    ("v1", "openai"),
+    ("v2", "openai"),
+    ("v2", "anthropic"),
+)
+_ANTHROPIC_GENERATE = "miles.utils.test_utils.anthropic_session_verify_agent.generate"
+_ANTHROPIC_AGENT = "miles.utils.test_utils.anthropic_session_verify_agent.run_agent"
 
 
 @dataclass(frozen=True)
@@ -42,6 +52,14 @@ class ModelConfig:
     # keeps trailing newline in reasoning_content) so the gate does not
     # block on a documented out-of-scope issue.
     assistant_text_threshold: float = ASSISTANT_TEXT_MISMATCH_RATIO_THRESHOLD
+    # Optional Anthropic-only override; None inherits the per-model threshold.
+    anthropic_assistant_text_threshold: float | None = None
+    # Endpoint capability gate; unsupported families still run both OpenAI versions.
+    verify_anthropic: bool = True
+    # Required per-model Anthropic E2E contract.  The agent still derives the
+    # live capability from the route and fixed template, then checks it against
+    # this expectation before issuing a request with an intermediate system.
+    anthropic_intermediate_system_expectation: AnthropicIntermediateSystemExpectation | None = None
     # Recovery mode when a TOOL_RESULT step finds the assistant emitted no
     # tool_calls.  Default "rollback" is universal (pop assistant + retry);
     # see ToolCallFailureMode for "append_tool" / "append_user" variants.
@@ -52,11 +70,21 @@ def run_one(
     cfg: ModelConfig,
     *,
     session_server_version: SessionServerVersion = "v2",
+    endpoint: SessionEndpoint = "openai",
     rollout_batch_size: int = SESSION_VERIFY_INVARIANT_ARGS["rollout_batch_size"],
 ) -> None:
+    if endpoint == "anthropic" and session_server_version != "v2":
+        raise ValueError("Anthropic per-model verification requires session server v2")
+    if endpoint == "anthropic" and cfg.anthropic_intermediate_system_expectation is None:
+        raise ValueError("Anthropic per-model verification requires an intermediate-system expectation")
+
     invariants = dict(SESSION_VERIFY_INVARIANT_ARGS)
     invariants["use_session_server"] = session_server_version
     invariants["rollout_batch_size"] = rollout_batch_size
+    if endpoint == "anthropic":
+        invariants["custom_generate_function_path"] = _ANTHROPIC_GENERATE
+        invariants["custom_agent_function_path"] = _ANTHROPIC_AGENT
+        invariants["session_message_matcher"] = "loose_tool_call"
     # This harness produces one rollout batch, so its train-side batch divisor
     # must track the actual sample count when large-model lanes reduce samples.
     invariants["global_batch_size"] = invariants["rollout_batch_size"] * cfg.n_samples_per_prompt
@@ -65,6 +93,9 @@ def run_one(
     invariants["sglang_ep_size"] = cfg.ep_size
     invariants["sglang_context_length"] = cfg.context_length
     invariants["enable_spec"] = cfg.enable_spec
+    assistant_text_threshold = cfg.assistant_text_threshold
+    if endpoint == "anthropic" and cfg.anthropic_assistant_text_threshold is not None:
+        assistant_text_threshold = cfg.anthropic_assistant_text_threshold
     args = argparse.Namespace(
         hf_checkpoint=cfg.model_name,
         tito_model=cfg.tito_model,
@@ -76,13 +107,18 @@ def run_one(
         n_samples_per_prompt=cfg.n_samples_per_prompt,
         session_verify_cycles=cfg.cycles,
         tool_call_failure_mode=cfg.tool_call_failure_mode,
-        assistant_text_threshold=cfg.assistant_text_threshold,
+        assistant_text_threshold=assistant_text_threshold,
+        anthropic_intermediate_system_expectation=(
+            cfg.anthropic_intermediate_system_expectation if endpoint == "anthropic" else None
+        ),
         **invariants,
     )
-    run_session_verify(args=args)
+    run_session_verify(args=args, wire_format=endpoint)
 
 
 def run_both_versions(cfg: ModelConfig) -> None:
     rollout_batch_size = SESSION_VERIFY_INVARIANT_ARGS["rollout_batch_size"] // len(_SESSION_SERVER_VERSIONS)
-    for version in _SESSION_SERVER_VERSIONS:
-        run_one(cfg, session_server_version=version, rollout_batch_size=rollout_batch_size)
+    for version, endpoint in _SESSION_RUNS:
+        if endpoint == "anthropic" and not cfg.verify_anthropic:
+            continue
+        run_one(cfg, session_server_version=version, endpoint=endpoint, rollout_batch_size=rollout_batch_size)
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_deepseekv4.py b/tests/e2e/sglang/test_session_server_multi_role/test_deepseekv4.py
index d52eff93988..554be283691 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_deepseekv4.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_deepseekv4.py
@@ -2,7 +2,7 @@
 from tests.ci.metric_history import register_ci_gate
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig, run_both_versions
 
-register_cuda_ci(est_time=1400, suite="stage-c-4-gpu-h200", labels=["sglang"])
+register_cuda_ci(est_time=2100, suite="stage-c-4-gpu-h200", labels=["sglang"])
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v1/assistant_text")
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v2/assistant_text")
 
@@ -25,6 +25,7 @@
     # order, so a sentinel tool_call_id would not roundtrip; use the
     # universal rollback recovery when the model emits no tool_calls.
     tool_call_failure_mode="rollback",
+    anthropic_intermediate_system_expectation="forbidden",
 )
 
 
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_glm47.py b/tests/e2e/sglang/test_session_server_multi_role/test_glm47.py
index a1dd2ac3f38..cc2a2a31956 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_glm47.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_glm47.py
@@ -2,7 +2,7 @@
 from tests.ci.metric_history import register_ci_gate
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig, run_both_versions
 
-register_cuda_ci(est_time=600, suite="stage-c-2-gpu-h200", labels=["sglang"])
+register_cuda_ci(est_time=900, suite="stage-c-2-gpu-h200", labels=["sglang"])
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v1/assistant_text")
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v2/assistant_text")
 
@@ -19,6 +19,7 @@
     # preceding assistant carries a matching tool_call.id, so the APPEND_TOOL
     # sentinel ("tool_call_id": "none") roundtrips cleanly.
     tool_call_failure_mode="append_tool",
+    anthropic_intermediate_system_expectation="required",
 )
 
 
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_inkling.py b/tests/e2e/sglang/test_session_server_multi_role/test_inkling.py
index 13efcb5dd22..ad8c19f1302 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_inkling.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_inkling.py
@@ -2,7 +2,7 @@
 from tests.ci.metric_history import register_ci_gate
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig, run_both_versions
 
-register_cuda_ci(est_time=1200, suite="stage-c-4-gpu-h200", labels=["sglang"])
+register_cuda_ci(est_time=1800, suite="stage-c-4-gpu-h200", labels=["sglang"])
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v1/assistant_text")
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v2/assistant_text")
 
@@ -21,6 +21,7 @@
     cycles=2,
     n_samples_per_prompt=1,
     tool_call_failure_mode="append_tool",
+    anthropic_intermediate_system_expectation="required",
 )
 
 
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_minimax_m27.py b/tests/e2e/sglang/test_session_server_multi_role/test_minimax_m27.py
index 0fedbaec2f2..1fd7cc259fb 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_minimax_m27.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_minimax_m27.py
@@ -12,7 +12,7 @@
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig, run_both_versions
 
 register_cuda_ci(
-    est_time=800,
+    est_time=1200,
     suite="stage-c-4-gpu-h200",
     labels=["sglang"],
     disabled="MiniMax-M2.7 is deprecated.",
@@ -48,6 +48,8 @@
     cycles=2,
     assistant_text_threshold=0.1,
     tool_call_failure_mode="append_user",
+    # MiniMax does not reliably honor forced Anthropic tool_use responses.
+    verify_anthropic=False,
 )
 
 
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_nemotron3.py b/tests/e2e/sglang/test_session_server_multi_role/test_nemotron3.py
index c1f4e8ad184..e097e5c0768 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_nemotron3.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_nemotron3.py
@@ -2,7 +2,7 @@
 from tests.ci.metric_history import register_ci_gate
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig, run_both_versions
 
-register_cuda_ci(est_time=700, suite="stage-c-2-gpu-h200", labels=["sglang"])
+register_cuda_ci(est_time=1050, suite="stage-c-2-gpu-h200", labels=["sglang"])
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v1/assistant_text")
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v2/assistant_text")
 
@@ -20,6 +20,7 @@
     cycles=2,
     assistant_text_threshold=1.0,
     tool_call_failure_mode="append_tool",
+    anthropic_intermediate_system_expectation="required",
 )
 
 
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_qwen3.py b/tests/e2e/sglang/test_session_server_multi_role/test_qwen3.py
index 3659c4e8f27..fc58a13027c 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_qwen3.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_qwen3.py
@@ -2,8 +2,8 @@
 from tests.ci.metric_history import register_ci_gate
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig, run_both_versions
 
-register_cuda_ci(est_time=700, suite="stage-c-2-gpu-h200", labels=["sglang"])
-register_rocm_ci(est_time=500, suite="nightly-stage-c-2-gpu-mi350", labels=["sglang"])
+register_cuda_ci(est_time=1050, suite="stage-c-2-gpu-h200", labels=["sglang"])
+register_rocm_ci(est_time=750, suite="nightly-stage-c-2-gpu-mi350", labels=["sglang"])
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v1/assistant_text")
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v2/assistant_text")
 
@@ -17,9 +17,11 @@
     tp_size=1,
     cycles=2,
     tool_call_failure_mode="append_tool",
-    # qwen3 assistant_text TITO roundtrip drifts just over the 0.2 default
-    # (0.203 observed in CI); raise the per-family soft gate to 0.25.
+    # Anthropic tool-call conversion preserves structure but normalizes its raw
+    # serialization (32/32 formatting-only mismatches in CI); keep OpenAI at 0.25.
     assistant_text_threshold=0.25,
+    anthropic_assistant_text_threshold=1.0,
+    anthropic_intermediate_system_expectation="required",
 )
 
 
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_qwen35.py b/tests/e2e/sglang/test_session_server_multi_role/test_qwen35.py
index d6bc33eb877..1d5b6c7f132 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_qwen35.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_qwen35.py
@@ -2,8 +2,8 @@
 from tests.ci.metric_history import register_ci_gate
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig, run_both_versions
 
-register_cuda_ci(est_time=800, suite="stage-c-4-gpu-h200", labels=["sglang"])
-register_rocm_ci(est_time=500, suite="nightly-stage-c-4-gpu-mi350", labels=["sglang"])
+register_cuda_ci(est_time=1200, suite="stage-c-4-gpu-h200", labels=["sglang"])
+register_rocm_ci(est_time=750, suite="nightly-stage-c-4-gpu-mi350", labels=["sglang"])
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v1/assistant_text")
 register_ci_gate(metric_key="rollout/tito_session_mismatch_rate/v2/assistant_text")
 
@@ -17,6 +17,10 @@
     enable_spec=True,
     cycles=2,
     tool_call_failure_mode="append_tool",
+    # Anthropic tool-call conversion changes raw assistant serialization;
+    # keep this endpoint-only formatting mismatch soft while hard gates stay at 0.
+    anthropic_assistant_text_threshold=1.0,
+    anthropic_intermediate_system_expectation="forbidden",
 )
 
 
diff --git a/tests/e2e/sglang/test_session_server_multi_role/test_qwen36.py b/tests/e2e/sglang/test_session_server_multi_role/test_qwen36.py
index 1686b5d4cf4..45bb7c1711c 100644
--- a/tests/e2e/sglang/test_session_server_multi_role/test_qwen36.py
+++ b/tests/e2e/sglang/test_session_server_multi_role/test_qwen36.py
@@ -17,6 +17,10 @@
     enable_spec=True,
     cycles=2,
     tool_call_failure_mode="append_tool",
+    # Anthropic tool-call conversion changes raw assistant serialization;
+    # keep this endpoint-only formatting mismatch soft while hard gates stay at 0.
+    anthropic_assistant_text_threshold=1.0,
+    anthropic_intermediate_system_expectation="forbidden",
 )
 
 
diff --git a/tests/fast/ray/rollout/test_router_manager.py b/tests/fast/ray/rollout/test_router_manager.py
index b92c3033509..fe63c0c8c9c 100644
--- a/tests/fast/ray/rollout/test_router_manager.py
+++ b/tests/fast/ray/rollout/test_router_manager.py
@@ -72,6 +72,13 @@ def test_none_auto_allocates_one_port(self):
         with patch("miles.ray.rollout.router_manager.find_available_port", return_value=20002):
             assert _resolve_session_server_ports(None, 1) == [20002]
 
+    def test_none_auto_allocates_each_port_independently(self):
+        with patch("miles.ray.rollout.router_manager.find_available_port", side_effect=[6360, 6380]) as find_port:
+            ports = _resolve_session_server_ports(None, 2)
+
+        assert ports == [6360, 6380]
+        assert find_port.call_count == 2
+
     def test_one_worker_uses_the_starting_port(self):
         assert _resolve_session_server_ports(30000, 1) == [30000]
 
diff --git a/tests/fast/router/test_sglang_anthropic_conversion.py b/tests/fast/router/test_sglang_anthropic_conversion.py
new file mode 100644
index 00000000000..8c7193588ec
--- /dev/null
+++ b/tests/fast/router/test_sglang_anthropic_conversion.py
@@ -0,0 +1,38 @@
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+from tests.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=60, suite="stage-a-cpu", labels=[])
+
+
+_SGLANG_TESTS = (
+    Path("test/registered/unit/entrypoints/anthropic/test_utils.py"),
+    Path("test/registered/unit/entrypoints/anthropic/test_serving.py"),
+)
+
+
+def test_sglang_anthropic_conversion_contract():
+    sglang_source_root = Path(os.environ["SGLANG_SOURCE_ROOT"])
+    sglang_repo_root = sglang_source_root.parent
+    missing_tests = [str(path) for path in _SGLANG_TESTS if not (sglang_repo_root / path).is_file()]
+    assert not missing_tests, f"Missing SGLang Anthropic tests: {missing_tests}"
+
+    env = os.environ.copy()
+    env["PYTHONDONTWRITEBYTECODE"] = "1"
+    subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "pytest",
+            "-p",
+            "no:cacheprovider",
+            "-q",
+            *(str(path) for path in _SGLANG_TESTS),
+        ],
+        cwd=sglang_repo_root,
+        env=env,
+        check=True,
+    )
diff --git a/tests/fast/utils/test_utils/test_anthropic_session_verify_agent.py b/tests/fast/utils/test_utils/test_anthropic_session_verify_agent.py
new file mode 100644
index 00000000000..32214f2cb84
--- /dev/null
+++ b/tests/fast/utils/test_utils/test_anthropic_session_verify_agent.py
@@ -0,0 +1,585 @@
+import asyncio
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from miles.rollout.base_types import GenerateFnOutput
+from miles.utils.test_utils import anthropic_session_verify_agent
+from miles.utils.types import Sample
+
+
+class _Response:
+    def __init__(self, body: dict, status_code: int = 200):
+        self._body = body
+        self.status_code = status_code
+        self.text = json.dumps(body)
+
+    def json(self) -> dict:
+        return self._body
+
+
+def _response_fixtures():
+    tool_uses = [
+        {
+            "type": "tool_use",
+            "id": f"call_weather_{index}",
+            "name": "get_weather",
+            "input": {"location": location},
+        }
+        for index, location in enumerate(("Beijing", "Shanghai", "London"), start=1)
+    ]
+    tool_bodies = [
+        {
+            "type": "message",
+            "role": "assistant",
+            "content": [
+                {"type": "thinking", "thinking": f"Think about turn {index}."},
+                tool_use,
+            ],
+            "stop_reason": "tool_use",
+        }
+        for index, tool_use in enumerate(tool_uses, start=1)
+    ]
+    text_bodies = [
+        {
+            "type": "message",
+            "role": "assistant",
+            "content": [
+                {"type": "thinking", "thinking": f"Use result {index}."},
+                {"type": "text", "text": f"Weather answer {index}."},
+            ],
+            "stop_reason": "end_turn",
+        }
+        for index in range(1, 4)
+    ]
+    bodies = [item for pair in zip(tool_bodies, text_bodies, strict=True) for item in pair]
+
+    canonical_messages = []
+    for response_index, body in enumerate(bodies):
+        reasoning = next(block["thinking"] for block in body["content"] if block["type"] == "thinking")
+        if response_index % 2 == 0:
+            tool_use = tool_uses[response_index // 2]
+            message = {
+                "content": "",
+                "reasoning_content": reasoning,
+                "tool_calls": [
+                    {
+                        "id": tool_use["id"],
+                        "type": "function",
+                        "function": {
+                            "name": tool_use["name"],
+                            "arguments": json.dumps(tool_use["input"]),
+                        },
+                    }
+                ],
+            }
+        else:
+            text = next(block["text"] for block in body["content"] if block["type"] == "text")
+            message = {"content": text, "reasoning_content": reasoning}
+        canonical_messages.append({"role": "assistant", **message})
+    return tool_uses, bodies, canonical_messages
+
+
+def _snapshot(canonical_messages, *, include_system: bool):
+    records = []
+    history_roles = []
+    for record_index, response_message in enumerate(canonical_messages):
+        turn_index = record_index // 2
+        if record_index % 2 == 0:
+            if turn_index == 2 and include_system:
+                history_roles.append("system")
+            history_roles.append("user")
+        else:
+            history_roles.extend(["assistant", "tool"])
+
+        prior_assistants = iter(canonical_messages[:record_index])
+        request_messages = [
+            dict(next(prior_assistants)) if role == "assistant" else {"role": role}
+            for role in ["system", *history_roles]
+        ]
+        input_ids = list(range(1, record_index * 2 + 2))
+        completion_id = record_index * 2 + 2
+        records.append(
+            {
+                "path": "/v1/chat/completions",
+                "request": {"input_ids": input_ids, "messages": request_messages},
+                "response": {
+                    "id": f"response_{record_index}",
+                    "choices": [
+                        {
+                            "message": response_message,
+                            "meta_info": {"output_token_logprobs": [[-0.1, completion_id, None]]},
+                        }
+                    ],
+                },
+            }
+        )
+        if record_index % 2 == 1:
+            history_roles.append("assistant")
+
+    final_ids = records[-1]["request"]["input_ids"] + [len(records) * 2]
+    metadata = {
+        "max_trim_tokens": 0,
+        "accumulated_token_ids": final_ids,
+        "tree": {
+            "nodes": [
+                {
+                    "id": index,
+                    "parent": None if index == 0 else index - 1,
+                    "response_id": f"response_{index}",
+                }
+                for index in range(6)
+            ],
+            "leaves": [{"node_id": 5, "path_node_ids": [0, 1, 2, 3, 4, 5]}],
+        },
+    }
+    return {"records": records, "metadata": metadata}
+
+
+@pytest.mark.parametrize(
+    ("tito_model", "route_supports", "expectation", "include_system"),
+    [
+        ("qwen3", True, "required", True),
+        ("qwen35", True, "forbidden", False),
+        ("qwen35", False, "forbidden", False),
+    ],
+)
+def test_run_agent_runs_six_turns_and_checks_canonical_records(
+    monkeypatch, tito_model, route_supports, expectation, include_system
+):
+    posted = []
+    _, response_bodies, canonical_messages = _response_fixtures()
+    snapshot = _snapshot(canonical_messages, include_system=include_system)
+
+    class FakeAsyncClient:
+        def __init__(self, *, timeout):
+            assert timeout == 180
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return None
+
+        async def post(self, url, *, json):
+            body = response_bodies[len(posted)]
+            posted.append((url, json))
+            return _Response(body)
+
+        async def get(self, url):
+            if url == "http://session/health":
+                return _Response({"status": "ok", "anthropic_intermediate_system_supported": route_supports})
+            assert url == "http://session"
+            return _Response(snapshot)
+
+    monkeypatch.setattr(anthropic_session_verify_agent.httpx, "AsyncClient", FakeAsyncClient)
+
+    result = asyncio.run(
+        anthropic_session_verify_agent.run_agent(
+            "http://session",
+            prompt=None,
+            request_kwargs={"max_tokens": 2048, "temperature": 0.2, "stop": "<stop>"},
+            metadata={
+                "anthropic_model": "/models/test",
+                "tito_model": tito_model,
+                "anthropic_intermediate_system_expectation": expectation,
+            },
+        )
+    )
+
+    assert len(posted) == 6
+    assert {url for url, _ in posted} == {"http://session/v1/messages"}
+    assert [payload["tool_choice"] for _, payload in posted] == [
+        {"type": "tool", "name": "get_weather"},
+        {"type": "none"},
+    ] * 3
+    assert all(payload["model"] == "/models/test" for _, payload in posted)
+    assert all(payload["max_tokens"] == 1024 for _, payload in posted)
+    assert all(payload["stream"] is False for _, payload in posted)
+    assert all(payload["stop_sequences"] == ["<stop>"] for _, payload in posted)
+    assert posted[0][1]["tools"][0]["input_schema"]["required"] == ["location"]
+
+    first_result = posted[1][1]["messages"][-1]["content"][0]
+    assert first_result["type"] == "tool_result"
+    assert isinstance(first_result["content"], str)
+    second_result = posted[3][1]["messages"][-1]["content"][0]
+    assert second_result["content"][0]["type"] == "text"
+    third_result = posted[5][1]["messages"][-1]["content"][0]
+    assert third_result["content"] == anthropic_session_verify_agent._TOOL_RECOVERY_TEXT
+    assert "is_error" not in third_result
+    tail_roles = [message["role"] for message in posted[4][1]["messages"][-2:]]
+    assert ("system" in tail_roles) is include_system
+    assert posted[4][1]["messages"][-1]["content"][0]["type"] == "text"
+    assert posted[1][1]["messages"][1]["content"][0]["type"] == "thinking"
+
+    expected_events = anthropic_session_verify_agent._expected_driver_events(include_system=include_system)
+    assert result == {
+        "endpoint": "anthropic",
+        "driver_events": expected_events,
+        "request_count": 6,
+        "tool_use_count": 3,
+        "tool_result_count": 3,
+        "text_turn_count": 3,
+        "tool_result_string_count": 2,
+        "tool_result_list_count": 1,
+        "intermediate_system_used": include_system,
+    }
+
+
+def test_canonical_records_allow_superseded_retry_leaves():
+    tool_uses, _, canonical_messages = _response_fixtures()
+    snapshot = _snapshot(canonical_messages, include_system=True)
+    snapshot["records"][-1]["response"]["id"] = "response_7"
+    snapshot["metadata"]["tree"] = {
+        "nodes": [
+            {
+                "id": index,
+                "parent": None if index == 0 else index - 1,
+                "response_id": f"response_{index}",
+            }
+            for index in range(5)
+        ]
+        + [
+            {"id": 5, "parent": 4, "response_id": "retry_1"},
+            {"id": 6, "parent": 4, "response_id": "retry_2"},
+            {"id": 7, "parent": 4, "response_id": "response_7"},
+        ],
+        "leaves": [
+            {"node_id": 5, "path_node_ids": [0, 1, 2, 3, 4, 5]},
+            {"node_id": 6, "path_node_ids": [0, 1, 2, 3, 4, 6]},
+            {"node_id": 7, "path_node_ids": [0, 1, 2, 3, 4, 7]},
+        ],
+    }
+
+    anthropic_session_verify_agent._assert_canonical_records(
+        snapshot,
+        [[tool_use] for tool_use in tool_uses],
+        include_system=True,
+    )
+
+
+def test_post_complete_retries_max_tokens():
+    responses = iter(
+        [
+            _Response(
+                {
+                    "type": "message",
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": "truncated"}],
+                    "stop_reason": "max_tokens",
+                }
+            ),
+            _Response(
+                {
+                    "type": "message",
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": "complete"}],
+                    "stop_reason": "end_turn",
+                }
+            ),
+        ]
+    )
+    posted = []
+
+    class FakeAsyncClient:
+        async def post(self, url, *, json):
+            posted.append((url, json))
+            return next(responses)
+
+    body = asyncio.run(
+        anthropic_session_verify_agent._post_complete(
+            FakeAsyncClient(),
+            "http://session/v1/messages",
+            {"messages": []},
+            label="turn",
+            assert_response=anthropic_session_verify_agent._assert_anthropic_text_response,
+        )
+    )
+
+    assert body["stop_reason"] == "end_turn"
+    assert posted == [
+        ("http://session/v1/messages", {"messages": []}),
+        ("http://session/v1/messages", {"messages": []}),
+    ]
+
+
+def test_post_complete_retries_empty_text_response():
+    responses = iter(
+        [
+            _Response(
+                {
+                    "type": "message",
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": ""}],
+                    "stop_reason": "end_turn",
+                }
+            ),
+            _Response(
+                {
+                    "type": "message",
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": "complete"}],
+                    "stop_reason": "end_turn",
+                }
+            ),
+        ]
+    )
+
+    class FakeAsyncClient:
+        async def post(self, url, *, json):
+            return next(responses)
+
+    body = asyncio.run(
+        anthropic_session_verify_agent._post_complete(
+            FakeAsyncClient(),
+            "http://session/v1/messages",
+            {"messages": []},
+            label="turn",
+            assert_response=anthropic_session_verify_agent._assert_anthropic_text_response,
+        )
+    )
+
+    assert body["content"] == [{"type": "text", "text": "complete"}]
+
+
+def test_post_complete_raises_after_bounded_incomplete_responses():
+    class FakeAsyncClient:
+        attempts = 0
+
+        async def post(self, url, *, json):
+            self.attempts += 1
+            return _Response(
+                {
+                    "type": "message",
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": ""}],
+                    "stop_reason": "end_turn",
+                }
+            )
+
+    client = FakeAsyncClient()
+    with pytest.raises(AssertionError):
+        asyncio.run(
+            anthropic_session_verify_agent._post_complete(
+                client,
+                "http://session/v1/messages",
+                {"messages": []},
+                label="turn",
+                assert_response=anthropic_session_verify_agent._assert_anthropic_text_response,
+            )
+        )
+
+    assert client.attempts == anthropic_session_verify_agent._MAX_ANTHROPIC_INCOMPLETE_TURN_RETRIES + 1
+
+
+def _successful_sample() -> Sample:
+    return Sample(
+        metadata={
+            "endpoint": "anthropic",
+            "driver_events": anthropic_session_verify_agent._expected_driver_events(include_system=True),
+            "request_count": 6,
+            "tool_use_count": 3,
+            "tool_result_count": 3,
+            "text_turn_count": 3,
+            "tool_result_string_count": 2,
+            "tool_result_list_count": 1,
+            "intermediate_system_used": True,
+            "leaf": {"path_node_ids": [0, 1, 2, 3, 4, 7]},
+            "tito_session_mismatch": [],
+        }
+    )
+
+
+def test_generate_injects_model_and_writes_tito_metrics(monkeypatch, tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    monkeypatch.setenv("MILES_SESSION_VERIFY_METRICS_PATH", str(metrics_path))
+    returned_sample = _successful_sample()
+
+    async def fake_base_generate(input):
+        assert input.sample.metadata["anthropic_model"] == "/models/test"
+        assert input.sample.metadata["tito_model"] == "qwen3"
+        assert input.sample.metadata["anthropic_intermediate_system_expectation"] == "required"
+        return GenerateFnOutput(samples=[returned_sample])
+
+    monkeypatch.setattr(anthropic_session_verify_agent, "_base_generate", fake_base_generate)
+    input_sample = Sample()
+    input_value = SimpleNamespace(
+        sample=input_sample,
+        args=SimpleNamespace(
+            hf_checkpoint="/models/test",
+            tito_model="qwen3",
+            anthropic_intermediate_system_expectation="required",
+        ),
+    )
+
+    output = asyncio.run(anthropic_session_verify_agent.generate(input_value))
+
+    assert output.samples == [returned_sample]
+    assert input_sample.metadata == {
+        "anthropic_model": "/models/test",
+        "tito_model": "qwen3",
+        "anthropic_intermediate_system_expectation": "required",
+    }
+    [metric] = [json.loads(line) for line in metrics_path.read_text().splitlines()]
+    assert metric["driver_events"] == anthropic_session_verify_agent._expected_driver_events(include_system=True)
+    assert metric["had_assistant_mismatch"] is False
+    assert metric["total_mismatches"] == 0
+    assert metric["hard_mismatch_count"] == 0
+
+
+def test_generate_records_hard_mismatch_without_dropping_sample(monkeypatch, tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    monkeypatch.setenv("MILES_SESSION_VERIFY_METRICS_PATH", str(metrics_path))
+    returned_sample = _successful_sample()
+    returned_sample.metadata["tito_session_mismatch"] = [
+        {
+            "type": "special_token_type",
+            "segment_index": 4,
+            "detail": "special token differs",
+        }
+    ]
+
+    async def fake_base_generate(input):
+        return GenerateFnOutput(samples=[returned_sample])
+
+    monkeypatch.setattr(anthropic_session_verify_agent, "_base_generate", fake_base_generate)
+    input_value = SimpleNamespace(
+        sample=Sample(),
+        args=SimpleNamespace(
+            hf_checkpoint="/models/test",
+            tito_model="qwen3",
+            anthropic_intermediate_system_expectation="required",
+        ),
+    )
+
+    output = asyncio.run(anthropic_session_verify_agent.generate(input_value))
+
+    assert output.samples == [returned_sample]
+    [metric] = [json.loads(line) for line in metrics_path.read_text().splitlines()]
+    assert metric["hard_mismatch_count"] == 1
+    assert metric["hard_mismatch_types"] == ["special_token_type"]
+    assert metric["hard_mismatch_example"]["segment_index"] == 4
+
+
+def test_generate_journals_stale_single_turn_metadata(monkeypatch, tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    monkeypatch.setenv("MILES_SESSION_VERIFY_METRICS_PATH", str(metrics_path))
+    sample = _successful_sample()
+    sample.metadata["request_count"] = 1
+
+    async def fake_base_generate(input):
+        return GenerateFnOutput(samples=[sample])
+
+    monkeypatch.setattr(anthropic_session_verify_agent, "_base_generate", fake_base_generate)
+    input_value = SimpleNamespace(
+        sample=Sample(),
+        args=SimpleNamespace(
+            hf_checkpoint="/models/test",
+            tito_model="qwen3",
+            anthropic_intermediate_system_expectation="required",
+        ),
+    )
+
+    with pytest.raises(AssertionError, match="request_count=1, expected 6"):
+        asyncio.run(anthropic_session_verify_agent.generate(input_value))
+
+    sample_record, error_record = [json.loads(line) for line in metrics_path.read_text().splitlines()]
+    assert sample_record["hard_mismatch_count"] == 0
+    assert error_record["verification_error"]["stage"] == "anthropic_session_verify_agent.generate"
+
+
+@pytest.mark.parametrize(
+    ("tito_model", "route_supports", "expected"),
+    [
+        ("qwen3", True, True),
+        ("qwen3", False, False),
+        ("qwen35", True, False),
+        ("qwen35", False, False),
+    ],
+)
+def test_intermediate_system_family_gate(tito_model, route_supports, expected):
+    assert (
+        anthropic_session_verify_agent._verify_intermediate_system(tito_model, route_supports=route_supports)
+        is expected
+    )
+
+
+@pytest.mark.parametrize(
+    ("tito_model", "route_supports", "expectation"),
+    [("qwen3", False, "required"), ("qwen35", True, "required")],
+)
+def test_run_agent_rejects_intermediate_system_capability_drift_before_post(
+    monkeypatch, tmp_path, tito_model, route_supports, expectation
+):
+    metrics_path = tmp_path / "metrics.jsonl"
+    monkeypatch.setenv("MILES_SESSION_VERIFY_METRICS_PATH", str(metrics_path))
+
+    class FakeAsyncClient:
+        def __init__(self, *, timeout):
+            assert timeout == 180
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return None
+
+        async def get(self, url):
+            assert url == "http://session/health"
+            return _Response({"status": "ok", "anthropic_intermediate_system_supported": route_supports})
+
+        async def post(self, url, *, json):
+            raise AssertionError("request must not be sent after a capability mismatch")
+
+    monkeypatch.setattr(anthropic_session_verify_agent.httpx, "AsyncClient", FakeAsyncClient)
+
+    with pytest.raises(AssertionError, match="did not match the per-model E2E contract"):
+        asyncio.run(
+            anthropic_session_verify_agent.run_agent(
+                "http://session",
+                prompt=None,
+                request_kwargs={"max_tokens": 128},
+                metadata={
+                    "anthropic_model": "/models/test",
+                    "tito_model": tito_model,
+                    "anthropic_intermediate_system_expectation": expectation,
+                },
+            )
+        )
+
+    [record] = [json.loads(line) for line in metrics_path.read_text().splitlines()]
+    assert record["verification_error"]["stage"] == "anthropic_session_verify_agent.run_agent"
+
+
+@pytest.mark.parametrize("tito_model", ["minimax_m25", "minimax_m27"])
+def test_minimax_is_rejected_before_base_generate(monkeypatch, tito_model):
+    called = False
+
+    async def fake_base_generate(input):
+        nonlocal called
+        called = True
+        return GenerateFnOutput(samples=[])
+
+    monkeypatch.setattr(anthropic_session_verify_agent, "_base_generate", fake_base_generate)
+    input_value = SimpleNamespace(
+        sample=Sample(),
+        args=SimpleNamespace(hf_checkpoint="/models/test", tito_model=tito_model),
+    )
+
+    with pytest.raises(AssertionError, match="does not support tito_model"):
+        asyncio.run(anthropic_session_verify_agent.generate(input_value))
+    assert called is False
+
+
+@pytest.mark.parametrize("tito_model", ["minimax_m25", "minimax_m27"])
+def test_run_agent_rejects_minimax_before_http(tito_model):
+    with pytest.raises(AssertionError, match="does not support tito_model"):
+        asyncio.run(
+            anthropic_session_verify_agent.run_agent(
+                "http://session",
+                prompt=None,
+                request_kwargs={"max_tokens": 128},
+                metadata={"anthropic_model": "/models/test", "tito_model": tito_model},
+            )
+        )
diff --git a/tests/fast/utils/test_utils/test_session_verify_agent.py b/tests/fast/utils/test_utils/test_session_verify_agent.py
index 7cc5551fc94..f203d34458b 100644
--- a/tests/fast/utils/test_utils/test_session_verify_agent.py
+++ b/tests/fast/utils/test_utils/test_session_verify_agent.py
@@ -1,8 +1,12 @@
 import asyncio
+import json
 from copy import deepcopy
+from types import SimpleNamespace
 
+import pytest
 from tests.e2e.sglang.test_session_server_multi_role._common import ModelConfig
 
+from miles.rollout.base_types import GenerateFnOutput
 from miles.utils.chat_template_utils.tito_tokenizer import VALID_APPEND_ROLES
 from miles.utils.test_utils import session_verify_agent
 from miles.utils.test_utils.session_verify_agent import (
@@ -13,6 +17,7 @@
     run_agent,
     select_schedule,
 )
+from miles.utils.types import Sample
 
 
 def test_all_role_schedule_places_assistant_input_last():
@@ -93,6 +98,35 @@ async def fake_chat(client, base_url, messages, request_kwargs, *, label):
     assert result["user_count"] == 1
 
 
+def test_chat_complete_retries_length_with_distinct_seed(monkeypatch):
+    attempts = []
+    responses = iter(
+        [
+            {"choices": [{"finish_reason": "length"}]},
+            {"choices": [{"finish_reason": "stop"}]},
+        ]
+    )
+
+    async def fake_chat(client, base_url, messages, request_kwargs, *, label):
+        attempts.append(request_kwargs)
+        return next(responses)
+
+    monkeypatch.setattr(session_verify_agent, "_chat", fake_chat)
+
+    response = asyncio.run(
+        session_verify_agent._chat_complete(
+            None,
+            "http://session",
+            [],
+            {"seed": 7},
+            label="turn",
+        )
+    )
+
+    assert response["choices"][0]["finish_reason"] == "stop"
+    assert [kwargs["seed"] for kwargs in attempts] == [7, 1_000_007]
+
+
 def test_minimax_schedule_excludes_system_and_keeps_assistant_rollback():
     roles = fixed_template_append_roles("minimax_m27")
     schedule = select_schedule(roles, cycles=1)
@@ -169,3 +203,137 @@ async def fake_chat(client, base_url, messages, request_kwargs, *, label):
     assert result["assistant_input_count"] == 2
     assert result["user_count"] == 1
     assert result["rollback_count"] == 1
+
+
+def _qwen3_verified_sample(mismatches):
+    return Sample(
+        metadata={
+            "driver_events": [
+                "initial",
+                "append_tool",
+                "append_user",
+                "append_system",
+                "append_assistant",
+                "rollback",
+            ],
+            "tito_session_mismatch": mismatches,
+        }
+    )
+
+
+def test_generate_records_hard_mismatch_without_dropping_sample(monkeypatch, tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    monkeypatch.setenv("MILES_SESSION_VERIFY_METRICS_PATH", str(metrics_path))
+    writes = []
+    real_write = session_verify_agent.os.write
+
+    def tracked_write(fd, payload):
+        writes.append(payload)
+        return real_write(fd, payload)
+
+    monkeypatch.setattr(session_verify_agent.os, "write", tracked_write)
+    returned_sample = _qwen3_verified_sample(
+        [
+            {
+                "type": "special_token_count",
+                "segment_index": -1,
+                "detail": "segment count differs: expected 99, got 97",
+            },
+            {
+                "type": "assistant_text",
+                "segment_index": 3,
+                "expected_text": "expected",
+                "actual_text": "actual",
+            },
+        ]
+    )
+
+    async def fake_base_generate(input):
+        return GenerateFnOutput(samples=[returned_sample])
+
+    monkeypatch.setattr(session_verify_agent, "_base_generate", fake_base_generate)
+    input_value = SimpleNamespace(
+        sample=Sample(),
+        args=SimpleNamespace(
+            tito_model="qwen3",
+            session_verify_cycles=3,
+            tool_call_failure_mode="rollback",
+        ),
+    )
+
+    output = asyncio.run(session_verify_agent.generate(input_value))
+
+    assert output.samples == [returned_sample]
+    [metric] = [json.loads(line) for line in metrics_path.read_text().splitlines()]
+    assert metric["hard_mismatch_count"] == 1
+    assert metric["hard_mismatch_types"] == ["special_token_count"]
+    assert metric["hard_mismatch_example"] == {
+        "type": "special_token_count",
+        "segment_index": -1,
+        "detail": "segment count differs: expected 99, got 97",
+    }
+    assert metric["had_assistant_mismatch"] is True
+    assert writes == [(json.dumps(metric) + "\n").encode()]
+
+
+def test_hard_mismatch_still_raises_without_metrics_sidecar(monkeypatch):
+    monkeypatch.delenv("MILES_SESSION_VERIFY_METRICS_PATH", raising=False)
+    sample = _qwen3_verified_sample([{"type": "non_assistant_text"}])
+
+    with pytest.raises(AssertionError, match="forbidden mismatches"):
+        session_verify_agent._verify_tito_samples(
+            [sample],
+            [sample.metadata["driver_events"]],
+            allowed_roles=list(VALID_APPEND_ROLES),
+        )
+
+
+def test_run_agent_journals_assertion_before_retry(monkeypatch, tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    monkeypatch.setenv("MILES_SESSION_VERIFY_METRICS_PATH", str(metrics_path))
+
+    def reject_template(_tito_model):
+        raise AssertionError("journal me")
+
+    monkeypatch.setattr(session_verify_agent, "fixed_template_append_roles", reject_template)
+
+    with pytest.raises(AssertionError, match="journal me"):
+        asyncio.run(
+            run_agent(
+                "http://session",
+                prompt=None,
+                request_kwargs={},
+                metadata={"tito_model": "qwen3"},
+            )
+        )
+
+    [record] = [json.loads(line) for line in metrics_path.read_text().splitlines()]
+    assert record["verification_error"]["stage"] == "session_verify_agent.run_agent"
+    assert record["verification_error"]["message"] == "journal me"
+
+
+def test_generate_journals_coverage_failure_after_hard_mismatch(monkeypatch, tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    monkeypatch.setenv("MILES_SESSION_VERIFY_METRICS_PATH", str(metrics_path))
+    sample = _qwen3_verified_sample([{"type": "special_token_count", "detail": "count differs"}])
+    sample.metadata["driver_events"].remove("rollback")
+
+    async def fake_base_generate(input):
+        return GenerateFnOutput(samples=[sample])
+
+    monkeypatch.setattr(session_verify_agent, "_base_generate", fake_base_generate)
+    input_value = SimpleNamespace(
+        sample=Sample(),
+        args=SimpleNamespace(
+            tito_model="qwen3",
+            session_verify_cycles=3,
+            tool_call_failure_mode="rollback",
+        ),
+    )
+
+    with pytest.raises(AssertionError, match="missing required driver events"):
+        asyncio.run(session_verify_agent.generate(input_value))
+
+    sample_record, error_record = [json.loads(line) for line in metrics_path.read_text().splitlines()]
+    assert sample_record["hard_mismatch_count"] == 1
+    assert error_record["verification_error"]["stage"] == "session_verify_agent.generate"
diff --git a/tests/fast/utils/test_utils/test_session_verify_runner.py b/tests/fast/utils/test_utils/test_session_verify_runner.py
index d9996dd0c51..f2d41f70336 100644
--- a/tests/fast/utils/test_utils/test_session_verify_runner.py
+++ b/tests/fast/utils/test_utils/test_session_verify_runner.py
@@ -1,14 +1,18 @@
 import argparse
 import json
+import os
+import subprocess
 
 import pytest
 from tests.e2e.sglang.test_session_server_multi_role import _common
 
+from miles.utils.test_utils import session_verify_runner
 from miles.utils.test_utils.session_verify_runner import (
     SESSION_VERIFY_INVARIANT_ARGS,
     assert_session_verify_metrics,
     namespace_to_train_args,
 )
+from miles.utils.tracking_utils.ci_history import RECORD_DIR_ENV
 
 
 def _build_args(**overrides) -> str:
@@ -26,6 +30,7 @@ def _build_args(**overrides) -> str:
         "sglang_tool_call_parser": "qwen25",
         "sglang_context_length": None,
         "sglang_cuda_graph_backend_prefill": None,
+        "anthropic_intermediate_system_expectation": None,
     }
     values.update(overrides)
     return namespace_to_train_args(argparse.Namespace(**values))
@@ -62,6 +67,18 @@ def test_namespace_to_train_args_allows_session_server_v1():
     assert "--use-session-server v1" in train_args
 
 
+def test_namespace_to_train_args_emits_session_message_matcher():
+    assert "--session-message-matcher strict" in _build_args()
+    assert "--session-message-matcher loose_tool_call" in _build_args(session_message_matcher="loose_tool_call")
+
+
+def test_namespace_to_train_args_emits_anthropic_intermediate_system_expectation():
+    assert "--anthropic-intermediate-system-expectation" not in _build_args()
+    assert "--anthropic-intermediate-system-expectation required" in _build_args(
+        anthropic_intermediate_system_expectation="required"
+    )
+
+
 def test_namespace_to_train_args_has_no_append_role_policy_flag():
     train_args = _build_args()
 
@@ -97,7 +114,11 @@ def test_run_one_aligns_global_batch_size_with_sample_count(
     monkeypatch, n_samples_per_prompt, expected_global_batch_size
 ):
     captured = {}
-    monkeypatch.setattr(_common, "run_session_verify", lambda args: captured.setdefault("args", args))
+    monkeypatch.setattr(
+        _common,
+        "run_session_verify",
+        lambda args, *, wire_format: captured.update(args=args, wire_format=wire_format),
+    )
     config = _common.ModelConfig(
         model_name="test-model",
         reasoning_parser="qwen3",
@@ -114,12 +135,17 @@ def test_run_one_aligns_global_batch_size_with_sample_count(
     assert captured["args"].rollout_batch_size == 16
     assert captured["args"].rollout_max_response_len == 4096
     assert captured["args"].sglang_cuda_graph_backend_prefill == "disabled"
+    assert captured["wire_format"] == "openai"
 
 
 @pytest.mark.parametrize("version", ["v1", "v2"])
 def test_run_one_uses_requested_session_server_version(monkeypatch, version):
     captured = {}
-    monkeypatch.setattr(_common, "run_session_verify", lambda args: captured.setdefault("args", args))
+    monkeypatch.setattr(
+        _common,
+        "run_session_verify",
+        lambda args, *, wire_format: captured.update(args=args, wire_format=wire_format),
+    )
     config = _common.ModelConfig(
         model_name="test-model",
         reasoning_parser="qwen3",
@@ -130,25 +156,108 @@ def test_run_one_uses_requested_session_server_version(monkeypatch, version):
     _common.run_one(config, session_server_version=version)
 
     assert captured["args"].use_session_server == version
+    assert captured["wire_format"] == "openai"
+
+
+def test_run_one_rejects_anthropic_on_v1():
+    config = _common.ModelConfig(
+        model_name="test-model",
+        reasoning_parser="qwen3",
+        tool_call_parser="qwen25",
+        tito_model="qwen3",
+    )
+
+    with pytest.raises(ValueError, match="requires session server v2"):
+        _common.run_one(config, session_server_version="v1", endpoint="anthropic")
+
+
+def test_run_one_requires_explicit_anthropic_intermediate_system_expectation():
+    config = _common.ModelConfig(
+        model_name="test-model",
+        reasoning_parser="qwen3",
+        tool_call_parser="qwen25",
+        tito_model="qwen3",
+    )
+
+    with pytest.raises(ValueError, match="requires an intermediate-system expectation"):
+        _common.run_one(config, endpoint="anthropic")
 
 
 @pytest.mark.parametrize(("n_samples_per_prompt", "expected_global_batch_size"), [(1, 8), (4, 32)])
-def test_run_both_versions_splits_rollout_batch(monkeypatch, n_samples_per_prompt, expected_global_batch_size):
+@pytest.mark.parametrize(
+    ("threshold_overrides", "expected_thresholds"),
+    [
+        ({"assistant_text_threshold": 0.25}, [0.25, 0.25, 0.25]),
+        (
+            {"assistant_text_threshold": 0.25, "anthropic_assistant_text_threshold": 1.0},
+            [0.25, 0.25, 1.0],
+        ),
+    ],
+)
+def test_run_both_versions_adds_v2_anthropic_pass(
+    monkeypatch,
+    n_samples_per_prompt,
+    expected_global_batch_size,
+    threshold_overrides,
+    expected_thresholds,
+):
     captured = []
-    monkeypatch.setattr(_common, "run_session_verify", lambda args: captured.append(args))
+    monkeypatch.setattr(
+        _common,
+        "run_session_verify",
+        lambda args, *, wire_format: captured.append((args, wire_format)),
+    )
     config = _common.ModelConfig(
         model_name="test-model",
         reasoning_parser="qwen3",
         tool_call_parser="qwen25",
         tito_model="qwen3",
         n_samples_per_prompt=n_samples_per_prompt,
+        anthropic_intermediate_system_expectation="required",
+        **threshold_overrides,
     )
 
     _common.run_both_versions(config)
 
-    assert [args.use_session_server for args in captured] == ["v1", "v2"]
-    assert [args.rollout_batch_size for args in captured] == [8, 8]
-    assert [args.global_batch_size for args in captured] == [expected_global_batch_size] * 2
+    args = [item[0] for item in captured]
+    assert [item[1] for item in captured] == ["openai", "openai", "anthropic"]
+    assert [item.use_session_server for item in args] == ["v1", "v2", "v2"]
+    assert [item.rollout_batch_size for item in args] == [8, 8, 8]
+    assert [item.global_batch_size for item in args] == [expected_global_batch_size] * 3
+    assert [item.assistant_text_threshold for item in args] == expected_thresholds
+    assert [item.session_message_matcher for item in args] == ["strict", "strict", "loose_tool_call"]
+    assert [item.anthropic_intermediate_system_expectation for item in args] == [None, None, "required"]
+    assert [item.custom_generate_function_path for item in args] == [
+        SESSION_VERIFY_INVARIANT_ARGS["custom_generate_function_path"],
+        SESSION_VERIFY_INVARIANT_ARGS["custom_generate_function_path"],
+        _common._ANTHROPIC_GENERATE,
+    ]
+    assert [item.custom_agent_function_path for item in args] == [
+        SESSION_VERIFY_INVARIANT_ARGS["custom_agent_function_path"],
+        SESSION_VERIFY_INVARIANT_ARGS["custom_agent_function_path"],
+        _common._ANTHROPIC_AGENT,
+    ]
+
+
+def test_run_both_versions_can_disable_anthropic(monkeypatch):
+    captured = []
+    monkeypatch.setattr(
+        _common,
+        "run_session_verify",
+        lambda args, *, wire_format: captured.append((args, wire_format)),
+    )
+    config = _common.ModelConfig(
+        model_name="test-model",
+        reasoning_parser="minimax-append-think",
+        tool_call_parser="minimax-m2",
+        tito_model="minimax_m27",
+        verify_anthropic=False,
+    )
+
+    _common.run_both_versions(config)
+
+    assert [wire_format for _, wire_format in captured] == ["openai", "openai"]
+    assert [args.use_session_server for args, _ in captured] == ["v1", "v2"]
 
 
 def test_namespace_to_train_args_omits_expert_parallel_for_single_expert():
@@ -201,3 +310,117 @@ def test_session_verify_metrics_requires_at_least_one_append_tool(tmp_path):
 
     with pytest.raises(AssertionError, match="no sample produced an append_tool action"):
         assert_session_verify_metrics(str(metrics_path), assistant_text_threshold=0.1)
+
+
+def test_session_verify_metrics_can_skip_multi_role_append_tool_gate(tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    _write_metrics(
+        metrics_path,
+        [{"driver_events": ["anthropic_tool_use"], "had_assistant_mismatch": False}],
+    )
+
+    assert_session_verify_metrics(str(metrics_path), assistant_text_threshold=0.1, require_append_tool=False)
+
+
+def test_session_verify_metrics_hard_mismatch_precedes_soft_threshold(tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    _write_metrics(
+        metrics_path,
+        [
+            {
+                "driver_events": [],
+                "had_assistant_mismatch": True,
+                "hard_mismatch_count": 1,
+                "hard_mismatch_types": ["special_token_count"],
+                "hard_mismatch_example": {"type": "special_token_count"},
+            }
+        ],
+    )
+
+    with pytest.raises(AssertionError, match="hard TITO mismatches.*special_token_count"):
+        assert_session_verify_metrics(str(metrics_path), assistant_text_threshold=0.0)
+
+
+@pytest.mark.parametrize(
+    ("include_clean_sample", "stage", "message", "completed_samples"),
+    [
+        (True, "anthropic_session_verify_agent.generate", "expected six requests", 1),
+        (False, "session_verify_agent.run_agent", "missing rollback", 0),
+    ],
+)
+def test_session_verify_metrics_rejects_verifier_error(
+    tmp_path, include_clean_sample, stage, message, completed_samples
+):
+    metrics_path = tmp_path / "metrics.jsonl"
+    entries = [{"verification_error": {"stage": stage, "type": "AssertionError", "message": message}}]
+    if include_clean_sample:
+        entries.append({"driver_events": ["append_tool"], "had_assistant_mismatch": False})
+    _write_metrics(metrics_path, entries)
+
+    with pytest.raises(AssertionError, match=f"completed_samples={completed_samples}.*{message}"):
+        assert_session_verify_metrics(str(metrics_path), assistant_text_threshold=1.0)
+
+
+def test_session_verify_metrics_keeps_assistant_text_soft(tmp_path):
+    metrics_path = tmp_path / "metrics.jsonl"
+    _write_metrics(
+        metrics_path,
+        [{"driver_events": ["append_tool"], "had_assistant_mismatch": True}],
+    )
+
+    assert_session_verify_metrics(str(metrics_path), assistant_text_threshold=1.0)
+    with pytest.raises(AssertionError, match="assistant_text mismatch ratio"):
+        assert_session_verify_metrics(str(metrics_path), assistant_text_threshold=0.0)
+
+
+@pytest.mark.parametrize("failure_phase", ["execute_train", "post_gate"])
+def test_run_session_verify_preserves_sidecar_on_failure(monkeypatch, tmp_path, failure_phase):
+    metrics_path = tmp_path / "metrics.jsonl"
+    metrics_fd = os.open(metrics_path, os.O_CREAT | os.O_RDWR)
+    monkeypatch.setattr(session_verify_runner.tempfile, "mkstemp", lambda **kwargs: (metrics_fd, str(metrics_path)))
+    monkeypatch.setattr(
+        session_verify_runner,
+        "resolve_reasoning_and_tool_call_parser",
+        lambda tito_model, reasoning_parser, tool_call_parser: (reasoning_parser, tool_call_parser),
+    )
+    monkeypatch.setattr(session_verify_runner, "_ensure_prompt_data", lambda: None)
+    monkeypatch.setattr(session_verify_runner, "_clear_proxy_env", lambda: None)
+    monkeypatch.setattr(session_verify_runner, "_ensure_model_downloaded", lambda checkpoint: checkpoint)
+    monkeypatch.setattr(session_verify_runner, "namespace_to_train_args", lambda args: "train args")
+
+    def fake_execute_train(**kwargs):
+        sidecar = kwargs["extra_env_vars"]["MILES_SESSION_VERIFY_METRICS_PATH"]
+        with open(sidecar, "w") as f:
+            f.write('{"hard_mismatch_count": 1}\n')
+        if failure_phase == "execute_train":
+            raise subprocess.CalledProcessError(1, "ray job submit")
+
+    monkeypatch.setattr(session_verify_runner.U, "execute_train", fake_execute_train)
+    args = argparse.Namespace(
+        tito_model="qwen3",
+        sglang_reasoning_parser="qwen3",
+        sglang_tool_call_parser="qwen25",
+        hf_checkpoint="/models/test",
+        actor_num_gpus_per_node=8,
+        assistant_text_threshold=0.2,
+    )
+
+    expected_error = subprocess.CalledProcessError if failure_phase == "execute_train" else AssertionError
+    with pytest.raises(expected_error):
+        session_verify_runner.run_session_verify(args)
+
+    assert not metrics_path.exists()
+    assert (tmp_path / "metrics.jsonl.failed").read_text() == '{"hard_mismatch_count": 1}\n'
+
+
+@pytest.mark.parametrize(("wire_format", "disables_history"), [("openai", False), ("anthropic", True)])
+def test_session_verify_env_isolates_anthropic_from_openai_history(wire_format, disables_history):
+    args = argparse.Namespace(tito_model="qwen3")
+
+    env = session_verify_runner._session_verify_env(args, "/tmp/metrics.jsonl", wire_format=wire_format)
+
+    assert env["MILES_TITO_MODEL"] == "qwen3"
+    assert env["MILES_SESSION_VERIFY_METRICS_PATH"] == "/tmp/metrics.jsonl"
+    assert (RECORD_DIR_ENV in env) is disables_history
+    if disables_history:
+        assert env[RECORD_DIR_ENV] == ""