diff --git a/apps/aiperf_qual/README.md b/apps/aiperf_qual/README.md
index ca9e4063d5..8340e40165 100644
--- a/apps/aiperf_qual/README.md
+++ b/apps/aiperf_qual/README.md
@@ -744,3 +744,42 @@ A selection failure or changed/missing input/output fails verification and
 preserves its ledger and request artifacts. Family quality criteria are unchanged;
 this workload verifies transport and reports performance, without adding a new
 numerical accuracy gate.
+
+### Pinned conversations
+
+Use the same environment and already-built Qwen3-0.6B FP16 bundle with
+`config/text/qwen3-0.6b-conversations.yaml`:
+
+```bash
+python -m trtmc_aiperf_qual text-profile \
+  --environment environment.yaml --config qwen3-0.6b-conversations.yaml \
+  --out artifacts/qwen-conversations
+```
+
+The current Qwen family library implements the existing native
+`TextConversation` contract for supported ChatML templates. The server forwards
+typed roles and text; Qwen owns template rendering and tokenization. Existing
+single-turn generation and its correctness criteria remain unchanged. Rebuild
+the server/runtime libraries for history support; existing engine bundles can
+be reused. This profile uses nonstreaming chat; history streaming, tools and
+non-text messages are unsupported.
+
+The profile pins the same ShareGPT file and tokenizer as the single-turn
+example. It selects six distinct fitting human/assistant/human prefixes and
+sends their first two original human strings through AIPerf's existing
+`multi_turn` loader. AIPerf captures actual native answers for subsequent
+history (`deltas_without_responses`); source assistant answers are not replayed.
+The second turn has 50 ms of think time. Twelve measured requests cover both
+turns of all six conversations, with two separate warmup requests.
+
+`corpus.json` preserves source indices, original user strings, considered
+rejections, the selection budget and the exact delay policy. The run's
+`corpus-verification.json` checks roles/order, original text, retained actual
+assistant output, native dispatch hashes, and every measured turn by request
+and conversation identity. Actual wire histories are counted with the pinned
+tokenizer and checked against the bundle context plus requested output;
+native input-token counts remain unavailable. Configured and observed
+between-turn gaps are recorded separately, without claiming exact scheduler
+precision. `report.json` includes client and native Task-call timings for each
+turn as well as the full workload. Startup/build/load/warmup are excluded from
+measured requests. No TTFT/ITL or cross-request cache acceleration is claimed.
diff --git a/apps/aiperf_qual/config/text/qwen3-0.6b-conversations.yaml b/apps/aiperf_qual/config/text/qwen3-0.6b-conversations.yaml
new file mode 100644
index 0000000000..c3e062fddc
--- /dev/null
+++ b/apps/aiperf_qual/config/text/qwen3-0.6b-conversations.yaml
@@ -0,0 +1,42 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+# Replace /path values with paths on the validation host. Validation must use
+# this bundle and the same family DSO that the server will load.
+aiperf:
+  version: 0.13.0
+checkpoint:
+  name: Qwen/Qwen3-0.6B
+  revision: c1899de289a04d12100db370d81485cdf75e47ca
+  precision: fp16
+tokenizer:
+  name: Qwen/Qwen3-0.6B
+  revision: c1899de289a04d12100db370d81485cdf75e47ca
+server:
+  bundle: /path/bundles/qwen3-0.6b-fp16.bundle
+  model_name: Qwen/Qwen3-0.6B
+  binary: /path/build/trtmc-server
+  runtime_root: /path/build
+  replicas: 1
+  port: 8000
+  startup_timeout: 120
+  request_timeout: 120
+validation_profile: qwen3-0.6b-fp16
+validation_commands:
+  - [/path/trtmc-python, -m, pytest, families/qwen/tests/test_e2e.py, --e2e-model, qwen3-0.6b-fp16, -q]
+# A workload configuration for the existing full FP16 model profile above.
+corpus:
+  format: sharegpt_multi_turn
+  count: 6
+  think_time_ms: 50
+  source:
+    repo: anon8231489123/ShareGPT_Vicuna_unfiltered
+    revision: 192ab2185289094fc556ec8ce5ce1e8e587154ca
+    filename: ShareGPT_V3_unfiltered_cleaned_split.json
+    sha256: 35f0e213ce091ed9b9af2a1f0755e9d39f9ccec34ab281cd4ca60d70f6479ba4
+workload:
+  endpoints: [chat]
+  output_tokens: [16]
+  concurrency: [1]
+  streaming: [false]
+  requests: 12
+  warmup_requests: 2
diff --git a/apps/aiperf_qual/tests/test_text_conversations.py b/apps/aiperf_qual/tests/test_text_conversations.py
new file mode 100644
index 0000000000..da1946752c
--- /dev/null
+++ b/apps/aiperf_qual/tests/test_text_conversations.py
@@ -0,0 +1,122 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+import copy
+import hashlib
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from trtmc_aiperf_qual.config import ConfigError
+from trtmc_aiperf_qual.text_conversations import fingerprint, select, template_tokens, verify_records
+from trtmc_aiperf_qual.text_profile import arguments, cases
+
+
+def conversation(first, second):
+    return {"conversations": [{"from": "human", "value": first}, {"from": "gpt", "value": "source answer"},
+                              {"from": "human", "value": second}]}
+
+
+def count_messages(messages):
+    return sum(len(message["content"]) for message in messages) + 10
+
+
+def test_context_budget_counts_token_ids_instead_of_tokenizer_container_fields():
+    class Tokenizer:
+        def apply_chat_template(self, messages, **options):
+            assert options == {"tokenize": False, "add_generation_prompt": True, "enable_thinking": False}
+            return "rendered original history"
+
+        def encode(self, text, *, add_special_tokens):
+            assert text == "rendered original history" and add_special_tokens is False
+            return [100, 200, 300, 400]
+
+    assert template_tokens(Tokenizer(), []) == 4
+
+
+def test_original_user_turns_survive_selection_and_rejections_are_visible():
+    original = "  first\n\u2028second  "
+    row = conversation(original, "Follow up?")
+    selected, ledger = select([None, conversation("x" * 200, "Q"), row, row,
+                               conversation("another", "follow up")], 2, 100, 8, count_messages)
+    assert [entry["source_index"] for entry in selected] == [2, 4]
+    assert selected[0]["user_turns"] == [original, "Follow up?"]
+    assert [entry["status"] for entry in ledger] == ["rejected", "rejected", "selected", "rejected", "selected"]
+
+
+def test_multi_turn_uses_the_existing_loader_and_counts_both_requests(tmp_path):
+    config = {"corpus": {"format": "sharegpt_multi_turn", "count": 2},
+              "tokenizer": {"name": "pinned", "revision": "a" * 40},
+              "workload": {"endpoints": ["chat"], "requests": 4, "warmup_requests": 2}}
+    case, = cases(config)
+    args = arguments("http://localhost", config, case, input_file=tmp_path / "inputs.jsonl")
+    assert args[args.index("--custom-dataset-type") + 1] == "multi_turn"
+    assert "--synthetic-input-tokens-mean" not in args
+    config["workload"]["requests"] = 3
+    with pytest.raises(ConfigError, match="every prompt"):
+        cases(config)
+
+
+@pytest.fixture
+def capture():
+    texts = ["Remember ORCHID.", "What word?"]
+    original = [{"role": "user", "content": texts[0]}]
+    history = [*original, {"role": "assistant", "content": "OK"}, {"role": "user", "content": texts[1]}]
+    raw, joined = [], []
+    for turn, (messages, text) in enumerate([(original, "OK"), (history, "ORCHID")]):
+        metadata = {"x_request_id": str(turn), "conversation_id": "sharegpt-0", "session_num": turn,
+                    "x_correlation_id": "conversation", "turn_index": turn, "benchmark_phase": "profiling",
+                    "request_start_ns": 1_000_000_000 + turn * 150_000_000,
+                    "request_end_ns": 1_100_000_000 + turn * 150_000_000}
+        raw.append({"metadata": metadata, "payload": {"messages": messages}, "status": 200,
+                    "responses": [{"text": json.dumps({"choices": [{"message": {"content": text}}]})}]})
+        joined.append({"request_id": str(turn), "server": {"status": 200,
+            "native_operation": "converse" if turn else "generate",
+            "prompt_sha256": hashlib.sha256(texts[0].encode()).hexdigest(),
+            "conversation_sha256": fingerprint(messages), "completion_sha256": hashlib.sha256(text.encode()).hexdigest()}})
+    corpus = {"selected": [{"session_id": "sharegpt-0", "user_turns": texts}],
+              "context_limit": 256, "output_tokens": 8, "think_time_ms": 50}
+    return SimpleNamespace(raw_records=lambda phase: raw), joined, corpus, raw
+
+
+def test_live_assistant_history_and_per_turn_identity_are_verified(capture):
+    run, joined, corpus, _ = capture
+    result = verify_records(run, joined, corpus, count_messages)
+    assert result["verified"] and result["measured_conversations"] == 1
+    assert [row["turn_index"] for row in result["requests"]] == [0, 1]
+    assert result["requests"][1]["observed_gap_ms"] == 50
+
+
+@pytest.mark.parametrize("phase", ["profiling", "warmup"])
+def test_incomplete_replayed_profiling_group_is_rejected_even_after_full_coverage(capture, phase):
+    run, joined, corpus, raw = capture
+    replay = copy.deepcopy(raw[0])
+    replay["metadata"].update(x_request_id="replay-first-only", x_correlation_id="another-instance",
+        session_num=2, benchmark_phase=phase, request_start_ns=1_400_000_000, request_end_ns=1_500_000_000)
+    raw.append(replay)
+    joined.append({"request_id": "replay-first-only", "server": copy.deepcopy(joined[0]["server"])})
+    if phase == "profiling":
+        with pytest.raises(ConfigError, match="both turns"):
+            verify_records(run, joined, corpus, count_messages)
+    else:
+        # Warmup request limits may intentionally end before a conversation does.
+        assert verify_records(run, joined, corpus, count_messages)["verified"]
+
+
+@pytest.mark.parametrize("change", ["drop_history", "role", "original_text", "assistant_text", "native_history", "capacity"])
+def test_missing_or_changed_history_cannot_be_reported_as_qualified(capture, change):
+    run, joined, corpus, raw = capture
+    if change == "drop_history":
+        raw[1]["payload"]["messages"] = raw[1]["payload"]["messages"][-1:]
+    elif change == "role":
+        raw[1]["payload"]["messages"][1]["role"] = "user"
+    elif change == "original_text":
+        raw[1]["payload"]["messages"][2]["content"] = "rewritten"
+    elif change == "assistant_text":
+        raw[1]["payload"]["messages"][1]["content"] = "source answer"
+    elif change == "native_history":
+        joined[1]["server"]["conversation_sha256"] = "0" * 64
+    else:
+        corpus["context_limit"] = 16
+    with pytest.raises(ConfigError):
+        verify_records(run, joined, corpus, count_messages)
diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py
new file mode 100644
index 0000000000..d2be5851d1
--- /dev/null
+++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py
@@ -0,0 +1,164 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Fixed original user turns using AIPerf's existing live-response history."""
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from pathlib import Path
+
+from .config import ConfigError
+from .text_corpus import load_source
+from .text_profile import digest, response_text
+
+
+def fingerprint(value) -> str:
+    return hashlib.sha256(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()
+
+
+def template_tokens(tokenizer, messages) -> int:
+    rendered = tokenizer.apply_chat_template(messages, tokenize=False,
+        add_generation_prompt=True, enable_thinking=False)
+    return len(tokenizer.encode(rendered, add_special_tokens=False))
+
+
+def select(records, count, capacity, output, count_messages):
+    if any(type(value) is not int or value < 1 for value in (count, capacity, output)) or output >= capacity:
+        raise ConfigError("conversation count/context/output must be positive integers fitting the bundle")
+    selected, ledger, seen = [], [], set()
+    for index, row in enumerate(records):
+        turns = row.get("conversations", []) if isinstance(row, dict) else []
+        entry = {"source_index": index}
+        if (not isinstance(turns, list) or len(turns) < 3 or
+                any(not isinstance(turns[i], dict) or turns[i].get("from") != role or
+                    not isinstance(turns[i].get("value"), str) or not turns[i]["value"].strip()
+                    for i, role in enumerate(("human", "gpt", "human")))):
+            entry.update(status="rejected", reason="no original human/assistant/human prefix")
+        else:
+            texts = (turns[0]["value"], turns[2]["value"])
+            base = count_messages([{"role": "user", "content": texts[0]},
+                                   {"role": "assistant", "content": ""},
+                                   {"role": "user", "content": texts[1]}])
+            entry["template_tokens_without_live_answer"] = base
+            if texts in seen or base + 2 * output > capacity:
+                entry.update(status="rejected", reason="duplicate turns or conversation/output reserve exceeds context")
+            else:
+                entry.update(status="selected", session_id=f"sharegpt-{index}", user_turns=list(texts))
+                selected.append(entry)
+                seen.add(texts)
+        ledger.append(entry)
+        if len(selected) == count:
+            break
+    return selected, ledger
+
+
+def prepare(config, capacity: int, output: int, out: Path):
+    source, source_hash, records, tokenizer = load_source(config)
+    spec = config["corpus"]
+    delay = spec.get("think_time_ms", 50)
+    if type(delay) not in (int, float) or not math.isfinite(delay) or delay < 0:
+        raise ConfigError("think_time_ms must be a finite nonnegative number")
+    def count_messages(messages):
+        return template_tokens(tokenizer, messages)
+    selected, ledger = select(records, spec["count"], capacity, output, count_messages)
+    path = out / "inputs.jsonl"
+    path.write_text("".join(json.dumps({"session_id": row["session_id"], "turns": [
+        {"text": text, "delay": 0 if index == 0 else delay, "output_length": output}
+        for index, text in enumerate(row["user_turns"])]}, ensure_ascii=False) + "\n"
+        for row in selected), encoding="utf-8")
+    report = {"format": "sharegpt_multi_turn", "source": source, "source_sha256": source_hash,
+              "tokenizer": config["tokenizer"], "context_limit": capacity, "output_tokens": output,
+              "think_time_ms": delay, "source_records": len(records), "considered_records": len(ledger),
+              "unexamined_records": len(records) - len(ledger), "selected": selected, "ledger": ledger,
+              "selection_policy": "first distinct fitting original human/assistant/human prefixes; send the first two human strings unchanged",
+              "assistant_history_policy": "AIPerf deltas_without_responses captures actual native answers; source assistant answers are not replayed",
+              "token_count_scope": "pinned tokenizer chat template; native input count unavailable; actual wire history counted after the run",
+              "input_file": str(path), "input_file_sha256": digest(path)}
+    (out / "corpus.json").write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    if len(selected) != spec["count"]:
+        raise ConfigError("not enough original conversations fit; see corpus.json; no history was truncated")
+    return report
+
+
+def messages_from_payload(payload):
+    messages = payload.get("messages") if isinstance(payload, dict) else None
+    if not isinstance(messages, list):
+        raise ConfigError("conversation payload is missing messages")
+    normalized = []
+    for message in messages:
+        if not isinstance(message, dict):
+            raise ConfigError("conversation message is not an object")
+        content = message.get("content")
+        if isinstance(content, list) and all(isinstance(p, dict) and p.get("type") == "text" and
+                                             isinstance(p.get("text"), str) for p in content):
+            content = "".join(p["text"] for p in content)
+        if not isinstance(content, str):
+            raise ConfigError("conversation content is not text")
+        normalized.append({"role": message.get("role"), "content": content})
+    return normalized
+
+
+def verify(run, joined, corpus):
+    from transformers import AutoTokenizer
+    tokenizer = AutoTokenizer.from_pretrained(corpus["tokenizer"]["name"],
+        revision=corpus["tokenizer"]["revision"], trust_remote_code=False)
+    return verify_records(run, joined, corpus, lambda messages: template_tokens(tokenizer, messages))
+
+
+def verify_records(run, joined, corpus, count_messages):
+    expected = {row["session_id"]: row for row in corpus["selected"]}
+    servers = {row["request_id"]: row["server"] for row in joined}
+    groups, seen, measured, identities = {}, set(), set(), []
+    raw_records = run.raw_records(phase=None)
+    for raw in raw_records:
+        meta = raw.get("metadata", {})
+        if (not isinstance(meta, dict) or not isinstance(meta.get("x_request_id"), str) or
+                not isinstance(meta.get("x_correlation_id"), str) or not meta["x_correlation_id"] or
+                not isinstance(meta.get("conversation_id"), str) or type(meta.get("session_num")) is not int or
+                type(meta.get("turn_index")) is not int or meta["turn_index"] not in (0, 1) or
+                any(type(meta.get(name)) is not int or meta[name] < 1 for name in ("request_start_ns", "request_end_ns"))):
+            raise ConfigError("conversation request metadata is missing identity or timing")
+    for raw in sorted(raw_records, key=lambda r: r["metadata"]["request_start_ns"]):
+        meta = raw["metadata"]
+        request_id, session, turn = meta.get("x_request_id"), meta.get("conversation_id"), meta.get("turn_index")
+        if request_id in seen or session not in expected or type(turn) is not int or turn not in (0, 1):
+            raise ConfigError("unknown or repeated conversation request identity")
+        seen.add(request_id)
+        key = (meta.get("benchmark_phase"), meta["x_correlation_id"])
+        previous = groups.setdefault(key, [])
+        if len(previous) != turn or (previous and previous[0]["session"] != session):
+            raise ConfigError("conversation turns are missing, repeated or out of order")
+        source = expected[session]
+        wanted = [{"role": "user", "content": source["user_turns"][0]}]
+        if turn:
+            wanted += [{"role": "assistant", "content": previous[0]["text"]},
+                       {"role": "user", "content": source["user_turns"][1]}]
+        messages = messages_from_payload(raw.get("payload"))
+        text = response_text(raw)
+        server = servers.get(request_id)
+        if (messages != wanted or not isinstance(text, str) or server is None or server.get("status") != 200 or
+                server.get("completion_sha256") != hashlib.sha256(text.encode()).hexdigest()):
+            raise ConfigError("conversation lost original text, roles, actual assistant history or native output")
+        if turn and (server.get("native_operation") != "converse" or server.get("conversation_sha256") != fingerprint(messages)):
+            raise ConfigError("native conversation dispatch changed history")
+        if not turn and server.get("prompt_sha256") != hashlib.sha256(source["user_turns"][0].encode()).hexdigest():
+            raise ConfigError("native first-turn dispatch changed the original text")
+        tokens = count_messages(messages)
+        if tokens + corpus["output_tokens"] > corpus["context_limit"]:
+            raise ConfigError("actual conversation exceeds bundle context; no history may be dropped")
+        gap = ((meta["request_start_ns"] - previous[0]["end_ns"]) / 1e6) if turn else None
+        if gap is not None and gap < 0:
+            raise ConfigError("conversation dispatched a turn before its previous response completed")
+        previous.append({"text": text, "end_ns": meta["request_end_ns"], "session": session})
+        if meta.get("benchmark_phase") == "profiling":
+            measured.add((session, turn))
+        identities.append({"request_id": request_id, "phase": meta.get("benchmark_phase"),
+            "session_id": session, "session_num": meta["session_num"], "turn_index": turn,
+            "correlation_id": meta.get("x_correlation_id"), "input_tokens": tokens,
+            "messages_sha256": fingerprint(messages), "completion_sha256": server["completion_sha256"],
+            "configured_think_time_ms": corpus["think_time_ms"] if turn else 0, "observed_gap_ms": gap})
+    if (any(len(group) != 2 for (phase, _), group in groups.items() if phase == "profiling") or
+            measured != {(session, turn) for session in expected for turn in (0, 1)}):
+        raise ConfigError("profiling did not cover both turns of every selected conversation")
+    return {"verified": True, "measured_conversations": len(expected), "requests": identities}
diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py
index aab734724d..970c473977 100644
--- a/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py
+++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py
@@ -43,14 +43,14 @@ def select_prompts(records: list[Any], count: int, context: int, output: int,
     return selected, ledger
 
 
-def prepare(config: dict[str, Any], capacity: int, output: int, out: Path) -> dict[str, Any]:
+def load_source(config: dict[str, Any]):
     from transformers import AutoTokenizer
     from .suites import _hf_file, _read_records
 
     spec = config["corpus"]
     source = spec["source"]
-    if spec.get("format") != "sharegpt" or not SHA.fullmatch(str(source.get("revision", ""))):
-        raise ConfigError("real prompts require ShareGPT format and an immutable dataset revision")
+    if not SHA.fullmatch(str(source.get("revision", ""))):
+        raise ConfigError("real prompts require an immutable dataset revision")
     path = _hf_file(source)
     source_hash = digest(path)
     if source.get("sha256") and source["sha256"] != source_hash:
@@ -60,6 +60,17 @@ def prepare(config: dict[str, Any], capacity: int, output: int, out: Path) -> di
         raise ConfigError("ShareGPT source must contain a list of conversations")
     tokenizer = AutoTokenizer.from_pretrained(config["tokenizer"]["name"],
         revision=config["tokenizer"]["revision"], trust_remote_code=False)
+    return source, source_hash, records, tokenizer
+
+
+def prepare(config: dict[str, Any], capacity: int, output: int, out: Path) -> dict[str, Any]:
+    if config["corpus"].get("format") == "sharegpt_multi_turn":
+        from .text_conversations import prepare as prepare_conversations
+        return prepare_conversations(config, capacity, output, out)
+    if config["corpus"].get("format") != "sharegpt":
+        raise ConfigError("unsupported real-prompt corpus format")
+    spec = config["corpus"]
+    source, source_hash, records, tokenizer = load_source(config)
     selected, ledger = select_prompts(records, spec["count"], capacity, output,
                                       lambda text: tokenizer.encode(text, add_special_tokens=False))
     prepared = out / "inputs.jsonl"
@@ -113,7 +124,11 @@ def verify_and_record(run, joined: list[dict[str, Any]], corpus: dict[str, Any])
     """Preserve a verification verdict for missing, changed or malformed inputs."""
     path = run.directory / "corpus-verification.json"
     try:
-        verified = verify(run, joined, corpus)
+        if corpus.get("format") == "sharegpt_multi_turn":
+            from .text_conversations import verify as verify_conversations
+            verified = verify_conversations(run, joined, corpus)
+        else:
+            verified = verify(run, joined, corpus)
         try:
             unchanged = digest(Path(corpus["input_file"])) == corpus["input_file_sha256"]
         except OSError as error:
diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py
index 90508d522f..3dd2ce7128 100644
--- a/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py
+++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py
@@ -102,10 +102,13 @@ def cases(config: dict[str, Any]) -> list[dict[str, Any]]:
             raise ConfigError(f"{name} has an invalid request count")
     if config.get("corpus"):
         count = config["corpus"].get("count")
-        if type(count) is not int or count < 1 or workload.get("requests", 20) < count:
+        multi_turn = config["corpus"].get("format") == "sharegpt_multi_turn"
+        if multi_turn and len({item["output_tokens"] for item in result}) != 1:
+            raise ConfigError("conversation profiles require one output-token cap for their per-turn settings")
+        if type(count) is not int or count < 1 or workload.get("requests", 20) < count * (2 if multi_turn else 1):
             raise ConfigError("corpus count must be positive and every prompt needs a measured request")
-        if any(item["endpoint"] != "completions" or item["streaming"] for item in result):
-            raise ConfigError("real-prompt profiles currently support nonstreaming completions")
+        if any(item["endpoint"] != ("chat" if multi_turn else "completions") or item["streaming"] for item in result):
+            raise ConfigError("real-prompt profiles require nonstreaming chat for conversations or completions for single turns")
     return result
 
 
@@ -121,7 +124,8 @@ def arguments(url: str, config: dict[str, Any], case: dict[str, Any], *, input_f
     if input_file is None:
         result += ["--synthetic-input-tokens-mean", str(case["input_tokens"]), "--synthetic-input-tokens-stddev", "0"]
     else:
-        result += ["--input-file", str(input_file), "--custom-dataset-type", "single_turn"]
+        result += ["--input-file", str(input_file), "--custom-dataset-type",
+                   "multi_turn" if config.get("corpus", {}).get("format") == "sharegpt_multi_turn" else "single_turn"]
     if workload.get("warmup_requests", 3):
         result += ["--warmup-request-count", str(workload.get("warmup_requests", 3)), "--warmup-concurrency", "1"]
     result += ["--extra-inputs", "temperature:0", "seed:0", "top_k:1"]
@@ -305,6 +309,9 @@ def profile(environment: Environment, config_path: Path, out: Path) -> dict[str,
                 entries.append({"case": case, "exit_code": run.exit_code or int(incomplete),
                                 "incomplete_export": incomplete, "command": run.command,
                                 "summary": summarize(rows), "aiperf_summary": run.summary})
+                if corpus and corpus.get("format") == "sharegpt_multi_turn":
+                    entries[-1]["per_turn"] = {str(turn): summarize([row for row in rows
+                        if row["client_metadata"].get("turn_index") == turn]) for turn in (0, 1)}
                 (out / "report.json").write_text(json.dumps({"runs": entries}, indent=2) + "\n")
                 if entries[-1]["exit_code"]:
                     break
diff --git a/apps/server/native_worker.cpp b/apps/server/native_worker.cpp
index ca68a1dcfe..6bf35d81fd 100644
--- a/apps/server/native_worker.cpp
+++ b/apps/server/native_worker.cpp
@@ -279,20 +279,21 @@ int run_protocol(const Json& ready, const Operations& operations, std::istream&
 
 template <class Generate>
 int run_text_protocol(std::int64_t default_tokens, Generate&& generate, std::istream& input,
-                      std::ostream& output, StreamGenerate streaming = {}) {
+                      std::ostream& output, StreamGenerate streaming = {},
+                      Operations operations = {}) {
     Json capabilities = Json::array({"text_generation"});
     if (streaming)
         capabilities.push_back("streaming_text_continuation");
+    if (operations.count("converse"))
+        capabilities.push_back("text_conversation");
     const Json ready{{"event", "ready"},
                      {"protocol_version", 2},
                      {"capabilities", capabilities},
                      {"default_max_new_tokens", default_tokens}};
-    return run_protocol(ready,
-                        {{"generate",
-                          [&](const Json& request) {
-                              return generate(string_field(request, "prompt", true), request);
-                          }}},
-                        input, output, std::move(streaming));
+    operations.emplace("generate", [&](const Json& request) {
+        return generate(string_field(request, "prompt", true), request);
+    });
+    return run_protocol(ready, operations, input, output, std::move(streaming));
 }
 
 const std::vector<float>& checked_embedding_values(const EmbeddingResult& result) {
@@ -943,6 +944,64 @@ int run_model_worker(const Model& model, std::istream& input, std::ostream& outp
     return run_text_worker(model, input, output);
 }
 
+Operations conversation_operations(const Model* model) {
+    if (model == nullptr || !model->supports<TextConversation>())
+        return {};
+    const auto task = model->task<TextConversation>();
+    const auto fields = task.config_fields();
+    return {{"converse", [task, fields](const Json& request) {
+                 const auto found = request.find("messages");
+                 if (found == request.end() || !found->is_array() || found->empty() ||
+                     found->size() > 64)
+                     throw ProtocolError("messages must contain between 1 and 64 text messages");
+                 TextConversationRequest conversation;
+                 for (const auto& message : *found) {
+                     const auto role = string_field(message, "role");
+                     MessageRole typed;
+                     if (role == "system")
+                         typed = MessageRole::System;
+                     else if (role == "user")
+                         typed = MessageRole::User;
+                     else if (role == "assistant")
+                         typed = MessageRole::Assistant;
+                     else
+                         throw ProtocolError("unsupported conversation role");
+                     conversation.messages.push_back(
+                         {typed, {TextPart{string_field(message, "content", true)}}});
+                 }
+                 const auto config = parse_sdk_config(request, fields);
+                 try {
+                     const auto started = std::chrono::steady_clock::now();
+                     const auto result = task.run(conversation, config);
+                     const auto wall_ms = std::chrono::duration<double, std::milli>(
+                                              std::chrono::steady_clock::now() - started)
+                                              .count();
+                     std::string text;
+                     const auto message = result.message();
+                     for (const auto& part : message.parts) {
+                         const auto* value = std::get_if<TextPart>(&part);
+                         if (value == nullptr)
+                             throw std::runtime_error(
+                                 "conversation Task returned a non-text assistant part");
+                         text += value->text;
+                     }
+                     const auto view = result.result().c_view();
+                     return Json{{"text", text},
+                                 {"completion_tokens", result.result().token_ids().size()},
+                                 {"setup_ms", view.setup_ms},
+                                 {"prefill_ms", view.prefill_ms},
+                                 {"decode_ms", view.decode_ms},
+                                 {"model_call_ms", wall_ms},
+                                 {"timing_scope", "public_task_call_wall"}};
+                 } catch (const Error& error) {
+                     if (error.code() == TRTMC_INVALID_ARGUMENT ||
+                         error.code() == TRTMC_INVALID_CONFIG)
+                         throw ProtocolError("conversation request is invalid for this model");
+                     throw;
+                 }
+             }}};
+}
+
 template <class Task, class Request>
 int run_sdk_worker(const Task& task, Request&& make_request, std::istream& input,
                    std::ostream& output, const Model* model = nullptr) {
@@ -1025,7 +1084,7 @@ int run_sdk_worker(const Task& task, Request&& make_request, std::istream& input
                 throw;
             }
         },
-        input, output, std::move(streaming));
+        input, output, std::move(streaming), conversation_operations(model));
 }
 
 } // namespace
diff --git a/apps/server/python/tests/test_app.py b/apps/server/python/tests/test_app.py
index 0a88c718bc..170ba0a8cc 100644
--- a/apps/server/python/tests/test_app.py
+++ b/apps/server/python/tests/test_app.py
@@ -87,6 +87,9 @@ def models(self) -> list[dict[str, Any]]:
     def supports_streaming(self, model: str) -> bool:
         return False
 
+    def supports(self, model: str, capability: str) -> bool:
+        return capability == "text_conversation" and getattr(self, "conversation", False)
+
     def max_tokens(self, model: str, hard_cap: int) -> int:
         if model != "test/model":
             raise ModelNotFoundError(model)
@@ -133,6 +136,44 @@ def test_real_prompt_and_native_output_identity_survive_http(tmp_path):
     assert "prompt" not in row and "text" not in row
 
 
+def test_history_forwards_typed_roles_and_content_without_a_client_template(tmp_path):
+    import hashlib
+
+    registry = FakeRegistry()
+    registry.conversation = True
+    path = tmp_path / "records.jsonl"
+    messages = [{"role": "system", "content": "  system\n"},
+                {"role": "user", "content": "Remember ORCHID."},
+                {"role": "assistant", "content": "OK\u2028"},
+                {"role": "user", "content": "What word?"}]
+    app = create_app(registry, ServerConfig(max_body_bytes=4096, max_prompt_bytes=1024,
+                                          max_generation_tokens=64, records=path))
+    with TestClient(app) as client:
+        response = client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages})
+    assert response.status_code == 200
+    operation, payload = registry.requests[0]
+    assert operation == "converse" and payload["messages"] == messages
+    assert "prompt" not in payload and "use_chat_template" not in payload["config"]
+    row, = [json.loads(line) for line in path.read_text().split("\n") if line]
+    assert row["conversation_sha256"] == hashlib.sha256(json.dumps(
+        messages, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()
+    assert row["conversation_roles"] == ["system", "user", "assistant", "user"]
+    assert "messages" not in row and row["native_operation"] == "converse"
+
+
+def test_history_rejects_unsupported_models_streaming_and_full_history_limits():
+    registry = FakeRegistry()
+    messages = [{"role": "user", "content": "first"}, {"role": "assistant", "content": "answer"},
+                {"role": "user", "content": "last"}]
+    with make_client(registry) as client:
+        assert client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages}).status_code == 400
+        registry.conversation = True
+        assert client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages, "stream": True}).status_code == 400
+        messages[0]["content"] = "x" * 2048
+        assert client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages}).status_code == 413
+    assert registry.requests == []
+
+
 def test_completion_and_chat_keep_model_semantics_in_worker(caplog: Any) -> None:
     registry = FakeRegistry()
     caplog.set_level(logging.INFO, logger="uvicorn.error")
diff --git a/apps/server/python/trtmc_server/app.py b/apps/server/python/trtmc_server/app.py
index 887daa35de..5a02c843e3 100644
--- a/apps/server/python/trtmc_server/app.py
+++ b/apps/server/python/trtmc_server/app.py
@@ -35,7 +35,7 @@
     WorkerTimeoutError,
 )
 from .metrics import Metrics
-from .protocol import chat_prompt, chat_image, extract_vision_result, extract_embedding, extract_ranking, extract_result, extract_transcription, extract_detection, generation_config, public_worker_error
+from .protocol import chat_prompt, chat_history, chat_image, extract_vision_result, extract_embedding, extract_ranking, extract_result, extract_transcription, extract_detection, generation_config, public_worker_error
 from .records import RequestPolicy, RequestRecords
 from .registry import ModelRegistry
 from .schemas import ChatCompletionRequest, CompletionRequest, EmbeddingRequest, GenerationRequest, RankingRequest, TranscriptionRequest, DetectionRequest
@@ -318,6 +318,7 @@ async def execute(
         system_prompt: str = "",
         chat: bool,
         chat_max_tokens: int | None = None,
+        messages: list[dict[str, str]] | None = None,
     ) -> Any:
         context["model"] = request.model
         context.update(prompt_sha256=hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
@@ -338,7 +339,9 @@ async def execute(
             return error_response(
                 400, "unsupported_parameter", "stop is not supported", param="stop"
             )
-        if len(prompt.encode("utf-8")) + len(system_prompt.encode("utf-8")) > config.max_prompt_bytes:
+        input_bytes = (sum(len(message["content"].encode("utf-8")) for message in messages) if messages is not None
+                       else len(prompt.encode("utf-8")) + len(system_prompt.encode("utf-8")))
+        if input_bytes > config.max_prompt_bytes:
             metrics.reject(route, 413)
             return error_response(
                 413, "content_too_large", "prompt exceeds the server limit", param="prompt"
@@ -349,6 +352,14 @@ async def execute(
         except ModelNotFoundError as error:
             metrics.reject(route, 404)
             return error_response(404, error.code, str(error), param="model")
+        if messages is not None:
+            if request.stream or not registry.supports(request.model, "text_conversation"):
+                metrics.reject(route, 400)
+                return error_response(400, "unsupported_parameter", "this model requires nonstreaming native conversation support for history",
+                                      param="stream" if request.stream else "messages")
+            encoded = json.dumps(messages, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+            context.update(conversation_sha256=hashlib.sha256(encoded).hexdigest(),
+                           conversation_messages=len(messages), conversation_roles=[message["role"] for message in messages])
         max_tokens = requested_tokens or default_tokens
         if max_tokens > config.max_generation_tokens:
             metrics.reject(route, 400)
@@ -382,13 +393,17 @@ async def execute(
         context["replica"] = getattr(session, "replica_name", None)
         context["admission_ms"] = queue_seconds * 1000
         worker_config = generation_config(request, max_tokens)
-        worker_config["use_chat_template"] = chat
-        if system_prompt:
-            worker_config["system_prompt"] = system_prompt
+        if messages is None:
+            worker_config["use_chat_template"] = chat
+            if system_prompt:
+                worker_config["system_prompt"] = system_prompt
         context["generation_config"] = {key: value for key, value in worker_config.items() if key != "system_prompt"}
         context["has_system_prompt"] = bool(system_prompt)
         metrics.begin()
         inference_started = time.monotonic()
+        operation = "converse" if messages is not None else "generate"
+        context["native_operation"] = operation
+        payload = {"messages": messages, "config": worker_config} if messages is not None else {"prompt": prompt, "config": worker_config}
         if request.stream and registry.supports_streaming(request.model):
             return await incremental_response(session, {"prompt": prompt, "config": worker_config},
                 model=request.model, chat=chat, include_usage=(request.stream_options is not None
@@ -396,7 +411,7 @@ async def execute(
                 route=route, queue_seconds=queue_seconds, started=inference_started, metrics=metrics)
         try:
             result = await _worker_request(
-                session, {"prompt": prompt, "config": worker_config}
+                session, payload, operation=operation
             )
             text, completion_tokens, timings = extract_result(result)
         except asyncio.CancelledError:
@@ -863,8 +878,8 @@ def validate(value):
     @app.post("/v1/chat/completions")
     async def chat_completions(http_request: Request, request: ChatCompletionRequest) -> Any:
         try:
-            prompt, system_prompt = chat_prompt(request)
             image_url = chat_image(request)
+            prompt, system_prompt = chat_prompt(request)
         except ValueError as error:
             metrics.reject("/v1/chat/completions", 400)
             return error_response(400, "invalid_request", str(error), param="messages")
@@ -886,6 +901,7 @@ async def chat_completions(http_request: Request, request: ChatCompletionRequest
             system_prompt=system_prompt,
             chat=True,
             chat_max_tokens=request.max_completion_tokens,
+            messages=chat_history(request),
         )
 
     videos = VideoJobs(registry, config, records, execute_task, error_response)
diff --git a/apps/server/python/trtmc_server/protocol.py b/apps/server/python/trtmc_server/protocol.py
index 852f8f270b..5c145c4681 100644
--- a/apps/server/python/trtmc_server/protocol.py
+++ b/apps/server/python/trtmc_server/protocol.py
@@ -24,15 +24,19 @@ def generation_config(request: GenerationRequest, max_tokens: int) -> dict[str,
 
 def chat_prompt(request: ChatCompletionRequest) -> tuple[str, str]:
     messages = request.messages
-    if len(messages) == 1 and messages[0].role == "user":
-        return _message_text(messages[0].content), ""
-    if (
-        len(messages) == 2
-        and messages[0].role == "system"
-        and messages[1].role == "user"
+    if messages[-1].role != "user" or any(
+        message.role not in {"user", "assistant"} and not (index == 0 and message.role == "system")
+        for index, message in enumerate(messages)
     ):
-        return _message_text(messages[1].content), _message_text(messages[0].content)
-    raise ValueError("messages must be one user message with an optional preceding system message")
+        raise ValueError("messages require user/assistant roles, an optional leading system, and a final user")
+    return _message_text(messages[-1].content), (
+        _message_text(messages[0].content) if messages[0].role == "system" else "")
+
+
+def chat_history(request: ChatCompletionRequest) -> list[dict[str, str]] | None:
+    if len(request.messages) == 1 or (len(request.messages) == 2 and request.messages[0].role == "system"):
+        return None
+    return [{"role": message.role, "content": _message_text(message.content)} for message in request.messages]
 
 
 def chat_image(request: ChatCompletionRequest) -> str | None:
diff --git a/apps/server/python/trtmc_server/schemas.py b/apps/server/python/trtmc_server/schemas.py
index 94af0e491e..47b051b0c5 100644
--- a/apps/server/python/trtmc_server/schemas.py
+++ b/apps/server/python/trtmc_server/schemas.py
@@ -58,7 +58,7 @@ class CompletionRequest(GenerationRequest):
 
 
 class ChatCompletionRequest(GenerationRequest):
-    messages: list[ChatMessage] = Field(min_length=1, max_length=2)
+    messages: list[ChatMessage] = Field(min_length=1, max_length=64)
     max_completion_tokens: int | None = Field(default=None, ge=1)
 
 
diff --git a/apps/server/tests/test_sdk_worker.cpp b/apps/server/tests/test_sdk_worker.cpp
index 2cdfa52e9e..44dee9fa1c 100644
--- a/apps/server/tests/test_sdk_worker.cpp
+++ b/apps/server/tests/test_sdk_worker.cpp
@@ -213,7 +213,8 @@ void test_incremental_stream(const std::filesystem::path& root, const trtmc::Loa
             "stream emits one delta and one terminal then shutdown");
     require(run.messages[0].at("protocol_version") == 2 &&
                 run.messages[0].at("capabilities") ==
-                    Json::array({"text_generation", "streaming_text_continuation"}),
+                    Json::array(
+                        {"text_generation", "streaming_text_continuation", "text_conversation"}),
             "ready advertises actual SDK stream capability");
     require(run.messages[1].at("id") == "stream" && run.messages[1].at("event") == "delta" &&
                 run.messages[1].at("result").at("text_delta") == "hello!",
@@ -224,6 +225,23 @@ void test_incremental_stream(const std::filesystem::path& root, const trtmc::Loa
                     "public_task_stream_wall_including_backpressure" &&
                 run.messages[2].at("result").at("model_call_ms").get<double>() >= 0,
             "terminal result carries scoped timing");
+
+    const auto invalid_history =
+        invoke(bundle, options,
+               {{{"id", "bad-history"}, {"op", "converse"}, {"messages", Json::array()}},
+                {{"id", "bad-role"},
+                 {"op", "converse"},
+                 {"messages", Json::array({{{"role", "tool"}, {"content", "unsupported"}}})}},
+                {{"id", "after-history-errors"}, {"op", "generate"}, {"prompt", "hello"}}});
+    require(invalid_history.status == 0 && invalid_history.messages.size() == 4,
+            "invalid conversation requests preserve the worker");
+    for (std::size_t index = 1; index < 3; ++index)
+        require(invalid_history.messages[index].at("ok") == false &&
+                    invalid_history.messages[index].at("error").at("type") ==
+                        "invalid_request_error",
+                "bad histories are recoverable request errors");
+    require(invalid_history.messages.back().at("result").at("text") == "sync",
+            "text generation remains usable after conversation errors");
 }
 void test_audio_task(const std::filesystem::path& root, const trtmc::LoadOptions& options) {
     const auto bundle = write_bundle(root, "speech_transcription", "audio_fixture");
diff --git a/families/qwen/runtime/chat_templates.cpp b/families/qwen/runtime/chat_templates.cpp
index 14adaf80ff..8b88105c1d 100644
--- a/families/qwen/runtime/chat_templates.cpp
+++ b/families/qwen/runtime/chat_templates.cpp
@@ -5,6 +5,7 @@
 
 #include "families/qwen/runtime/chat_templates.h"
 
+#include <stdexcept>
 #include <string>
 
 namespace trtmc {
@@ -47,4 +48,38 @@ std::string qwen_apply_chat_template(const std::string& format, const std::strin
     return prompt;
 }
 
+std::string
+qwen_apply_conversation_template(const std::string& format,
+                                 const std::vector<std::pair<std::string, std::string>>& messages,
+                                 bool enable_thinking) {
+    if (format != "chatml" && format != "chatml_thinking")
+        throw std::invalid_argument("Qwen conversation requires a supported chat template");
+    if (messages.empty() || messages.back().first != "user")
+        throw std::invalid_argument("Qwen conversation must end with a user message");
+    std::string result;
+    for (std::size_t index = 0; index < messages.size(); ++index) {
+        const auto& [role, original] = messages[index];
+        if (role != "user" && role != "assistant" && !(role == "system" && index == 0))
+            throw std::invalid_argument(
+                "Qwen conversation supports user/assistant and a leading system message");
+        auto content = original;
+        // The pinned hybrid Qwen3 template omits reasoning from assistant turns
+        // before the final user query. Preserve the original typed input; only
+        // this family-owned rendering applies that model-specific convention.
+        if (role == "assistant" && format == "chatml_thinking") {
+            const auto end = content.rfind("</think>");
+            if (end != std::string::npos) {
+                content.erase(0, end + std::string("</think>").size());
+                const auto first = content.find_first_not_of('\n');
+                content.erase(0, first == std::string::npos ? content.size() : first);
+            }
+        }
+        result += "<|im_start|>" + role + "\n" + content + "<|im_end|>\n";
+    }
+    result += "<|im_start|>assistant\n";
+    if (format == "chatml_thinking" && !enable_thinking)
+        result += "<think>\n\n</think>\n\n";
+    return result;
+}
+
 } // namespace trtmc
diff --git a/families/qwen/runtime/chat_templates.h b/families/qwen/runtime/chat_templates.h
index 7e5c787afd..7ff188e8d2 100644
--- a/families/qwen/runtime/chat_templates.h
+++ b/families/qwen/runtime/chat_templates.h
@@ -6,6 +6,8 @@
 #pragma once
 
 #include <string>
+#include <utility>
+#include <vector>
 
 namespace trtmc {
 
@@ -13,5 +15,9 @@ std::string qwen_detect_chat_template_format(const std::string& jinja_template);
 std::string qwen_apply_chat_template(const std::string& format, const std::string& prompt,
                                      bool enable_thinking = true,
                                      const std::string& system_prompt = {});
+std::string
+qwen_apply_conversation_template(const std::string& format,
+                                 const std::vector<std::pair<std::string, std::string>>& messages,
+                                 bool enable_thinking = true);
 
 } // namespace trtmc
diff --git a/families/qwen/runtime/pipeline.cpp b/families/qwen/runtime/pipeline.cpp
index 7cbd70dfd4..1b0024bb20 100644
--- a/families/qwen/runtime/pipeline.cpp
+++ b/families/qwen/runtime/pipeline.cpp
@@ -139,8 +139,16 @@ const ConfigField qwen_text_fields[] = {
     {"enable_thinking", ConfigKind::Bool, true, "Enable Qwen thinking"},
     {"system_prompt", ConfigKind::String, std::string_view{}, "ChatML system message"},
 };
-TextGenerationConfig qwen_text_config(internal::ConfigView supplied) {
-    internal::validate_config(qwen_text_fields, supplied);
+const auto qwen_conversation_fields = [] {
+    std::vector<ConfigField> fields;
+    for (const auto& field : qwen_text_fields)
+        if (field.name != "use_chat_template" && field.name != "system_prompt")
+            fields.push_back(field);
+    return fields;
+}();
+TextGenerationConfig qwen_text_config(internal::ConfigView supplied,
+                                      Span<const ConfigField> fields = qwen_text_fields) {
+    internal::validate_config(fields, supplied);
     TextGenerationConfig result;
     const auto integer = [&](std::string_view name) {
         const auto value = *internal::config_get<std::int64_t>(supplied, qwen_text_fields, name);
@@ -180,12 +188,65 @@ std::vector<internal::TaskInstance> QwenTextGenerationPipeline::task_bindings()
     // move a tensor-parallel communicator onto a producer thread.
     std::vector<internal::TaskInstance> result{
         internal::bind<internal::ITextContinuation>(*this, qwen_text_fields)};
+    if (config_.chat_template_format == "chatml" ||
+        config_.chat_template_format == "chatml_thinking")
+        result.push_back(internal::bind<internal::ITextConversation>(
+            *this, {qwen_conversation_fields.data(), qwen_conversation_fields.size()}));
     if (!distributed_owner_)
         result.push_back(
             internal::bind<internal::IStreamingTextContinuation>(*this, qwen_text_fields));
     return result;
 }
 
+internal::ConversationResult
+QwenTextGenerationPipeline::run(const internal::TextConversationRequest& request,
+                                internal::ConfigView config) {
+    if (!request.tools.empty())
+        throw std::invalid_argument("Qwen conversation tools are unsupported");
+    const auto cfg = qwen_text_config(
+        config, {qwen_conversation_fields.data(), qwen_conversation_fields.size()});
+    std::vector<std::pair<std::string, std::string>> messages;
+    for (const auto& message : request.messages) {
+        std::string role;
+        switch (message.role) {
+        case internal::MessageRole::System:
+            role = "system";
+            break;
+        case internal::MessageRole::User:
+            role = "user";
+            break;
+        case internal::MessageRole::Assistant:
+            role = "assistant";
+            break;
+        default:
+            throw std::invalid_argument("Qwen conversation role is unsupported");
+        }
+        std::string text;
+        for (const auto& part : message.parts) {
+            const auto* value = std::get_if<internal::TextPartView>(&part);
+            if (value == nullptr)
+                throw std::invalid_argument("Qwen conversation requires text-only message parts");
+            text += value->text;
+        }
+        messages.emplace_back(std::move(role), std::move(text));
+    }
+    const auto rendered = qwen_apply_conversation_template(config_.chat_template_format, messages,
+                                                           cfg.enable_thinking);
+    try {
+        const auto generated = generate(rendered, cfg);
+        internal::ConversationResult result;
+        result.parts.emplace_back(internal::TextPart{generated.text});
+        result.token_ids = generated.token_ids;
+        result.setup_ms = generated.setup_ms;
+        result.prefill_ms = generated.prefill_ms;
+        result.decode_ms = generated.decode_ms;
+        result.usage.output_tokens = generated.token_ids.size();
+        return result;
+    } catch (const QwenGenerationCapacityError& error) {
+        throw std::invalid_argument(error.what());
+    }
+}
+
 TextResult QwenTextGenerationPipeline::run(const internal::TextContinuationRequest& request,
                                            internal::ConfigView config) {
     const auto cfg = qwen_text_config(config);
diff --git a/families/qwen/runtime/pipeline.h b/families/qwen/runtime/pipeline.h
index 7b5c0ba55a..cc726a21ad 100644
--- a/families/qwen/runtime/pipeline.h
+++ b/families/qwen/runtime/pipeline.h
@@ -14,6 +14,7 @@
 #include "families/qwen/runtime/inference_state.h"
 #include "families/qwen/runtime/sampler.h"
 #include "families/qwen/runtime/tokenizer.h"
+#include "trtmc/internal/language.h"
 #include "trtmc/internal/model.h"
 #include "trtmc/internal/stream.h"
 #include "trtmc/runtime/trt_module.h"
@@ -48,6 +49,7 @@ struct QwenTextGenConfig {
 class QwenTextGenerationPipeline final : public ITextGeneration,
                                          public internal::IModel,
                                          public internal::ITextContinuation,
+                                         public internal::ITextConversation,
                                          public internal::IStreamingTextContinuation {
   public:
     QwenTextGenerationPipeline(std::unique_ptr<ITrtModule> decoder,
@@ -61,6 +63,8 @@ class QwenTextGenerationPipeline final : public ITextGeneration,
     int32_t default_max_new_tokens() const override { return 128; }
     std::vector<internal::TaskInstance> task_bindings() override;
     TextResult run(const internal::TextContinuationRequest&, internal::ConfigView) override;
+    internal::ConversationResult run(const internal::TextConversationRequest&,
+                                     internal::ConfigView) override;
     std::unique_ptr<internal::ITextStream> start(const internal::TextContinuationRequest&,
                                                  internal::ConfigView) override;
 
diff --git a/families/qwen/tests/cpp/test_qwen_chat_templates.cpp b/families/qwen/tests/cpp/test_qwen_chat_templates.cpp
index c1d7d26e60..3f53a772cd 100644
--- a/families/qwen/tests/cpp/test_qwen_chat_templates.cpp
+++ b/families/qwen/tests/cpp/test_qwen_chat_templates.cpp
@@ -6,6 +6,7 @@
 #include "families/qwen/runtime/chat_templates.h"
 
 #include <cstdio>
+#include <stdexcept>
 #include <string>
 
 namespace {
@@ -62,7 +63,46 @@ int main() {
     check_render("{{ messages[0]['content'] }}", false, "Q",
                  "unsupported template preserves the prompt");
 
+    const std::vector<std::pair<std::string, std::string>> history{
+        {"system", "  system\n"},
+        {"user", "Remember ORCHID."},
+        {"assistant", "<think>\nreasoning\n</think>\n\nOK"},
+        {"user", "What word?"}};
+    const std::string history_prefix = "<|im_start|>system\n  system\n<|im_end|>\n"
+                                       "<|im_start|>user\nRemember ORCHID.<|im_end|>\n";
+    const std::string history_suffix =
+        "<|im_end|>\n<|im_start|>user\nWhat word?<|im_end|>\n<|im_start|>assistant\n";
+    if (trtmc::qwen_apply_conversation_template("chatml_thinking", history, false) !=
+        history_prefix + "<|im_start|>assistant\nOK" + history_suffix + empty_think) {
+        std::fprintf(
+            stderr,
+            "FAIL: hybrid conversation retains roles/history and removes prior reasoning\n");
+        ++failures;
+    }
+    if (trtmc::qwen_apply_conversation_template("chatml", history, false) !=
+        history_prefix + "<|im_start|>assistant\n<think>\nreasoning\n</think>\n\nOK" +
+            history_suffix) {
+        std::fprintf(stderr, "FAIL: ordinary ChatML preserves assistant content\n");
+        ++failures;
+    }
+    for (const auto& invalid : std::vector<std::vector<std::pair<std::string, std::string>>>{
+             {},
+             {{"assistant", "unfinished"}},
+             {{"tool", "unsupported"}, {"user", "Q"}},
+             {{"user", "Q"}, {"system", "misplaced"}, {"user", "Q"}}}) {
+        bool rejected = false;
+        try {
+            trtmc::qwen_apply_conversation_template("chatml_thinking", invalid, false);
+        } catch (const std::invalid_argument&) {
+            rejected = true;
+        }
+        if (!rejected) {
+            std::fprintf(stderr, "FAIL: invalid conversation accepted\n");
+            ++failures;
+        }
+    }
+
     if (failures == 0)
-        std::fprintf(stderr, "All 13 Qwen chat template cases passed.\n");
+        std::fprintf(stderr, "All Qwen single-turn and conversation template cases passed.\n");
     return failures;
 }
diff --git a/website/docs/user-guides/profile-text-with-aiperf.md b/website/docs/user-guides/profile-text-with-aiperf.md
index c11629123c..dd4347aca6 100644
--- a/website/docs/user-guides/profile-text-with-aiperf.md
+++ b/website/docs/user-guides/profile-text-with-aiperf.md
@@ -137,3 +137,30 @@ For a reproducible selection of real ShareGPT prompts, the optional
 provides a pinned Qwen3-0.6B workload. It preserves selected strings, records
 over-limit rejections, and verifies HTTP input/output against native dispatch
 records. It uses the same published AIPerf package.
+
+## Benchmark a conversation
+
+With the current Qwen server and runtime, save this as `conversations.jsonl`:
+
+```json
+{"session_id":"example","turns":[{"text":"Remember the code word ORCHID. Reply OK."},{"text":"What code word did I give you?","delay":50}]}
+```
+
+Use the benchmark command above with `--endpoint-type chat`, replace the two
+synthetic-input options with
+`--input-file conversations.jsonl --custom-dataset-type multi_turn`, and use
+`--extra-inputs temperature:0 enable_thinking:false`. Each turn is one request;
+`--request-count 2` measures one complete conversation. The delay is 50
+milliseconds after the first response. AIPerf includes the original user turn
+and the actual model answer in the follow-up request.
+
+Keep conversation requests nonstreaming. History requires the loaded model's
+native `TextConversation` capability; unsupported models return an explicit
+error. Qwen owns chat formatting and tokenization. Its existing bundles work
+with the updated family library; follow the existing build guide for the
+server/runtime installation. The complete history and requested output must
+fit the bundle's context limit; history is never silently dropped.
+
+For pinned ShareGPT conversations, per-turn timings, and automatic history
+verification, use the optional
+[conversation profile](https://github.com/NVIDIA/TensorRT-Model-Connect/blob/main/apps/aiperf_qual/README.md#pinned-conversations).
diff --git a/website/docs/user-guides/serve-text-generation.md b/website/docs/user-guides/serve-text-generation.md
index 56926a478f..8001476937 100644
--- a/website/docs/user-guides/serve-text-generation.md
+++ b/website/docs/user-guides/serve-text-generation.md
@@ -96,7 +96,7 @@ support that placement.
 | Route | Supported behavior |
 | --- | --- |
 | `POST /v1/completions` | One string `prompt`; `model` is required. |
-| `POST /v1/chat/completions` | One text `user` message and optional preceding `system` message; vision-language bundles also accept one user message with text and one embedded image. |
+| `POST /v1/chat/completions` | One text `user` message and optional preceding `system` message. Native `TextConversation` models also accept nonstreaming text history (up to 64 messages, final role `user`). Vision-language bundles accept one user message with text and one embedded image. |
 | `GET /v1/models` | Configured models and their native capabilities. |
 | `GET /health/live` | Control-plane process liveness. |
 | `GET /health/ready` | Worker readiness and admission state. |
@@ -107,9 +107,15 @@ Generation accepts `temperature`, `top_p`, `min_p`, `top_k`, `seed`,
 Completions accept `max_tokens`;
 chat accepts either `max_tokens` or `max_completion_tokens`. The MVP
 requires `n=1`. It rejects unknown fields, prompt arrays, unsupported media content,
-multi-turn chat, stop sequences, tools, and log probabilities instead
+unsupported history, stop sequences, tools, and log probabilities instead
 of silently ignoring them.
 
+For text history, the current Qwen family implements the existing
+`TextConversation` Task for its supported ChatML templates. Send user/assistant
+messages in their original order, optionally preceded by a system message,
+and end with a user message. History requests are nonstreaming and must fit
+the complete context limit; unsupported models return an explicit error.
+
 For single-image chat, see [Profile Image Understanding with AIPerf](./profile-vision-with-aiperf.md).
 Image requests require a vision-language bundle and accept one embedded RGB PNG/JPEG
 plus text. System prompts, streaming, and thinking controls are rejected for this workload.
@@ -160,6 +166,8 @@ composition, and validation remain inside the family selected by the bundle.
 For migrated bundles, the native worker calls the public header-only C++ Task
 wrapper over the stable C ABI. It uses `TextContinuation`, or the primary
 `ConditionalTextGeneration` / `TextTranslation` contract for those bundles.
+History calls the existing `TextConversation` contract only when the loaded
+model advertises it. The family formats and tokenizes the typed messages.
 Translation uses the family's bundled language defaults; this HTTP protocol
 does not add source/target-language parameters.