diff --git a/packages/ai/README.md b/packages/ai/README.md
index 355dcdce..829bb567 100644
--- a/packages/ai/README.md
+++ b/packages/ai/README.md
@@ -73,7 +73,7 @@ result = await evals.run(
 )
 ```
 
-`LD_API_TOKEN` is required. Configure `LD_SDK_KEY` — or initialize your own client with `init_client(client=...)` — to emit one `$ld:ai:offline-evals:generation` event per generated row, plus one `$ld:ai:offline-evals:criterion` event per `(row, criterion)` when `criteria` are supplied, through the standard SDK event transport. The SDK reports scores; LaunchDarkly rules on them at ingest. A judge served by a different provider than `generation` needs a handler for it in `judge_handlers`. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. Evaluation-run links use the explicit `ui_base_uri` option or `LD_UI_BASE_URI`, defaulting to `https://app.launchdarkly.com`; set it when the project is not in production, or a run created elsewhere still links to the production app. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code).
+`LD_API_TOKEN` is required. Configure `LD_SDK_KEY` — or initialize your own client with `init_client(client=...)` — to emit one `$ld:ai:offline-evals:generation` event per generated row, plus one `$ld:ai:offline-evals:criterion` event per `(row, criterion)` when `criteria` are supplied, through the standard SDK event transport. The SDK reports scores; LaunchDarkly rules on them at ingest. A judge served by a different provider than `generation` needs a handler for it in `judge_handlers`. Each row's tool calls are recorded during generation and rendered into the judge's `{{message_history}}`, between the row input and the generated output, so a rubric can grade the tool trajectory as well as the final answer. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. Evaluation-run links use the explicit `ui_base_uri` option or `LD_UI_BASE_URI`, defaulting to `https://app.launchdarkly.com`; set it when the project is not in production, or a run created elsewhere still links to the production app. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code).
 
 ---
 
diff --git a/packages/client/README.md b/packages/client/README.md
index 66a3924e..b2607d4e 100644
--- a/packages/client/README.md
+++ b/packages/client/README.md
@@ -116,6 +116,51 @@ result = await init_evaluations().run(
 
 **The SDK reports scores and never rules on them.** LaunchDarkly derives each row's verdict at ingest by comparing the score against the criterion's stored threshold and success direction, so pass/fail policy is one server-side implementation that applies to every SDK version and to runs already recorded. A judge's direction lives on its AI Config and is injected server-side, keeping the one input a verdict turns on server-attested; a `Scorer` has no LaunchDarkly-side config to read, so it declares its own `success_direction` (default `"higher_is_better"` — set `"lower_is_better"` for a scorer that counts something unwanted, like a regex hit count).
 
+#### Judge the tool trajectory
+
+A judge is shown the tool calls the row made on the way to its output, so a rubric can grade *how* the agent answered and not only *what* it answered — whether it called the right tool, in the right order, with the right arguments, and how it handled a tool that failed.
+
+The harness records this itself: it wraps your tool implementations once per row before handing them to the handler, so every handler package is covered without changes and your tools still return and raise exactly what they did before.
+
+**This is not specific to offline evaluations.** Online judges — both the inline ones sampled by `config().invoke()` and the deferred ones you run from a `JudgeTask` on a background thread — are shown the same trajectory, built by the same function. See [Judges see one conversation](#judges-see-one-conversation).
+
+The trajectory is rendered into **`{{message_history}}`** — the row input, then the trajectory, then the generated output, then the formatting instructions, in that order. There is no separate trajectory variable: `message_history` is already the transcript variable every judge reads, and judges built from the AI Library's default templates reference it, so a trajectory rubric can be written against an existing judge template with no new placeholder.
+
+```
+Tools available: lookup_order, issue_refund
+Tool calls made while producing the response, in order:
+1. lookup_order
+   arguments: {"id":"A1"}
+   result: order A1 shipped 2026-08-02
+2. issue_refund
+   arguments: {"id":"A1","amount":19.99}
+   error: refund window closed
+```
+
+A row with tools that called none of them says so explicitly, which is the finding a tool-selection rubric most needs. A run with no tools adds no block at all, so judges written before trajectories existed read exactly the history they read before.
+
+#### Judges see one conversation
+
+All three judge paths build `{{message_history}}` through a single function, `judge_scoring.build_message_history`:
+
+| Path | Entry point |
+| --- | --- |
+| Online, inline | `config().invoke()` → `run_judges` |
+| Online, deferred | `config(skip_judges=True).invoke()` → `run_judge(task, handlers)` on your own thread |
+| Offline | `init_evaluations().run(criteria=[Judge(...)])` |
+
+Each one is the input, then the tool trajectory, then the output, then the `{score, reasoning}` format block, with empty parts skipped. A judge therefore grades the same conversation wherever it runs, which is what makes a rubric portable between a production sample and a dataset replay.
+
+They did not always agree, and that is why this is a single function now: each path used to join its own history. The offline one carried the row input, the inline one carried the user input, and the deferred one carried **neither** — so a deferred judge graded a response with no request beside it. `JudgeTask` gained `user_input` and `trajectory` to close that.
+
+For the deferred path those two fields travel on the task, which stays picklable — the trajectory crosses as the rendered string, not the structured record.
+
+A **graph-level** judge (`graph_judge`) gets no trajectory: it grades a final answer produced across several nodes, and splicing their trajectories together would describe a conversation that never happened. Per-node judges inside a graph do get their own node's.
+
+Two limits keep a trajectory from spending the judge's context window: at most 50 recorded calls per row and 2000 characters per rendered argument bag or result, with anything beyond either reported as a count or marked truncated. Calls past the limit still execute — truncation drops the record, never the work. A `NativeTool` runs inside the provider, so no local wrapper sees it; such a tool is left out of the trajectory and out of the "Tools available" line, since naming a tool whose use cannot be shown would invite a judge to conclude the model ignored it.
+
+A tool result is now judge-prompt input. It stays literal for the same reason the generated output does: the judge config is handed to the handler unrendered and the handler makes exactly one template pass, so a `{{...}}` sequence coming back from a tool is never expanded into the judge prompt.
+
 **Judges are independent AI Configs, so handlers are routed per judge.** A judge may resolve to a different provider or mode than `generation`, and a handler built for one provider cannot execute another's config. `handler` runs a judge when it provides for that judge's provider; pass handlers for any other providers in `judge_handlers`. Selection prefers a handler naming the judge's provider outright over a wildcard multi-provider adapter, and an agent-mode handler can serve a messages-mode judge with its messages collapsed into one instructions block. A plain callable that declares no `provides_for` routes itself, exactly as it already does for the generation config.
 
 Judges are resolved through flag delivery, and handlers are matched to them, **before** any evaluation records are created — a missing judge or one no handler covers fails the run up front rather than after the generation spend. After that point a criterion failure never aborts the run: an unparseable judge response, an out-of-range score, a raising handler or scorer, and a row whose generation errored each become a per-criterion `ERROR` event with a cause code (`invalid_judge_output`, `invalid_score`, `handler_raised`, `scorer_raised`, `generation_incomplete`) and a top-level `errorMessage`. Event *delivery* is different: the backend needs one result per `(row, criterion)` to finish row accounting, so if tracking a criterion event fails, every remaining result is still attempted and flushed and then `run()` raises — rather than polling to its timeout with the cause hidden.
diff --git a/packages/client/src/launchdarkly_ai_server/client.py b/packages/client/src/launchdarkly_ai_server/client.py
index f64422db..f9f598f2 100644
--- a/packages/client/src/launchdarkly_ai_server/client.py
+++ b/packages/client/src/launchdarkly_ai_server/client.py
@@ -123,6 +123,8 @@ async def invoke(
                 handlers=resolved_handler_list,
                 llm_response=llm_str,
                 base_track_data=track_data,
+                user_input=user_input,
+                trajectory=result.get("trajectory", ""),
             )
             return ProviderResponse(
                 response=parsed_response,
@@ -137,6 +139,7 @@ async def invoke(
             handler=handler,
             handlers=resolved_handler_list,
             user_input=user_input,
+            trajectory=result.get("trajectory", ""),
             llm_response=llm_str,
             base_track_data=track_data,
             tool_handlers=resolved_tools,
@@ -215,6 +218,7 @@ async def _stream_events(
                     handler=handler,
                     handlers=resolved_handler_list,
                     user_input=user_input,
+                    trajectory=done_event.get("trajectory", ""),
                     llm_response=done_event.get("response", ""),
                     base_track_data=track_data,
                     tool_handlers=resolved_tools,
diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py
index 164ea57d..07acdb9a 100644
--- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py
+++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py
@@ -14,10 +14,16 @@
 
 from ..judge_scoring import (
     FORMATTING_INSTRUCTIONS,
+    build_message_history,
     numeric_score,
     parse_judge_response,
 )
 from ..lifecycle import extract_variation
+from ..trajectory import (
+    TrajectoryRecorder,
+    render_row_trajectory,
+    row_fields,
+)
 from ..types import NativeTool
 from ..utils import (
     collapse_messages_to_instructions,
@@ -544,11 +550,15 @@ async def _run_rows(
 
         async def invoke(row: DatasetRow) -> dict[str, Any]:
             await controller.acquire(config["provider"]["name"])
+            # One per row, not per run: rows generate concurrently against the
+            # same tool map, so a shared recorder would splice their calls.
+            recorder = TrajectoryRecorder()
+            row_tool_handlers = recorder.wrap(tool_handlers)
             started = datetime.now(UTC)
             started_clock = time.perf_counter()
             try:
                 result = await handler(
-                    config, row.input, tool_handlers, dict(row.variables)
+                    config, row.input, row_tool_handlers, dict(row.variables)
                 )
                 if not isinstance(result, Mapping):
                     raise TypeError("handler result must be a mapping")
@@ -564,6 +574,7 @@ async def invoke(row: DatasetRow) -> dict[str, Any]:
                     "generated_at": completed.isoformat().replace("+00:00", "Z"),
                     "latency_ms": round((time.perf_counter() - started_clock) * 1000),
                     "status": "COMPLETE",
+                    **row_fields(recorder),
                 }
                 usage = result.get("usage")
                 if isinstance(usage, Mapping):
@@ -583,6 +594,8 @@ async def invoke(row: DatasetRow) -> dict[str, Any]:
                     "latency_ms": round((time.perf_counter() - started_clock) * 1000),
                     "status": "ERROR",
                     "error": {"code": 5001, "message": f"handler raised: {error}"},
+                    # The calls that ran are what explain why it raised.
+                    **row_fields(recorder),
                 }
             finally:
                 controller.release()
@@ -696,25 +709,23 @@ def _judge_variables(
             ground_truth = parse_template(ground_truth, variables)
         elif expected is not None:
             ground_truth = str(expected)
-        # message_history carries FORMATTING_INSTRUCTIONS the same way the
-        # online path builds it (judges.run_judges), because that -- not the
-        # standalone formatting_instructions variable below -- is what every
-        # judge built from the AI Library's default templates (accuracy,
-        # relevance, toxicity, and any judge cloned from them) actually
-        # references. A judge authored before this variable existed must keep
-        # getting scored without edits.
+        # The calls the row made on its way to `output`, recorded during
+        # generation. Sits between input and output in message_history, which
+        # is where it happened.
+        trajectory = render_row_trajectory(row_result)
+        # Shared builder, not an inline join: this path and both online paths
+        # must show a judge the same conversation. The trajectory goes into
+        # message_history and nowhere else -- it is already the transcript
+        # variable judges read, and a second one would just let a rubric
+        # interpolate both and pay for the trajectory twice.
         variables.update(
             {
                 "input": row_result.get("input") or "",
                 "response_to_evaluate": output if output is not None else "",
-                "message_history": "\n\n".join(
-                    str(value)
-                    for value in (
-                        row_result.get("input"),
-                        output,
-                        FORMATTING_INSTRUCTIONS,
-                    )
-                    if value
+                "message_history": build_message_history(
+                    user_input=row_result.get("input"),
+                    trajectory=trajectory,
+                    output=output,
                 ),
                 "expected_output": expected if expected is not None else "",
                 "ground_truth_context": (
diff --git a/packages/client/src/launchdarkly_ai_server/graph.py b/packages/client/src/launchdarkly_ai_server/graph.py
index 4546c3b8..ace169dd 100644
--- a/packages/client/src/launchdarkly_ai_server/graph.py
+++ b/packages/client/src/launchdarkly_ai_server/graph.py
@@ -226,6 +226,7 @@ async def run_node(
                 base_track_data=result["track_data"],
                 tool_handlers=tool_handlers,
                 graph_key=key,
+                trajectory=result.get("trajectory", ""),
             )
 
             if from_node:
@@ -380,6 +381,7 @@ def _fn(*a: Any, **kw: Any) -> str:
                 base_track_data=result["track_data"],
                 tool_handlers=tool_handlers,
                 graph_key=key,
+                trajectory=result.get("trajectory", ""),
             )
 
             next_node = nodes.get(chosen[0]) if chosen else None
diff --git a/packages/client/src/launchdarkly_ai_server/judge_scoring.py b/packages/client/src/launchdarkly_ai_server/judge_scoring.py
index 8375b413..99fc3a05 100644
--- a/packages/client/src/launchdarkly_ai_server/judge_scoring.py
+++ b/packages/client/src/launchdarkly_ai_server/judge_scoring.py
@@ -1,10 +1,12 @@
-"""Shared scoring contract for LaunchDarkly AI Judge invocations.
+"""Shared contract for LaunchDarkly AI Judge invocations.
 
-Both judge execution paths — the online path (``judges.run_judges``, sampled
-per invocation) and the offline evaluations path (``evaluations.runner``) —
-prompt a judge model for the same ``{"score": <0-1>, "reasoning": <string>}``
-JSON shape and must parse it the same way. This module owns that contract so
-the two paths cannot drift.
+Three judge execution paths exist — the online inline path
+(``judges.run_judges``, sampled per invocation), the online deferred path
+(``judges.run_judge``, from a ``JudgeTask`` on a background thread), and the
+offline evaluations path (``evaluations.runner``). All three prompt a judge
+model for the same ``{"score": <0-1>, "reasoning": <string>}`` JSON shape, and
+all three must show the judge the same conversation. This module owns both
+halves of that contract so the paths cannot drift.
 """
 
 from __future__ import annotations
@@ -27,6 +29,30 @@
 )
 
 
+def build_message_history(
+    *,
+    user_input: Any = None,
+    trajectory: Any = None,
+    output: Any = None,
+) -> str:
+    """The conversation a judge is shown, as its ``message_history`` variable.
+
+    Ordered as it happened: what was asked, what the agent did, what it
+    answered, then how to format the verdict. Empty parts are skipped, so a
+    run with no tools yields the history it did before trajectories existed.
+
+    The formatting block is appended here, not by callers: judges built from
+    the AI Library's default templates read the JSON shape from
+    ``{{message_history}}``, and one that stopped being told it would return
+    prose and fail every result as invalid output.
+    """
+    return "\n\n".join(
+        str(part)
+        for part in (user_input, trajectory, output, FORMATTING_INSTRUCTIONS)
+        if part
+    )
+
+
 def numeric_score(score: Any) -> float | None:
     """Return ``score`` as a float only when it already is a finite number.
 
diff --git a/packages/client/src/launchdarkly_ai_server/judges.py b/packages/client/src/launchdarkly_ai_server/judges.py
index 126cc700..56d7e032 100644
--- a/packages/client/src/launchdarkly_ai_server/judges.py
+++ b/packages/client/src/launchdarkly_ai_server/judges.py
@@ -7,7 +7,7 @@
 
 from .conversation import with_judge_evaluation
 from .judge_scoring import (
-    FORMATTING_INSTRUCTIONS,
+    build_message_history,
     numeric_score,
     parse_judge_response,
 )
@@ -54,10 +54,16 @@ async def run_judges(
     base_track_data: TrackData,
     tool_handlers: dict[str, Callable[..., Any] | NativeTool] | None = None,
     graph_key: str | None = None,
+    trajectory: str = "",
 ) -> dict[str, JudgeResult]:
     """
     Runs any judges configured on ``config['judgeConfiguration']`` against the
     produced output. Each judge is itself a tracked AI call.
+
+    ``trajectory`` is the rendered tool-call trajectory of the invocation being
+    judged, from ``execute_and_track``. It defaults to empty so a caller that
+    has none -- a graph-level judge over several nodes, for instance -- is
+    unchanged, and so is a judge for a config with no tools.
     """
     from .lifecycle import extract_variation
     from .tracking import execute_and_track
@@ -146,8 +152,10 @@ async def run_judges(
                 else judge_ai_config
             )
 
-            message_history = "\n\n".join(
-                filter(None, [user_input, llm_response, FORMATTING_INSTRUCTIONS])
+            message_history = build_message_history(
+                user_input=user_input,
+                trajectory=trajectory,
+                output=llm_response,
             )
 
             async with with_judge_evaluation(judge_key) as record_evaluation:
@@ -209,6 +217,8 @@ async def build_judge_tasks(
     handlers: list[ProviderHandler] | None = None,
     llm_response: str,
     base_track_data: TrackData,
+    user_input: str | None = None,
+    trajectory: str = "",
 ) -> list[JudgeTask]:
     """
     Resolves all judges configured on ``config['judgeConfiguration']`` into
@@ -307,6 +317,8 @@ async def build_judge_tasks(
                     judge_config=judge_ai_config,
                     judge_meta=judge_meta,
                     actual_output=llm_response,
+                    user_input=user_input,
+                    trajectory=trajectory,
                     user_context=user_context,
                     judge_provider=judge_provider,
                     judge_mode=judge_mode,
@@ -375,8 +387,13 @@ def _matches(h: ProviderHandler) -> bool:
         else task.judge_config
     )
 
-    message_history = "\n\n".join(
-        filter(None, [task.actual_output, FORMATTING_INSTRUCTIONS])
+    # user_input and trajectory come off the task rather than being omitted:
+    # this path used to build a history with neither, so a judge grading the
+    # same response saw a different conversation than the inline path did.
+    message_history = build_message_history(
+        user_input=task.user_input,
+        trajectory=task.trajectory,
+        output=task.actual_output,
     )
 
     async with with_judge_evaluation(task.config_key) as record_evaluation:
diff --git a/packages/client/src/launchdarkly_ai_server/tracking.py b/packages/client/src/launchdarkly_ai_server/tracking.py
index 9beae0ba..9e0235c5 100644
--- a/packages/client/src/launchdarkly_ai_server/tracking.py
+++ b/packages/client/src/launchdarkly_ai_server/tracking.py
@@ -7,6 +7,7 @@
 from collections.abc import AsyncGenerator, Callable
 from typing import Any
 
+from .trajectory import HANDOFF_TOOL_PREFIX, TrajectoryRecorder, render_trajectory
 from .types import (
     NATIVE_TOOL_KEY,
     AiConfigRep,
@@ -81,7 +82,7 @@ def _make_regular_wrapper(
                 tool_name: str, original: Callable[..., Any]
             ) -> Callable[..., Any]:
                 async def wrapper(*args: Any, **kwargs: Any) -> Any:
-                    if not tool_name.startswith("__handoff_"):
+                    if not tool_name.startswith(HANDOFF_TOOL_PREFIX):
                         get_client().track(
                             "$ld:ai:tool_call",
                             user_context,
@@ -98,6 +99,16 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
     return wrapped
 
 
+def _exposed_tool_keys(config: AiConfigRep) -> set[str]:
+    """The tool keys this config offered the model.
+
+    The map handed to a handler can be wider -- ``config()`` merges a
+    ``Registry``'s tools in -- and only the offered set belongs in a trajectory.
+    """
+    tools = config.get("tools") if isinstance(config, dict) else None
+    return set(tools) if isinstance(tools, dict) else set()
+
+
 async def execute_and_track(
     *,
     config_key: str,
@@ -139,7 +150,16 @@ async def execute_and_track(
     client = get_client()
     ld_ctx = to_ld_context(client, user_context)
 
-    tracked_tool_handlers = wrap_tool_handlers(tool_handlers, ld_ctx, track_data)
+    # Recording composes *inside* wrap_tool_handlers, on the original map, so
+    # the recorder still sees a NativeTool and skips it -- wrapping the tracked
+    # map would record the callable stub instead and show a judge a tool call
+    # with an empty result. One recorder per invocation; they run concurrently.
+    recorder = TrajectoryRecorder()
+    tracked_tool_handlers = wrap_tool_handlers(
+        recorder.wrap(tool_handlers or {}, exposed=_exposed_tool_keys(config)),
+        ld_ctx,
+        track_data,
+    )
     merged_variables: dict[str, Any] = {
         **(variables or {}),
         "ldContext": {**user_context},
@@ -173,7 +193,18 @@ async def execute_and_track(
 
     raw_output = result.get("output")
     response = raw_output if raw_output is not None else ""
-    return {"usage": usage, "response": response, "track_data": track_data}
+    return {
+        "usage": usage,
+        "response": response,
+        "track_data": track_data,
+        # Rendered, not structural: message_history is the only consumer, and
+        # a JudgeTask must stay picklable.
+        "trajectory": render_trajectory(
+            recorder.invocations,
+            observable_tools=recorder.observable_tools,
+            omitted=recorder.omitted,
+        ),
+    }
 
 
 async def execute_and_stream(
@@ -221,7 +252,12 @@ async def execute_and_stream(
     client = get_client()
     ld_ctx = to_ld_context(client, user_context)
 
-    tracked_tool_handlers = wrap_tool_handlers(tool_handlers, ld_ctx, track_data)
+    recorder = TrajectoryRecorder()
+    tracked_tool_handlers = wrap_tool_handlers(
+        recorder.wrap(tool_handlers or {}, exposed=_exposed_tool_keys(config)),
+        ld_ctx,
+        track_data,
+    )
     merged_variables: dict[str, Any] = {
         **(variables or {}),
         "ldContext": {**user_context},
@@ -277,4 +313,9 @@ async def execute_and_stream(
         "response": full_text,
         "usage": usage,
         "track_data": track_data,
+        "trajectory": render_trajectory(
+            recorder.invocations,
+            observable_tools=recorder.observable_tools,
+            omitted=recorder.omitted,
+        ),
     }
diff --git a/packages/client/src/launchdarkly_ai_server/trajectory.py b/packages/client/src/launchdarkly_ai_server/trajectory.py
new file mode 100644
index 00000000..bf9dfe21
--- /dev/null
+++ b/packages/client/src/launchdarkly_ai_server/trajectory.py
@@ -0,0 +1,288 @@
+"""Tool-call trajectory capture, shared by both judge paths.
+
+Handlers return only ``{output, usage}`` and record tool traffic onto spans, so
+the calls made on the way to an output were unavailable to a judge. Both paths
+therefore record them here, by wrapping the caller's tool implementations
+before the handler is invoked -- which works with every handler package without
+changing any of them, since a handler still looks a tool up by key and calls
+it.
+
+The trajectory reaches a judge through ``message_history``, built by
+:func:`judge_scoring.build_message_history` so both paths show the same shape.
+
+Two rules the rest of this module exists to keep: the recorder never changes
+what a tool does or how it is called, and a recorder belongs to exactly one
+invocation or row, since both run concurrently against one shared tool map.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import json
+from collections.abc import Callable, Collection, Mapping
+from dataclasses import dataclass, replace
+from typing import Any
+
+from .types import NativeTool
+
+#: A trajectory goes into a judge prompt, so an agent looping over a large
+#: result set would otherwise spend the judge's context window on a tail it
+#: never reads. Calls past the limit still execute; they are only counted.
+MAX_RECORDED_TOOL_CALLS = 50
+
+#: Bounds a single tool that returns a whole document, for the same reason.
+MAX_RECORDED_VALUE_CHARS = 2000
+
+_TRUNCATION_SUFFIX = "… (truncated)"
+
+#: Synthetic routing tools ``graph.route`` injects on a multi-edge node. Not
+#: tools the agent was given, and absent from the config a per-node judge is
+#: scored against -- showing them would let a judge grade a handoff as tool
+#: use. ``wrap_tool_handlers`` skips them too.
+HANDOFF_TOOL_PREFIX = "__handoff_"
+
+ToolImplementation = Callable[..., Any] | NativeTool
+
+
+@dataclass(frozen=True)
+class ToolInvocation:
+    """One tool call, with how it turned out.
+
+    ``result`` and ``error`` are mutually exclusive; both are ``None`` while
+    the call is still in flight.
+    """
+
+    name: str
+    arguments: Any = None
+    result: Any = None
+    error: str | None = None
+
+
+class TrajectoryRecorder:
+    """Records one unit's tool calls, in the order the calls were started.
+
+    A slot is reserved on call and filled in on completion, so concurrent tool
+    calls keep their start order instead of their return order.
+    """
+
+    def __init__(self, limit: int = MAX_RECORDED_TOOL_CALLS) -> None:
+        self._limit = limit
+        self._invocations: list[ToolInvocation] = []
+        self._omitted = 0
+        self._observable: list[str] = []
+
+    @property
+    def invocations(self) -> list[ToolInvocation]:
+        """The recorded calls, oldest first."""
+        return list(self._invocations)
+
+    @property
+    def omitted(self) -> int:
+        """How many calls executed past the recording limit."""
+        return self._omitted
+
+    @property
+    def observable_tools(self) -> list[str]:
+        """Keys of the tools this recorder can observe being called."""
+        return list(self._observable)
+
+    def wrap(
+        self,
+        tool_handlers: Mapping[str, ToolImplementation],
+        *,
+        exposed: Collection[str] | None = None,
+    ) -> dict[str, ToolImplementation]:
+        """Return ``tool_handlers`` with each callable recording into this unit.
+
+        Keys are preserved exactly; a handler resolves a tool by the key the
+        model named.
+
+        ``exposed`` is the tool keys the config actually offered the model.
+        Pass it when the implementation map can be wider -- online,
+        ``config()`` merges a ``Registry``'s tools in, and describing those
+        would let a judge penalise an agent for ignoring a tool it never had.
+        ``None`` means every callable is in scope, which is the offline case:
+        the runner resolves the config's tools from this same map.
+        """
+        wrapped: dict[str, ToolImplementation] = {}
+        # Rebuilt, so re-wrapping a map cannot list a tool twice.
+        self._observable = []
+        for name, implementation in tool_handlers.items():
+            if (
+                isinstance(implementation, NativeTool)
+                or not callable(implementation)
+                or name.startswith(HANDOFF_TOOL_PREFIX)
+                or (exposed is not None and name not in exposed)
+            ):
+                # Provider-executed, invalid, synthetic routing, or never
+                # offered to the model: nothing to observe, or nothing to grade
+                # the agent on. Passed through untouched so the handler treats
+                # it exactly as it would have.
+                wrapped[name] = implementation
+                continue
+            self._observable.append(name)
+            wrapped[name] = self._record(name, implementation)
+        return wrapped
+
+    def _record(self, name: str, original: Callable[..., Any]) -> Callable[..., Any]:
+        """Wrap ``original`` without changing how it is called.
+
+        A sync tool stays sync: offline these are passed straight to the
+        handler, and a caller's own handler may call a sync tool directly and
+        use the value. A blanket async wrapper handed it a coroutine object
+        instead. (Online, ``wrap_tool_handlers`` wraps this again and is always
+        async, so a handler awaits there as it always has.)
+        """
+        if asyncio.iscoroutinefunction(original):
+
+            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
+                slot = self._reserve(name, _call_arguments(args, kwargs))
+                return await self._await_and_record(slot, original(*args, **kwargs))
+
+            return async_wrapper
+
+        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
+            slot = self._reserve(name, _call_arguments(args, kwargs))
+            try:
+                result = original(*args, **kwargs)
+            except Exception as error:
+                self._complete(slot, error=f"{error}")
+                raise
+            if inspect.isawaitable(result):
+                # Record on completion, so a pending coroutine's repr never
+                # reaches a judge. Never awaited means never recorded, which is
+                # accurate: the call did not complete.
+                return self._await_and_record(slot, result)
+            self._complete(slot, result=result)
+            return result
+
+        return sync_wrapper
+
+    async def _await_and_record(self, slot: int | None, awaitable: Any) -> Any:
+        try:
+            result = await awaitable
+        except Exception as error:
+            self._complete(slot, error=f"{error}")
+            raise
+        self._complete(slot, result=result)
+        return result
+
+    def _reserve(self, name: str, arguments: Any) -> int | None:
+        if len(self._invocations) >= self._limit:
+            self._omitted += 1
+            return None
+        self._invocations.append(ToolInvocation(name=name, arguments=arguments))
+        return len(self._invocations) - 1
+
+    def _complete(
+        self, slot: int | None, *, result: Any = None, error: str | None = None
+    ) -> None:
+        if slot is None:
+            return
+        self._invocations[slot] = replace(
+            self._invocations[slot], result=result, error=error
+        )
+
+
+def row_fields(recorder: TrajectoryRecorder) -> dict[str, Any]:
+    """The trajectory keys a generated-row record carries.
+
+    Paired with :func:`render_row_trajectory` so one module owns both halves:
+    renaming a key without its reader would render every trajectory empty,
+    which reads exactly like an agent that called no tools.
+    """
+    return {
+        "tool_calls": recorder.invocations,
+        "tool_calls_omitted": recorder.omitted,
+        "observable_tools": recorder.observable_tools,
+    }
+
+
+def render_row_trajectory(row_result: Mapping[str, Any]) -> str:
+    """Render the trajectory carried by a generated-row record."""
+    return render_trajectory(
+        row_result.get("tool_calls") or [],
+        observable_tools=row_result.get("observable_tools") or [],
+        omitted=int(row_result.get("tool_calls_omitted") or 0),
+    )
+
+
+def render_trajectory(
+    invocations: list[ToolInvocation],
+    *,
+    observable_tools: list[str] | None = None,
+    omitted: int = 0,
+) -> str:
+    """Render a trajectory as the text a judge reads.
+
+    ``""`` when there was nothing observable, so the caller adds no block at
+    all. A unit that *had* tools and called none says so explicitly instead:
+    "called nothing" is the finding a tool-selection judge most needs, and an
+    omitted block would read as a unit with no tools.
+    """
+    available = list(observable_tools or [])
+    if not available and not invocations:
+        return ""
+
+    lines: list[str] = []
+    if available:
+        lines.append(f"Tools available: {', '.join(available)}")
+    if not invocations:
+        lines.append("No tool calls were made while producing the response.")
+        return "\n".join(lines)
+
+    lines.append("Tool calls made while producing the response, in order:")
+    for position, invocation in enumerate(invocations, start=1):
+        lines.append(f"{position}. {invocation.name}")
+        lines.append(f"   arguments: {_render_value(invocation.arguments)}")
+        if invocation.error is not None:
+            lines.append(f"   error: {_render_value(invocation.error)}")
+        else:
+            lines.append(f"   result: {_render_value(invocation.result)}")
+    if omitted > 0:
+        lines.append(f"({omitted} further tool call(s) were made but not recorded.)")
+    return "\n".join(lines)
+
+
+def _call_arguments(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any:
+    """Normalize how a handler passed a tool its arguments.
+
+    Every handler here calls a tool with the model's argument bag as one
+    positional mapping, so that shape is preserved verbatim; anything else is
+    recorded structurally rather than guessed at.
+    """
+    if len(args) == 1 and not kwargs:
+        return args[0]
+    if kwargs and not args:
+        return dict(kwargs)
+    if not args and not kwargs:
+        return None
+    return {"args": list(args), "kwargs": dict(kwargs)}
+
+
+def _render_value(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return _truncate(value)
+    try:
+        # ensure_ascii=False: a model reads this. The default would show the
+        # judge "caf\u00e9" instead of "café". Keys stay sorted for
+        # deterministic output.
+        rendered = json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+            ensure_ascii=False,
+        )
+    except (TypeError, ValueError):
+        rendered = str(value)
+    return _truncate(rendered)
+
+
+def _truncate(text: str) -> str:
+    if len(text) <= MAX_RECORDED_VALUE_CHARS:
+        return text
+    return text[:MAX_RECORDED_VALUE_CHARS] + _TRUNCATION_SUFFIX
diff --git a/packages/client/src/launchdarkly_ai_server/types.py b/packages/client/src/launchdarkly_ai_server/types.py
index dcbc5701..cab356e7 100644
--- a/packages/client/src/launchdarkly_ai_server/types.py
+++ b/packages/client/src/launchdarkly_ai_server/types.py
@@ -301,6 +301,19 @@ class JudgeTask:
     variables: dict[str, Any] | None = None
     """Optional extra template variables for the judge prompt."""
     evaluation_metric_key: str | None = None
+    user_input: str | None = None
+    """The input that produced ``actual_output``.
+
+    Carried so this path builds the same ``message_history`` as the inline one.
+    Previously absent, which showed a deferred judge a response with no
+    request beside it.
+    """
+    trajectory: str = ""
+    """The invocation's rendered tool-call trajectory.
+
+    A plain string rather than the structured record, since every field here
+    must stay picklable.
+    """
     """LD metric key to track the score against."""
 
 
diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py
index c7ca278b..08706a59 100644
--- a/packages/client/tests/test_evaluations_run.py
+++ b/packages/client/tests/test_evaluations_run.py
@@ -310,6 +310,17 @@ async def test_complete_run_with_zero_failed_and_error_rows_passes(
     assert event["emittedAt"].endswith("Z")
     assert datetime.fromisoformat(event["emittedAt"]).tzinfo is not None
     assert {"input", "expected_output", "metadata", "variables"}.isdisjoint(event)
+    # The captured tool trajectory reaches LaunchDarkly only inside the prompt a
+    # judge was shown, never as a generation wire field the backend has not
+    # specified.
+    assert {
+        "toolCalls",
+        "tool_calls",
+        "toolTrajectory",
+        "tool_trajectory",
+        "observableTools",
+        "observable_tools",
+    }.isdisjoint(event)
     emit_logs = [
         record.getMessage()
         for record in caplog.records
@@ -2249,3 +2260,331 @@ async def handler(
 
     assert result.passed is True
     assert max_in_flight == 2
+
+
+def tool_run_transport(*, rows: int = 1) -> SequencedTransport:
+    """Transport for a run that resolves one tool before its dataset."""
+    return SequencedTransport(
+        [
+            response(200, {"key": "lookup_order", "version": 4, "schema": {}}),
+            response(200, {"id": "dataset-id", "name": "golden"}),
+            response(
+                200,
+                dataset_page(
+                    [
+                        {
+                            "rowIndex": index,
+                            "input": f"Question {index}",
+                            "expectedOutput": "Answer",
+                        }
+                        for index in range(rows)
+                    ],
+                    total=rows,
+                ),
+            ),
+            response(201, {"id": "evaluation-id", "name": "support-qa", "version": 3}),
+            response(
+                201,
+                {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"},
+            ),
+            response(
+                200,
+                {
+                    "statusCounts": {
+                        "total": rows,
+                        "passed": rows,
+                        "error": 0,
+                        "pending": 0,
+                    }
+                },
+            ),
+        ]
+    )
+
+
+@pytest.mark.asyncio
+async def test_tool_trajectory_reaches_the_judge_via_message_history(
+    monkeypatch: pytest.MonkeyPatch,
+    stub_sdk_client: MagicMock,
+) -> None:
+    """The calls a row made are what let a judge grade its tool use.
+
+    Handler packages return only {output, usage}, so without the runner
+    recording the trajectory itself a judge sees the answer and nothing about
+    how the agent arrived at it.
+    """
+    transport = tool_run_transport()
+    accuracy_judge_variation(monkeypatch)
+    evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport)
+    seen: dict[str, str] = {}
+
+    def lookup_order(args: dict[str, Any]) -> str:
+        return f"order {args['id']} shipped"
+
+    async def handler(
+        config: dict[str, Any],
+        user_input: str | None,
+        tool_handlers: dict[str, Callable[..., Any]],
+        variables: dict[str, Any],
+    ) -> dict[str, Any]:
+        if "Judge" in config.get("instructions", ""):
+            seen["message_history"] = variables["message_history"]
+            # The trajectory lives in message_history and nowhere else: this is
+            # already the transcript variable every judge reads, so a second
+            # overlapping variable only invited a rubric to pay for the
+            # trajectory twice.
+            assert "tool_trajectory" not in variables
+            return {"output": '{"score": 1, "reasoning": "used the right tool"}'}
+        # Called without await: a sync tool stays sync through the recorder.
+        assert tool_handlers["lookup_order"]({"id": "A1"}) == "order A1 shipped"
+        return {"output": "Your order shipped."}
+
+    result = await evals.run(
+        project_key="proj",
+        key="support-qa",
+        dataset="golden",
+        handler=handler,
+        tools={"lookup_order": lookup_order},
+        generation={"provider": "OpenAI", "model": "gpt-4o"},
+        criteria=[Judge(key="$ld:ai:judge:accuracy")],
+    )
+
+    assert result.passed is True
+    history = seen["message_history"]
+    assert "Tools available: lookup_order" in history
+    assert '1. lookup_order\n   arguments: {"id":"A1"}' in history
+    assert "result: order A1 shipped" in history
+    # The trajectory sits between the request and the answer, because that is
+    # where it happened: a judge reading the history sees the question, what the
+    # agent did about it, then what it replied.
+    assert history.index("Question 0") < history.index("Tools available")
+    assert history.index("Tools available") < history.index("Your order shipped.")
+    assert history.index("Your order shipped.") < history.index(
+        "Your response MUST be in valid JSON"
+    )
+
+
+@pytest.mark.asyncio
+async def test_each_row_gets_only_its_own_tool_trajectory(
+    monkeypatch: pytest.MonkeyPatch,
+    stub_sdk_client: MagicMock,
+) -> None:
+    """Rows generate concurrently against one shared tool map.
+
+    A recorder shared across rows would splice row 0's calls into row 1's
+    trajectory and hand the judge a conversation that never happened.
+    """
+    import asyncio
+
+    transport = tool_run_transport(rows=2)
+    accuracy_judge_variation(monkeypatch)
+    evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport)
+    histories: dict[str, str] = {}
+    both_started = asyncio.Barrier(2)
+
+    def lookup_order(args: dict[str, Any]) -> str:
+        return f"order {args['id']}"
+
+    async def handler(
+        config: dict[str, Any],
+        user_input: str | None,
+        tool_handlers: dict[str, Callable[..., Any]],
+        variables: dict[str, Any],
+    ) -> dict[str, Any]:
+        if "Judge" in config.get("instructions", ""):
+            histories[str(user_input)] = variables["message_history"]
+            return {"output": '{"score": 1, "reasoning": "ok"}'}
+        row = str(user_input).split()[-1]
+        # Interleave the two rows' tool calls so a shared recorder would be
+        # caught rather than merely be possible.
+        await both_started.wait()
+        tool_handlers["lookup_order"]({"id": row})
+        return {"output": f"answered {row}"}
+
+    result = await evals.run(
+        project_key="proj",
+        key="support-qa",
+        dataset="golden",
+        handler=handler,
+        tools={"lookup_order": lookup_order},
+        generation={"provider": "OpenAI", "model": "gpt-4o"},
+        criteria=[Judge(key="$ld:ai:judge:accuracy")],
+        concurrency=2,
+    )
+
+    assert result.passed is True
+    assert '{"id":"0"}' in histories["answered 0"]
+    assert '{"id":"1"}' not in histories["answered 0"]
+    assert '{"id":"1"}' in histories["answered 1"]
+    assert '{"id":"0"}' not in histories["answered 1"]
+
+
+@pytest.mark.asyncio
+async def test_a_row_that_called_no_tools_says_so_to_the_judge(
+    monkeypatch: pytest.MonkeyPatch,
+    stub_sdk_client: MagicMock,
+) -> None:
+    """A judge grading tool selection needs to see the tool that went unused."""
+    transport = tool_run_transport()
+    accuracy_judge_variation(monkeypatch)
+    evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport)
+    seen: dict[str, str] = {}
+
+    async def handler(
+        config: dict[str, Any],
+        user_input: str | None,
+        tool_handlers: dict[str, Callable[..., Any]],
+        variables: dict[str, Any],
+    ) -> dict[str, Any]:
+        if "Judge" in config.get("instructions", ""):
+            seen["message_history"] = variables["message_history"]
+            return {"output": '{"score": 0, "reasoning": "should have looked it up"}'}
+        return {"output": "I do not know."}
+
+    await evals.run(
+        project_key="proj",
+        key="support-qa",
+        dataset="golden",
+        handler=handler,
+        tools={"lookup_order": lambda args: "unused"},
+        generation={"provider": "OpenAI", "model": "gpt-4o"},
+        criteria=[Judge(key="$ld:ai:judge:accuracy")],
+    )
+
+    assert (
+        "Tools available: lookup_order\n"
+        "No tool calls were made while producing the response."
+    ) in seen["message_history"]
+
+
+@pytest.mark.asyncio
+async def test_a_run_without_tools_leaves_message_history_unchanged(
+    monkeypatch: pytest.MonkeyPatch,
+    stub_sdk_client: MagicMock,
+) -> None:
+    """Judges authored before trajectories existed must read the same history.
+
+    With no observable tools there is nothing to report, so no trajectory block
+    is added rather than one saying no tools were called.
+    """
+    transport = judge_run_transport()
+    accuracy_judge_variation(monkeypatch)
+    evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport)
+    seen: dict[str, str] = {}
+
+    async def handler(
+        config: dict[str, Any],
+        user_input: str | None,
+        tool_handlers: dict[str, Callable[..., Any]],
+        variables: dict[str, Any],
+    ) -> dict[str, Any]:
+        if "Judge" in config.get("instructions", ""):
+            seen["message_history"] = variables["message_history"]
+            return {"output": '{"score": 1, "reasoning": "ok"}'}
+        return {"output": "generated"}
+
+    await evals.run(
+        project_key="proj",
+        key="support-qa",
+        dataset="golden",
+        handler=handler,
+        generation={"provider": "OpenAI", "model": "gpt-4o"},
+        criteria=[Judge(key="$ld:ai:judge:accuracy")],
+    )
+
+    assert seen["message_history"].startswith("Question A\n\ngenerated\n\n")
+    assert "Tools available" not in seen["message_history"]
+
+
+@pytest.mark.asyncio
+async def test_tool_result_placeholders_are_not_expanded_into_the_judge_prompt(
+    monkeypatch: pytest.MonkeyPatch,
+    stub_sdk_client: MagicMock,
+) -> None:
+    """A tool result is now judge-prompt input, so it is an injection surface.
+
+    It stays literal for the same reason the generated output does: the judge
+    config is handed over unrendered and the handler makes exactly one template
+    pass, so a substituted value is never rescanned for placeholders.
+    """
+    from launchdarkly_ai_server import parse_template
+
+    transport = tool_run_transport()
+
+    async def fake_extract_variation(
+        key: str, context: dict[str, Any]
+    ) -> dict[str, Any]:
+        return {
+            "config": {
+                "provider": {"name": "OpenAI"},
+                "model": {"name": "gpt-4o"},
+                "instructions": "Judge this history: {{message_history}}",
+            },
+            "meta": {"variationKey": "default", "version": 12},
+        }
+
+    monkeypatch.setattr(
+        "launchdarkly_ai_server.evaluations.runner.extract_variation",
+        fake_extract_variation,
+    )
+    evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport)
+
+    async def handler(
+        config: dict[str, Any],
+        user_input: str | None,
+        tool_handlers: dict[str, Callable[..., Any]],
+        variables: dict[str, Any],
+    ) -> dict[str, Any]:
+        if "Judge this history" in config.get("instructions", ""):
+            rendered = parse_template(config["instructions"], variables)
+            assert "result: {{expected_output}} leaked?" in rendered
+            assert "Answer leaked?" not in rendered
+            return {"output": '{"score": 1, "reasoning": "ok"}'}
+        tool_handlers["lookup_order"]({"id": "A1"})
+        return {"output": "done"}
+
+    result = await evals.run(
+        project_key="proj",
+        key="support-qa",
+        dataset="golden",
+        handler=handler,
+        tools={"lookup_order": lambda args: "{{expected_output}} leaked?"},
+        generation={"provider": "OpenAI", "model": "gpt-4o"},
+        criteria=[Judge(key="$ld:ai:judge:accuracy")],
+    )
+
+    assert result.passed is True
+
+
+@pytest.mark.asyncio
+async def test_a_failed_row_keeps_the_calls_made_before_the_handler_raised() -> None:
+    """The trajectory of a row that errored is what explains why it errored."""
+    from launchdarkly_ai_server.evaluations.api import LDApiClient
+    from launchdarkly_ai_server.evaluations.runner import EvaluationsRunner
+
+    runner = EvaluationsRunner(
+        LDApiClient(api_token="token", transport=failing_transport)
+    )
+
+    async def handler(
+        config: dict[str, Any],
+        user_input: str | None,
+        tool_handlers: dict[str, Callable[..., Any]],
+        variables: dict[str, Any],
+    ) -> dict[str, Any]:
+        tool_handlers["lookup_order"]({"id": "A1"})
+        raise RuntimeError("model refused")
+
+    results = await runner._run_rows(
+        [DatasetRow(row_index=0, input="Question")],
+        handler,
+        {"provider": {"name": "OpenAI"}, "model": {"name": "gpt-4o"}},
+        {"lookup_order": lambda args: "shipped"},
+        1,
+    )
+
+    assert results[0]["status"] == "ERROR"
+    assert [invocation.name for invocation in results[0]["tool_calls"]] == [
+        "lookup_order"
+    ]
+    assert results[0]["tool_calls"][0].result == "shipped"
diff --git a/packages/client/tests/test_judge_message_history.py b/packages/client/tests/test_judge_message_history.py
new file mode 100644
index 00000000..74dabd62
--- /dev/null
+++ b/packages/client/tests/test_judge_message_history.py
@@ -0,0 +1,448 @@
+"""One message_history for every judge path.
+
+The three paths -- online inline, online deferred, and offline evaluations --
+each joined their own, and disagreed. These tests hold them to
+``build_message_history``.
+"""
+
+from __future__ import annotations
+
+import pickle
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import launchdarkly_ai_server.lifecycle as lifecycle_module
+from launchdarkly_ai_server import JudgeTask, ProviderHandler, run_judge, run_judges
+from launchdarkly_ai_server.judge_scoring import (
+    FORMATTING_INSTRUCTIONS,
+    build_message_history,
+)
+from launchdarkly_ai_server.tracking import execute_and_track
+from launchdarkly_ai_server.trajectory import TrajectoryRecorder, render_trajectory
+
+CONTEXT = {"kind": "user", "key": "u1"}
+
+JUDGE_CONFIG = {
+    "model": {"name": "gpt-4"},
+    "provider": {"name": "TestProvider"},
+    "instructions": "Judge it",
+}
+
+
+@pytest.fixture
+def mock_ld_client() -> Any:
+    client = MagicMock()
+    client.track = MagicMock()
+    client.flush = AsyncMock()
+    client.close = AsyncMock()
+    client.variation = AsyncMock(return_value=None)
+    lifecycle_module._set_client_for_testing(client)
+    yield client
+    lifecycle_module._reset_for_testing()
+
+
+def capturing_judge_handler(seen: list[dict[str, Any]]) -> ProviderHandler:
+    async def fn(
+        config: Any,
+        user_input: Any,
+        tool_handlers: Any,
+        variables: Any,
+        history: Any = None,
+    ) -> dict[str, Any]:
+        seen.append(dict(variables or {}))
+        return {
+            "output": '{"score": 1, "reasoning": "ok"}',
+            "usage": {"input_tokens": 1, "output_tokens": 1},
+        }
+
+    return ProviderHandler(fn=fn, provides_for=("TestProvider", "messages"))  # type: ignore[arg-type]
+
+
+def judged_config() -> dict[str, Any]:
+    return {
+        "model": {"name": "gpt-4"},
+        "provider": {"name": "TestProvider"},
+        "instructions": "hi",
+        "judgeConfiguration": {"judges": [{"key": "judge-1", "samplingRate": 1}]},
+    }
+
+
+@pytest.fixture
+def judge_variation(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_extract_variation(key: str, context: Any) -> dict[str, Any]:
+        return {
+            "config": dict(JUDGE_CONFIG),
+            "meta": {"variationKey": "v", "version": 1},
+        }
+
+    monkeypatch.setattr(
+        "launchdarkly_ai_server.lifecycle.extract_variation", fake_extract_variation
+    )
+
+
+# ─── the builder itself ──────────────────────────────────────────────────────
+
+
+def test_builder_orders_the_conversation_and_appends_the_format_block() -> None:
+    history = build_message_history(user_input="Q", trajectory="T", output="A")
+
+    assert history == f"Q\n\nT\n\nA\n\n{FORMATTING_INSTRUCTIONS}"
+
+
+def test_builder_skips_empty_parts() -> None:
+    """Keeps a judge authored before trajectories existed scoring unchanged."""
+    assert build_message_history(user_input="Q", trajectory="", output="A") == (
+        f"Q\n\nA\n\n{FORMATTING_INSTRUCTIONS}"
+    )
+    assert build_message_history(output="A") == f"A\n\n{FORMATTING_INSTRUCTIONS}"
+
+
+def test_builder_always_carries_the_format_block() -> None:
+    """Judges from the AI Library's templates read the JSON shape from here."""
+    assert FORMATTING_INSTRUCTIONS in build_message_history()
+
+
+# ─── online: inline ──────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_inline_judge_is_shown_the_trajectory(
+    mock_ld_client: Any, judge_variation: None
+) -> None:
+    seen: list[dict[str, Any]] = []
+    await run_judges(
+        config=judged_config(),
+        user_context=CONTEXT,
+        handler=capturing_judge_handler(seen),
+        user_input="Where is order A1?",
+        llm_response="It shipped.",
+        base_track_data={},
+        trajectory="Tools available: lookup\n1. lookup\n   result: shipped",
+    )
+
+    history = seen[0]["message_history"]
+    assert "1. lookup" in history
+    assert (
+        history.index("Where is order A1?")
+        < history.index("1. lookup")
+        < history.index("It shipped.")
+    )
+
+
+@pytest.mark.asyncio
+async def test_inline_judge_without_a_trajectory_is_unchanged(
+    mock_ld_client: Any, judge_variation: None
+) -> None:
+    seen: list[dict[str, Any]] = []
+    await run_judges(
+        config=judged_config(),
+        user_context=CONTEXT,
+        handler=capturing_judge_handler(seen),
+        user_input="Q",
+        llm_response="A",
+        base_track_data={},
+    )
+
+    assert seen[0]["message_history"] == build_message_history(
+        user_input="Q", output="A"
+    )
+
+
+# ─── online: deferred ────────────────────────────────────────────────────────
+
+
+def deferred_task(**overrides: Any) -> JudgeTask:
+    fields: dict[str, Any] = {
+        "config_key": "judge-1",
+        "judge_config": dict(JUDGE_CONFIG),
+        "judge_meta": {"variationKey": "v", "version": 1},
+        "actual_output": "It shipped.",
+        "user_context": CONTEXT,
+        "judge_provider": "TestProvider",
+        "judge_mode": "messages",
+        "collapse_messages": False,
+        "parent_track_data": {},
+    }
+    fields.update(overrides)
+    return JudgeTask(**fields)
+
+
+@pytest.mark.asyncio
+async def test_deferred_judge_is_shown_the_input_and_the_trajectory(
+    mock_ld_client: Any,
+) -> None:
+    """This path carried neither before, grading a response in isolation."""
+    seen: list[dict[str, Any]] = []
+    task = deferred_task(
+        user_input="Where is order A1?",
+        trajectory="Tools available: lookup\n1. lookup\n   result: shipped",
+    )
+
+    await run_judge(task, [capturing_judge_handler(seen)])
+
+    history = seen[0]["message_history"]
+    assert (
+        history.index("Where is order A1?")
+        < history.index("1. lookup")
+        < history.index("It shipped.")
+    )
+
+
+@pytest.mark.asyncio
+async def test_deferred_and_inline_agree_on_the_same_row(
+    mock_ld_client: Any, judge_variation: None
+) -> None:
+    """The point of the shared builder: same inputs, identical history."""
+    inline_seen: list[dict[str, Any]] = []
+    deferred_seen: list[dict[str, Any]] = []
+    trajectory = "Tools available: lookup\n1. lookup\n   result: shipped"
+
+    await run_judges(
+        config=judged_config(),
+        user_context=CONTEXT,
+        handler=capturing_judge_handler(inline_seen),
+        user_input="Where is order A1?",
+        llm_response="It shipped.",
+        base_track_data={},
+        trajectory=trajectory,
+    )
+    await run_judge(
+        deferred_task(user_input="Where is order A1?", trajectory=trajectory),
+        [capturing_judge_handler(deferred_seen)],
+    )
+
+    assert inline_seen[0]["message_history"] == deferred_seen[0]["message_history"]
+
+
+def test_judge_task_stays_picklable_with_the_new_fields() -> None:
+    """JudgeTask crosses a thread boundary, so it must stay primitives."""
+    task = deferred_task(user_input="Q", trajectory="T")
+
+    assert pickle.loads(pickle.dumps(task)).trajectory == "T"
+
+
+# ─── online: capture ─────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_execute_and_track_records_the_trajectory(mock_ld_client: Any) -> None:
+    def lookup(args: dict[str, Any]) -> str:
+        return f"order {args['id']} shipped"
+
+    async def handler(
+        config: Any,
+        user_input: Any,
+        tool_handlers: Any,
+        variables: Any,
+        history: Any = None,
+    ) -> dict[str, Any]:
+        # Awaited: wrap_tool_handlers wraps the recorder's wrapper and is
+        # always async, so an online handler awaits as it always has. Offline
+        # there is no such wrapper -- see test_trajectory.py.
+        await tool_handlers["lookup"]({"id": "A1"})
+        return {"output": "It shipped.", "usage": {}}
+
+    result = await execute_and_track(
+        config_key="c",
+        config={
+            "model": {"name": "m"},
+            "provider": {"name": "TestProvider"},
+            # Only tools the config offers are described.
+            "tools": {"lookup": {"description": "", "parameters": {}}},
+        },
+        meta={"variationKey": "v", "version": 1},
+        user_context=CONTEXT,
+        handler=handler,  # type: ignore[arg-type]
+        user_input="Where is order A1?",
+        tool_handlers={"lookup": lookup},
+    )
+
+    assert "Tools available: lookup" in result["trajectory"]
+    assert '1. lookup\n   arguments: {"id":"A1"}' in result["trajectory"]
+    assert "result: order A1 shipped" in result["trajectory"]
+
+
+@pytest.mark.asyncio
+async def test_a_native_tool_is_not_recorded_online_either(
+    mock_ld_client: Any,
+) -> None:
+    """The stub wrap_tool_handlers substitutes for a native tool is callable,
+    so the call *is* observable online -- but it returns nothing, so recording
+    it would show a judge a call with an empty result. Both paths skip it.
+    """
+    from launchdarkly_ai_server import NativeTool
+
+    async def handler(
+        config: Any,
+        user_input: Any,
+        tool_handlers: Any,
+        variables: Any,
+        history: Any = None,
+    ) -> dict[str, Any]:
+        # Not awaited: the native stub is sync, unlike the async wrapper a
+        # real callable gets. Pre-existing asymmetry.
+        tool_handlers["web_search"]({"q": "x"})
+        return {"output": "done", "usage": {}}
+
+    result = await execute_and_track(
+        config_key="c",
+        config={"model": {"name": "m"}, "provider": {"name": "TestProvider"}},
+        meta={"variationKey": "v", "version": 1},
+        user_context=CONTEXT,
+        handler=handler,  # type: ignore[arg-type]
+        user_input="q",
+        tool_handlers={"web_search": NativeTool("WebSearch")},
+    )
+
+    assert result["trajectory"] == ""
+
+
+@pytest.mark.asyncio
+async def test_tool_tracking_still_fires_under_the_recorder(
+    mock_ld_client: Any,
+) -> None:
+    """Recording composes inside wrap_tool_handlers, which must still fire."""
+
+    async def handler(
+        config: Any,
+        user_input: Any,
+        tool_handlers: Any,
+        variables: Any,
+        history: Any = None,
+    ) -> dict[str, Any]:
+        await tool_handlers["lookup"]({"id": "A1"})
+        return {"output": "ok", "usage": {}}
+
+    await execute_and_track(
+        config_key="c",
+        config={
+            "model": {"name": "m"},
+            "provider": {"name": "TestProvider"},
+            "tools": {"lookup": {"description": "", "parameters": {}}},
+        },
+        meta={"variationKey": "v", "version": 1},
+        user_context=CONTEXT,
+        handler=handler,  # type: ignore[arg-type]
+        user_input="q",
+        tool_handlers={"lookup": lambda args: "shipped"},
+    )
+
+    tool_events = [
+        call.args
+        for call in mock_ld_client.track.call_args_list
+        if call.args[0] == "$ld:ai:tool_call"
+    ]
+    assert len(tool_events) == 1
+    assert tool_events[0][2]["toolKey"] == "lookup"
+
+
+@pytest.mark.asyncio
+async def test_a_run_with_no_tools_reports_no_trajectory(mock_ld_client: Any) -> None:
+    async def handler(
+        config: Any,
+        user_input: Any,
+        tool_handlers: Any,
+        variables: Any,
+        history: Any = None,
+    ) -> dict[str, Any]:
+        return {"output": "ok", "usage": {}}
+
+    result = await execute_and_track(
+        config_key="c",
+        config={"model": {"name": "m"}, "provider": {"name": "TestProvider"}},
+        meta={"variationKey": "v", "version": 1},
+        user_context=CONTEXT,
+        handler=handler,  # type: ignore[arg-type]
+        user_input="q",
+    )
+
+    assert result["trajectory"] == ""
+
+
+def test_the_recorder_renders_identically_for_both_paths() -> None:
+    """Both paths call render_trajectory, so one fixture pins the shape."""
+    recorder = TrajectoryRecorder()
+    recorder.wrap({"lookup": lambda args: "shipped"})
+
+    assert render_trajectory(
+        recorder.invocations,
+        observable_tools=recorder.observable_tools,
+        omitted=recorder.omitted,
+    ) == (
+        "Tools available: lookup\nNo tool calls were made while producing the response."
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_registry_tool_the_config_omits_is_not_described(
+    mock_ld_client: Any,
+) -> None:
+    """Describing a registry tool the variation omits would let a judge
+    penalise an agent for ignoring a tool it never had.
+    """
+
+    async def handler(
+        config: Any,
+        user_input: Any,
+        tool_handlers: Any,
+        variables: Any,
+        history: Any = None,
+    ) -> dict[str, Any]:
+        await tool_handlers["lookup"]({"id": "A1"})
+        return {"output": "ok", "usage": {}}
+
+    result = await execute_and_track(
+        config_key="c",
+        config={
+            "model": {"name": "m"},
+            "provider": {"name": "TestProvider"},
+            # The variation offers one tool; the map carries two.
+            "tools": {"lookup": {"description": "", "parameters": {}}},
+        },
+        meta={"variationKey": "v", "version": 1},
+        user_context=CONTEXT,
+        handler=handler,  # type: ignore[arg-type]
+        user_input="q",
+        tool_handlers={
+            "lookup": lambda args: "shipped",
+            "registry_only": lambda args: "never offered",
+        },
+    )
+
+    assert "Tools available: lookup" in result["trajectory"]
+    assert "registry_only" not in result["trajectory"]
+
+
+@pytest.mark.asyncio
+async def test_a_handoff_tool_is_not_described_online(mock_ld_client: Any) -> None:
+    """A per-node judge is scored against the node's original config, which
+    lists no handoffs -- so a handoff must not read as tool use.
+    """
+
+    async def handler(
+        config: Any,
+        user_input: Any,
+        tool_handlers: Any,
+        variables: Any,
+        history: Any = None,
+    ) -> dict[str, Any]:
+        await tool_handlers["__handoff_billing"]({})
+        return {"output": "ok", "usage": {}}
+
+    result = await execute_and_track(
+        config_key="c",
+        config={
+            "model": {"name": "m"},
+            "provider": {"name": "TestProvider"},
+            "tools": {"__handoff_billing": {"description": "", "parameters": {}}},
+        },
+        meta={"variationKey": "v", "version": 1},
+        user_context=CONTEXT,
+        handler=handler,  # type: ignore[arg-type]
+        user_input="q",
+        tool_handlers={"__handoff_billing": lambda args: "Handoff recorded"},
+    )
+
+    assert result["trajectory"] == ""
diff --git a/packages/client/tests/test_trajectory.py b/packages/client/tests/test_trajectory.py
new file mode 100644
index 00000000..f7841721
--- /dev/null
+++ b/packages/client/tests/test_trajectory.py
@@ -0,0 +1,302 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from launchdarkly_ai_server.trajectory import (
+    MAX_RECORDED_VALUE_CHARS,
+    ToolInvocation,
+    TrajectoryRecorder,
+    render_trajectory,
+)
+from launchdarkly_ai_server.types import NativeTool
+
+
+def test_wrapped_sync_tool_stays_sync() -> None:
+    """A blanket async wrapper would hand a caller's handler a coroutine
+    object where it used to get the tool's value.
+    """
+
+    def lookup(args: dict[str, Any]) -> str:
+        return f"order {args['id']}"
+
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap({"lookup": lookup})
+
+    assert not asyncio.iscoroutinefunction(wrapped["lookup"])
+    assert wrapped["lookup"]({"id": "A1"}) == "order A1"
+    assert recorder.invocations == [
+        ToolInvocation(name="lookup", arguments={"id": "A1"}, result="order A1")
+    ]
+
+
+@pytest.mark.asyncio
+async def test_wrapped_async_tool_is_awaited() -> None:
+    async def lookup(args: dict[str, Any]) -> str:
+        await asyncio.sleep(0)
+        return f"order {args['id']}"
+
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap({"lookup": lookup})
+
+    assert asyncio.iscoroutinefunction(wrapped["lookup"])
+    assert await wrapped["lookup"]({"id": "A1"}) == "order A1"
+    assert recorder.invocations[0].result == "order A1"
+
+
+def test_wrapped_tool_reraises_and_records_the_failure() -> None:
+    """A tool that failed must still fail its caller."""
+
+    def refund(args: dict[str, Any]) -> str:
+        raise RuntimeError("gateway timeout")
+
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap({"refund": refund})
+
+    with pytest.raises(RuntimeError, match="gateway timeout"):
+        wrapped["refund"]({"id": "A1"})
+
+    assert recorder.invocations == [
+        ToolInvocation(name="refund", arguments={"id": "A1"}, error="gateway timeout")
+    ]
+
+
+@pytest.mark.asyncio
+async def test_concurrent_calls_keep_their_start_order() -> None:
+    """A judge asked whether search came before refund is reading a sequence,
+    so completion order would answer a different question.
+    """
+    started: dict[str, asyncio.Event] = {"slow": asyncio.Event()}
+
+    async def slow(args: dict[str, Any]) -> str:
+        started["slow"].set()
+        await asyncio.sleep(0.02)
+        return "slow done"
+
+    async def fast(args: dict[str, Any]) -> str:
+        await started["slow"].wait()
+        return "fast done"
+
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap({"slow": slow, "fast": fast})
+
+    slow_task = asyncio.create_task(wrapped["slow"]({}))
+    await started["slow"].wait()
+    await wrapped["fast"]({})
+    await slow_task
+
+    assert [invocation.name for invocation in recorder.invocations] == ["slow", "fast"]
+
+
+def test_calls_past_the_limit_still_execute_but_are_only_counted() -> None:
+    calls: list[int] = []
+
+    def append(args: dict[str, Any]) -> str:
+        calls.append(args["n"])
+        return "ok"
+
+    recorder = TrajectoryRecorder(limit=2)
+    wrapped = recorder.wrap({"append": append})
+    for n in range(5):
+        wrapped["append"]({"n": n})
+
+    # Every call ran: truncation bounds the record, not the behaviour.
+    assert calls == [0, 1, 2, 3, 4]
+    assert len(recorder.invocations) == 2
+    assert recorder.omitted == 3
+
+
+def test_native_tools_pass_through_unwrapped_and_undescribed() -> None:
+    """Invisible, so not advertised either: listing it would let a judge
+    conclude the model ignored a tool it may well have used.
+    """
+    native = NativeTool("WebSearch")
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap({"web_search": native, "lookup": lambda args: "ok"})
+
+    assert wrapped["web_search"] is native
+    assert recorder.observable_tools == ["lookup"]
+
+
+def test_keyword_arguments_are_recorded() -> None:
+    def lookup(**kwargs: Any) -> str:
+        return "ok"
+
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap({"lookup": lookup})
+    wrapped["lookup"](id="A1")
+
+    assert recorder.invocations[0].arguments == {"id": "A1"}
+
+
+def test_render_lists_available_tools_calls_arguments_and_results() -> None:
+    rendered = render_trajectory(
+        [
+            ToolInvocation(name="lookup", arguments={"id": "A1"}, result="shipped"),
+            ToolInvocation(
+                name="refund", arguments={"id": "A1"}, error="gateway timeout"
+            ),
+        ],
+        observable_tools=["lookup", "refund"],
+    )
+
+    assert rendered == (
+        "Tools available: lookup, refund\n"
+        "Tool calls made while producing the response, in order:\n"
+        "1. lookup\n"
+        '   arguments: {"id":"A1"}\n'
+        "   result: shipped\n"
+        "2. refund\n"
+        '   arguments: {"id":"A1"}\n'
+        "   error: gateway timeout"
+    )
+
+
+def test_render_reports_an_empty_trajectory_when_tools_were_available() -> None:
+    """ "Called nothing" is the finding a tool-selection judge most needs."""
+    rendered = render_trajectory([], observable_tools=["lookup"])
+
+    assert rendered == (
+        "Tools available: lookup\nNo tool calls were made while producing the response."
+    )
+
+
+def test_render_is_empty_when_there_was_nothing_observable() -> None:
+    assert render_trajectory([], observable_tools=[]) == ""
+
+
+def test_render_reports_omitted_calls() -> None:
+    rendered = render_trajectory(
+        [ToolInvocation(name="lookup", arguments=None, result="ok")],
+        observable_tools=["lookup"],
+        omitted=3,
+    )
+
+    assert "(3 further tool call(s) were made but not recorded.)" in rendered
+
+
+def test_render_truncates_an_oversized_value() -> None:
+    rendered = render_trajectory(
+        [ToolInvocation(name="fetch", arguments={}, result="x" * 5000)],
+        observable_tools=["fetch"],
+    )
+
+    assert f"   result: {'x' * MAX_RECORDED_VALUE_CHARS}… (truncated)" in rendered
+
+
+def test_render_serializes_unserializable_values_without_raising() -> None:
+    class Opaque:
+        def __str__(self) -> str:
+            return "<opaque>"
+
+    rendered = render_trajectory(
+        [ToolInvocation(name="fetch", arguments={"k": Opaque()}, result=Opaque())],
+        observable_tools=["fetch"],
+    )
+
+    assert "<opaque>" in rendered
+
+
+def test_render_does_not_escape_non_ascii() -> None:
+    """A judge reads this string, so it must see the actual characters.
+
+    json.dumps escapes non-ASCII by default, which reached the judge as
+    ``caf\\u00e9`` -- noise it then has to grade a tool result through.
+    """
+    rendered = render_trajectory(
+        [
+            ToolInvocation(
+                name="lookup",
+                arguments={"city": "café", "place": "東京"},
+                result={"note": "naïve"},
+            )
+        ],
+        observable_tools=["lookup"],
+    )
+
+    assert '   arguments: {"city":"café","place":"東京"}' in rendered
+    assert '   result: {"note":"naïve"}' in rendered
+    assert "\\u" not in rendered
+
+
+def test_render_leaves_a_non_ascii_string_result_alone() -> None:
+    """A string result was never JSON-encoded, so it never escaped. Pins that
+    both paths agree now.
+    """
+    rendered = render_trajectory(
+        [ToolInvocation(name="lookup", arguments={}, result="café 東京")],
+        observable_tools=["lookup"],
+    )
+
+    assert "   result: café 東京" in rendered
+
+
+def test_a_sync_tool_returning_an_awaitable_records_on_completion() -> None:
+    """Records on completion, so a judge never reads a pending coroutine's
+    repr. Never awaited means never recorded -- the call did not complete.
+    """
+
+    async def inner() -> str:
+        return "shipped"
+
+    def lookup(args: dict[str, Any]) -> Any:
+        return inner()
+
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap({"lookup": lookup})
+    pending = wrapped["lookup"]({"id": "A1"})
+
+    assert recorder.invocations[0].result is None
+    assert (
+        asyncio.get_event_loop_policy().new_event_loop().run_until_complete(pending)
+        == "shipped"
+    )
+    assert recorder.invocations[0].result == "shipped"
+
+
+def test_handoff_tools_are_not_recorded_or_described() -> None:
+    """Not tools the agent was given, and absent from the config a per-node
+    judge is scored against. wrap_tool_handlers excludes them too.
+    """
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap(
+        {
+            "lookup": lambda args: "shipped",
+            "__handoff_billing": lambda args: "Handoff to billing recorded",
+        }
+    )
+
+    assert recorder.observable_tools == ["lookup"]
+    # Still passed through, so routing keeps working.
+    assert "__handoff_billing" in wrapped
+    wrapped["__handoff_billing"]({})
+    assert [i.name for i in recorder.invocations] == []
+
+
+def test_only_tools_the_config_exposed_are_recorded_or_described() -> None:
+    """The map can be wider than what the model was offered; describing the
+    extras would let a judge penalise an agent for ignoring them.
+    """
+    recorder = TrajectoryRecorder()
+    wrapped = recorder.wrap(
+        {"lookup": lambda args: "shipped", "refund": lambda args: "refunded"},
+        exposed={"lookup"},
+    )
+
+    assert recorder.observable_tools == ["lookup"]
+    wrapped["refund"]({})
+    assert [i.name for i in recorder.invocations] == []
+    wrapped["lookup"]({})
+    assert [i.name for i in recorder.invocations] == ["lookup"]
+
+
+def test_no_exposed_set_means_every_callable_is_in_scope() -> None:
+    """The offline case: the runner resolves the config's tools from this same
+    map, so no filter is needed.
+    """
+    recorder = TrajectoryRecorder()
+    recorder.wrap({"lookup": lambda args: "a", "refund": lambda args: "b"})
+
+    assert recorder.observable_tools == ["lookup", "refund"]