diff --git a/src/agent_env/cli/task/run.py b/src/agent_env/cli/task/run.py
index 6a9eb702..4230d470 100644
--- a/src/agent_env/cli/task/run.py
+++ b/src/agent_env/cli/task/run.py
@@ -18,6 +18,7 @@
 from agent_env.store.ids import derive_id, fs_safe, is_local_id, validate_local_id
 from agent_env.task.interrupts import Interrupts
 from agent_env.task.teardown import TeardownReport, kind, teardown_run
+from agent_env.task_step.context import read_dual_keyed
 from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep
 
 
@@ -100,12 +101,10 @@ def _format_prompt_response(pr) -> str:
         "  response: |",
         indented_response,
     ]
-    if pr.agent_trajectory_s3_uri:
-        lines.append(f"  agent_trajectory_s3_uri: {pr.agent_trajectory_s3_uri}")
-    if pr.agent_trajectory_s3_prefix:
-        lines.append(f"  agent_trajectory_s3_prefix: {pr.agent_trajectory_s3_prefix}")
-    if pr.compact_trajectory_s3_uri:
-        lines.append(f"  compact_trajectory_s3_uri: {pr.compact_trajectory_s3_uri}")
+    if pr.agent_trajectory_object_url:
+        lines.append(f"  agent_trajectory_object_url: {pr.agent_trajectory_object_url}")
+    if pr.agent_trajectory_object_prefix:
+        lines.append(f"  agent_trajectory_object_prefix: {pr.agent_trajectory_object_prefix}")
     if pr.agent_trajectory_file_path:
         lines.append(f"  agent_trajectory_file_path: {pr.agent_trajectory_file_path}")
     if pr.tool_call_count is not None:
@@ -192,12 +191,12 @@ def _step_output_lines(step, context) -> list[str]:
 
 def _format_verifier_output(vid: str, vdata: dict) -> list[str]:
     lines: list[str] = []
-    compact_uri = vdata.get("compact_trajectory_s3_uri")
-    if compact_uri:
-        lines.append(f"compact_trajectory_s3_uri (verifier_id={vid}): {compact_uri}")
-    judge_uri = vdata.get("judge_trajectory_s3_uri")
-    if judge_uri:
-        lines.append(f"judge_trajectory_s3_uri (verifier_id={vid}): {judge_uri}")
+    compact_url = read_dual_keyed(vdata, "compact_trajectory_s3_uri", "compact_trajectory_object_url")
+    if compact_url:
+        lines.append(f"compact_trajectory_object_url (verifier_id={vid}): {compact_url}")
+    judge_url = read_dual_keyed(vdata, "judge_trajectory_s3_uri", "judge_trajectory_object_url")
+    if judge_url:
+        lines.append(f"judge_trajectory_object_url (verifier_id={vid}): {judge_url}")
     command = vdata.get("command")
     if command is not None:
         lines.append(f"command (verifier_id={vid}): {command}")
@@ -209,7 +208,8 @@ def _format_verifier_output(vid: str, vdata: dict) -> list[str]:
     if stdout_artifact:
         lines.append(
             f"stdout_artifact (verifier_id={vid}): "
-            f"{stdout_artifact.get('id')} v{stdout_artifact.get('version')} -> {stdout_artifact.get('s3_url')}"
+            f"{stdout_artifact.get('id')} v{stdout_artifact.get('version')} -> "
+            f"{read_dual_keyed(stdout_artifact, 's3_url', 'object_url')}"
         )
     stdout_head = vdata.get("stdout_head")
     if stdout_head:
@@ -220,7 +220,8 @@ def _format_verifier_output(vid: str, vdata: dict) -> list[str]:
     if stderr_artifact:
         lines.append(
             f"stderr_artifact (verifier_id={vid}): "
-            f"{stderr_artifact.get('id')} v{stderr_artifact.get('version')} -> {stderr_artifact.get('s3_url')}"
+            f"{stderr_artifact.get('id')} v{stderr_artifact.get('version')} -> "
+            f"{read_dual_keyed(stderr_artifact, 's3_url', 'object_url')}"
         )
     stderr_head = vdata.get("stderr_head")
     if stderr_head:
diff --git a/src/agent_env/task_step/context.py b/src/agent_env/task_step/context.py
index 89e0ad06..ed69bf42 100644
--- a/src/agent_env/task_step/context.py
+++ b/src/agent_env/task_step/context.py
@@ -7,6 +7,7 @@
 from typing import Any
 
 from agent_env.env.env import DeployedEnv
+from agent_env.utils.deprecation import warn_deprecated
 
 _REDACTED_KEYS = {
     "litellm_api_key", "judge_litellm_api_key", "usersim_api_key", "remote_tokens", "cf_access_client_secret",
@@ -110,6 +111,27 @@ def from_dict(cls, data: dict) -> DeployedSandbox:
         )
 
 
+def dual_keyed(legacy: str, neutral: str, value: Any) -> dict[str, Any]:
+    """A renamed run-context key under both its names: readers on older versions read the legacy one."""
+    return {legacy: value, neutral: value}
+
+
+def read_dual_keyed(data: dict[str, Any], legacy: str, neutral: str) -> Any:
+    """A renamed run-context key, legacy name first while both are written: a raw-doc writer that knows only the
+    legacy name leaves the neutral one stale. A legacy key holding None falls through to the neutral one."""
+    value = data.get(legacy)
+    return value if value is not None else data.get(neutral)
+
+
+# (legacy, neutral) PromptResponse field names. The legacy fields stay real fields, mirrored from the neutral ones at
+# construction, so asdict (which persists prompt responses) writes both keys.
+_PROMPT_RESPONSE_TWINS = (
+    ("agent_trajectory_s3_uri", "agent_trajectory_object_url"),
+    ("agent_trajectory_s3_prefix", "agent_trajectory_object_prefix"),
+    ("target_agent_per_turn_trajectory_s3_uris", "target_agent_per_turn_trajectory_object_urls"),
+)
+
+
 @dataclass
 class PromptResponse:
     prompt_id: str
@@ -133,6 +155,22 @@ class PromptResponse:
     agent_name: str | None = None
     step_id: str | None = None
     structured_output: Any = None
+    # Neutral names for the S3-named fields above, which are deprecated mirrors of them; appended, so positional
+    # construction is unchanged.
+    agent_trajectory_object_url: str | None = None
+    agent_trajectory_object_prefix: str | None = None
+    target_agent_per_turn_trajectory_object_urls: list[str | None] | None = None
+
+    def __post_init__(self):
+        for legacy, neutral in _PROMPT_RESPONSE_TWINS:
+            old, new = getattr(self, legacy), getattr(self, neutral)
+            if old is None:
+                setattr(self, legacy, list(new) if isinstance(new, list) else new)
+            elif new is None:
+                warn_deprecated(f"PromptResponse({legacy}=)", f"{neutral}=", kind="keyword", stacklevel=4)
+                setattr(self, neutral, list(old) if isinstance(old, list) else old)
+            elif old != new:
+                raise ValueError(f"PromptResponse got {legacy}={old!r} and {neutral}={new!r}, which disagree")
 
     @classmethod
     def from_dict(cls, data: dict) -> PromptResponse:
@@ -140,10 +178,14 @@ def from_dict(cls, data: dict) -> PromptResponse:
             prompt_id=data["prompt_id"],
             response=data["response"],
             prompt_text=data.get("prompt_text"),
-            agent_trajectory_s3_uri=data.get("agent_trajectory_s3_uri"),
-            agent_trajectory_s3_prefix=data.get("agent_trajectory_s3_prefix"),
+            agent_trajectory_object_url=read_dual_keyed(data, "agent_trajectory_s3_uri", "agent_trajectory_object_url"),
+            agent_trajectory_object_prefix=read_dual_keyed(
+                data, "agent_trajectory_s3_prefix", "agent_trajectory_object_prefix"
+            ),
             agent_trajectory_file_path=data.get("agent_trajectory_file_path"),
-            target_agent_per_turn_trajectory_s3_uris=data.get("target_agent_per_turn_trajectory_s3_uris"),
+            target_agent_per_turn_trajectory_object_urls=read_dual_keyed(
+                data, "target_agent_per_turn_trajectory_s3_uris", "target_agent_per_turn_trajectory_object_urls"
+            ),
             source_agent_per_turn_prompt_parts=data.get("source_agent_per_turn_prompt_parts"),
             compact_trajectory_s3_uri=data.get("compact_trajectory_s3_uri"),
             tool_call_count=data.get("tool_call_count"),
diff --git a/src/agent_env/task_step/snapshot_utils/snapshot_series.py b/src/agent_env/task_step/snapshot_utils/snapshot_series.py
index fecaee22..fe9628d2 100644
--- a/src/agent_env/task_step/snapshot_utils/snapshot_series.py
+++ b/src/agent_env/task_step/snapshot_utils/snapshot_series.py
@@ -18,7 +18,7 @@
 from typing import Optional
 
 from agent_env.store.ids import derive_id, is_local_id, validate_local_id
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import TaskStepContext, dual_keyed
 from agent_env.task_step.snapshot_utils import agent_state_capture as capture
 
 logger = logging.getLogger(__name__)
@@ -434,7 +434,7 @@ def remaining() -> float:
         row["bundle_object_url"] = workspace.bundle_object_url
 
         # Every point reads its trajectory the same way, cumulatively.
-        # `PromptResponse.agent_trajectory_s3_uri` is only the LAST turn's, so using
+        # `PromptResponse.agent_trajectory_object_url` is only the LAST turn's, so using
         # it for the final row would make that point narrower than its predecessors.
         # It stays the fallback: on a failed run the live read may be gone while the
         # per-turn upload already landed.
@@ -448,18 +448,19 @@ def remaining() -> float:
         )
         if traj.reason and is_final:
             recorded = self._recorded_trajectory_uri(context)
-            row["trajectory_s3_uri"] = recorded
+            row.update(dual_keyed("trajectory_s3_uri", "trajectory_object_url", recorded))
             reasons.append(traj.reason if recorded else f"{traj.reason}; trajectory_missing")
         elif traj.reason:
             reasons.append(traj.reason)
         elif traj.object_url:
-            row["trajectory_s3_uri"] = traj.object_url
+            row.update(dual_keyed("trajectory_s3_uri", "trajectory_object_url", traj.object_url))
         else:
             try:
-                row["trajectory_s3_uri"] = await asyncio.to_thread(
+                uploaded = await asyncio.to_thread(
                     capture.upload_trajectory, traj.trajectory,
                     self.trajectory_output_prefix,
                 )
+                row.update(dual_keyed("trajectory_s3_uri", "trajectory_object_url", uploaded))
             except Exception as exc:
                 logger.warning("%s: trajectory upload failed: %s", self.step_id, exc)
                 reasons.append("trajectory_upload_failed")
@@ -534,7 +535,7 @@ def _recorded_trajectory_uri(self, context: TaskStepContext) -> Optional[str]:
                 prompt.prompt_id == self.prompt_id
                 and prompt.a2a_context_id == self.a2a_context_id
             ):
-                return prompt.agent_trajectory_s3_uri
+                return prompt.agent_trajectory_object_url
         # None is legitimate: the response is recorded only after the turn loop, so
         # a capture during an in-flight or failed run has no entry yet.
         return None
@@ -564,7 +565,7 @@ def _row(
             "is_final": is_final,
             "capture_status": status,
             "capture_reason": reason,
-            "trajectory_s3_uri": None,
+            **dual_keyed("trajectory_s3_uri", "trajectory_object_url", None),
             "env_universe_id": None,
             "env_universe_version": None,
         }
diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py
index 6b34136b..3765cf4a 100644
--- a/src/agent_env/task_step/task_steps/prompt_agent.py
+++ b/src/agent_env/task_step/task_steps/prompt_agent.py
@@ -474,11 +474,11 @@ async def _execute_conversation(
         current_user_parts = initial_parts
         final_terminal = protocol.TerminalResponse(response_text="")
         final_task_id = ""
-        per_turn_trajectory_s3_uris: list[str | None] = []
+        per_turn_trajectory_urls: list[str | None] = []
         source_agent_per_turn_prompt_parts: list[list[dict] | None] = []
         traj_ext_cached = A2AAgent.find_extension(card, A2AAgent.EXT_TRAJECTORY)
         final_state: str = TaskState.completed.value
-        trajectory_s3_uri: Optional[str] = None
+        trajectory_url: Optional[str] = None
         # What the target has been sent: of the files its replies name, the only ones a user-sim is made able
         # to read, so a reply naming any other object a store owns can't read it out through the user-sim.
         sent_to_target: set[str] = set()
@@ -534,7 +534,7 @@ def record_turn() -> None:
                     )
                 except Exception as e:
                     logger.warning(f"Turn {turn+1} trajectory fetch failed (continuing): {e}")
-                per_turn_trajectory_s3_uris.append(turn_traj_uri)
+                per_turn_trajectory_urls.append(turn_traj_uri)
 
             final_task_id = sent_task_id
             final_state = target_state
@@ -639,8 +639,8 @@ def record_turn() -> None:
             except Exception as e:
                 logger.warning(f"agent-trigger state readback failed (continuing): {e}")
 
-        trajectory_s3_uri = next(
-            (uri for uri in reversed(per_turn_trajectory_s3_uris) if uri is not None),
+        trajectory_url = next(
+            (uri for uri in reversed(per_turn_trajectory_urls) if uri is not None),
             None,
         )
 
@@ -678,9 +678,9 @@ def record_turn() -> None:
             prompt_id=self.prompt_id,
             response=final_terminal.response_text,
             prompt_text=prompt_text,
-            agent_trajectory_s3_uri=trajectory_s3_uri,
-            agent_trajectory_s3_prefix=trajectory_output_prefix,
-            target_agent_per_turn_trajectory_s3_uris=per_turn_trajectory_s3_uris or None,
+            agent_trajectory_object_url=trajectory_url,
+            agent_trajectory_object_prefix=trajectory_output_prefix,
+            target_agent_per_turn_trajectory_object_urls=per_turn_trajectory_urls or None,
             source_agent_per_turn_prompt_parts=source_agent_per_turn_prompt_parts or None,
             tool_call_count=final_terminal.tool_call_count,
             model=model,
diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py
index e0b44bc2..7f8d5a3c 100644
--- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py
+++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py
@@ -18,7 +18,7 @@
 from agent_env.a2a_agent.staging import transfer_store
 from agent_env.config import get_config
 from agent_env.config.model import ModelParam
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import TaskStepContext, dual_keyed
 from agent_env.task_step.task_steps.verifiers.judge_utils.judge_output_format import (
     SCREENSHOT_GROUNDING,
     JudgeOutputFormat,
@@ -206,11 +206,11 @@ def _image_url_block(frame: ImageFrame) -> dict:
 class _CompactedTrajectory:
     """Every representation the filter produced; the caller picks what its judge consumes."""
 
-    container_s3_uri: Optional[str] = None       # trajectory file (loaded into an agent-judge container)
+    container_object_url: Optional[str] = None       # trajectory file (loaded into an agent-judge container)
     tool_result_files: list[tuple[str, str]] = field(default_factory=list)
     inline_text: Optional[str] = None            # pre-rendered text (screenshot); None for DEFAULT (read lazily)
     image_blocks: list[dict] = field(default_factory=list)  # frame images (screenshot); a label text block before each per-criterion
-    compact_s3_uri: Optional[str] = None         # compacted URI, if produced; recorded on the result only
+    compact_object_url: Optional[str] = None         # compacted URI, if produced; recorded on the result only
     per_criterion: bool = False                  # labelled per-criterion frames were attached (False when that selection found none)
 
     @classmethod
@@ -219,19 +219,19 @@ def empty(cls) -> "_CompactedTrajectory":
 
     @classmethod
     def for_default(
-        cls, container_s3_uri: Optional[str], tool_result_files: list[tuple[str, str]],
-        compact_s3_uri: Optional[str] = None,
+        cls, container_object_url: Optional[str], tool_result_files: list[tuple[str, str]],
+        compact_object_url: Optional[str] = None,
     ) -> "_CompactedTrajectory":
-        return cls(container_s3_uri=container_s3_uri, tool_result_files=tool_result_files,
-                   compact_s3_uri=compact_s3_uri)
+        return cls(container_object_url=container_object_url, tool_result_files=tool_result_files,
+                   compact_object_url=compact_object_url)
 
     @classmethod
     def for_screenshot(
         cls, inline_text: Optional[str], image_blocks: list[dict],
-        container_s3_uri: Optional[str] = None, per_criterion: bool = False,
+        container_object_url: Optional[str] = None, per_criterion: bool = False,
     ) -> "_CompactedTrajectory":
-        # container_s3_uri = raw file, kept so an agent judge (no inline frames) can still load it.
-        return cls(inline_text=inline_text, image_blocks=image_blocks or [], container_s3_uri=container_s3_uri,
+        # container_object_url = raw file, kept so an agent judge (no inline frames) can still load it.
+        return cls(inline_text=inline_text, image_blocks=image_blocks or [], container_object_url=container_object_url,
                    per_criterion=per_criterion)
 
 
@@ -463,8 +463,8 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 context, "prompt_error", f"Skipped: prompt had error_type={prompt_response.error_type}")
             return context
 
-        per_turn_uris = [u for u in (prompt_response.target_agent_per_turn_trajectory_s3_uris or []) if u]
-        if self.use_trajectory and not (per_turn_uris or prompt_response.agent_trajectory_s3_uri):
+        per_turn_uris = [u for u in (prompt_response.target_agent_per_turn_trajectory_object_urls or []) if u]
+        if self.use_trajectory and not (per_turn_uris or prompt_response.agent_trajectory_object_url):
             raise RuntimeError(f"No trajectory available for prompt_id='{self.prompt_id}'")
 
         overrides = context.metadata.get("user_overrides", {})
@@ -586,7 +586,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 if multiturn:
                     await self._load_per_turn_trajectories(sandbox, per_turn_uris, multiturn_dir, filter_to_apply)
                 else:
-                    await sandbox.write_file_from_object(compacted.container_s3_uri, self.TRAJECTORY_CONTAINER_PATH)
+                    await sandbox.write_file_from_object(compacted.container_object_url, self.TRAJECTORY_CONTAINER_PATH)
                     if compacted.tool_result_files:
                         await self._load_tool_result_files(sandbox, compacted.tool_result_files)
                         compacted.tool_result_files = []
@@ -599,7 +599,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             ]
             criteria_json = json.dumps(criteria_for_prompt, indent=2)
             agent_prompt_text = prompt_response.prompt_text or ""
-            judge_trajectory_s3_uri: Optional[str] = None
+            judge_trajectory_url: Optional[str] = None
             judge_effort = overrides.get("judge_effort") or self.effort
             judge_max_thinking_tokens = overrides.get("judge_max_thinking_tokens") or self.max_thinking_tokens
             loaded_for_judge: list[dict] = []
@@ -639,8 +639,8 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                     trajectory_inline = await asyncio.to_thread(self._merge_per_turn_text, per_turn_uris, filter_to_apply)
                 else:
                     trajectory_inline = compacted.inline_text
-                    if trajectory_inline is None and compacted.container_s3_uri:
-                        trajectory_inline = await asyncio.to_thread(self._read_trajectory_text, compacted.container_s3_uri)
+                    if trajectory_inline is None and compacted.container_object_url:
+                        trajectory_inline = await asyncio.to_thread(self._read_trajectory_text, compacted.container_object_url)
                 eval_prompt = self._build_eval_prompt(
                     agent_prompt=agent_prompt_text,
                     agent_response=prompt_response.response,
@@ -657,7 +657,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                         f"eval prompt {len(eval_prompt)} chars (~{len(eval_prompt) // 4} tokens)."
                     )
 
-            verification_results, judge_output_retries, judge_output_discrepancies, judge_trajectory_s3_uri = (
+            verification_results, judge_output_retries, judge_output_discrepancies, judge_trajectory_url = (
                 await self._run_judge_with_output_retries(
                     eval_prompt=eval_prompt,
                     context=context,
@@ -690,8 +690,8 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                     logger.warning(f"Failed to terminate auto-deployed judge sandbox: {e}")
 
         judge_call_fields: dict = {
-            "compact_trajectory_s3_uri": compacted.compact_s3_uri,
-            "judge_trajectory_s3_uri": judge_trajectory_s3_uri,
+            **dual_keyed("compact_trajectory_s3_uri", "compact_trajectory_object_url", compacted.compact_object_url),
+            **dual_keyed("judge_trajectory_s3_uri", "judge_trajectory_object_url", judge_trajectory_url),
         }
         if judge_output_retries:
             judge_call_fields["judge_output_retries"] = judge_output_retries
@@ -753,10 +753,10 @@ def _compact_trajectory(
         if not self.use_trajectory:
             return _CompactedTrajectory.empty()
 
-        # Fall back to the last per-turn URI when agent_trajectory_s3_uri is unset (the value
+        # Fall back to the last per-turn URI when agent_trajectory_object_url is unset (the value
         # it normally holds). The execute() guard guarantees one source exists; the raise only narrows the type.
-        raw_uri = prompt_response.agent_trajectory_s3_uri or next(
-            (u for u in reversed(prompt_response.target_agent_per_turn_trajectory_s3_uris or []) if u),
+        raw_uri = prompt_response.agent_trajectory_object_url or next(
+            (u for u in reversed(prompt_response.target_agent_per_turn_trajectory_object_urls or []) if u),
             None,
         )
         if raw_uri is None:
@@ -767,15 +767,15 @@ def _compact_trajectory(
             raw_text = self._read_trajectory_text(raw_uri)
             return self._compact_screenshot(raw_text, raw_uri, filter_to_apply, criteria or [], label_ids or {})
 
-        s3_uri, tool_files = raw_uri, []
-        compact_s3_uri: Optional[str] = None
+        object_url, tool_files = raw_uri, []
+        compact_object_url: Optional[str] = None
         if filter_to_apply and ctype == CompactionType.DEFAULT:
-            s3_uri, tool_files = self._filter_trajectory(raw_uri, filter_to_apply)
+            object_url, tool_files = self._filter_trajectory(raw_uri, filter_to_apply)
             # Return the URI; never write it back onto prompt_response. It's an appended
             # item in context.prompt_responses ($addToSet, immutable) — mutating it re-adds
             # a duplicate (hub renders the trajectory twice). Recorded on the result instead.
-            compact_s3_uri = s3_uri
-        return _CompactedTrajectory.for_default(s3_uri, tool_files, compact_s3_uri)
+            compact_object_url = object_url
+        return _CompactedTrajectory.for_default(object_url, tool_files, compact_object_url)
 
     def _compact_screenshot(
         self, raw_text: str, raw_uri: str, tf: TrajectoryFilter, criteria: list[dict],
@@ -803,24 +803,24 @@ def _compact_screenshot(
         else:
             inline_text = strip_trajectory_images(raw_text, attached=attached)
         # A trajectory with no screenshots selects nothing: the judge gets the text alone, with no frame rules.
-        return _CompactedTrajectory.for_screenshot(inline_text, image_blocks, container_s3_uri=raw_uri,
+        return _CompactedTrajectory.for_screenshot(inline_text, image_blocks, container_object_url=raw_uri,
                                                    per_criterion=per_criterion and bool(image_blocks))
 
-    def _filter_trajectory(self, s3_uri: str, trajectory_filter: TrajectoryFilter) -> tuple[str, list[tuple[str, str]]]:
-        """Download trajectory from S3, apply compact filter, re-upload as a compact version.
+    def _filter_trajectory(self, object_url: str, trajectory_filter: TrajectoryFilter) -> tuple[str, list[tuple[str, str]]]:
+        """Download the trajectory, apply the compact filter, and upload the compact version.
 
         Returns:
-            (compact_s3_uri, tool_result_files) where tool_result_files is a list of
+            (compact_object_url, tool_result_files) where tool_result_files is a list of
             (container_path, content_json) pairs for externalized tool results.
         """
         from agent_env.config import get_config
 
         config = get_config()
-        spans = json.loads(config.get_object_store_at(s3_uri).get(s3_uri))
+        spans = json.loads(config.get_object_store_at(object_url).get(object_url))
         filtered, tool_result_files = compact_otel_trajectory(spans, trajectory_filter)
         compact_key = f"{config.get_artifact_key_prefix()}compacted-trajectories/{uuid.uuid4().hex}.json"
-        object_url = config.get_object_store().put(compact_key, json.dumps(filtered).encode(), content_type="application/json")
-        return object_url, tool_result_files
+        compact_url = config.get_object_store().put(compact_key, json.dumps(filtered).encode(), content_type="application/json")
+        return compact_url, tool_result_files
 
     async def _load_tool_result_files(self, sandbox, tool_result_files: list[tuple[str, str]]) -> None:
         """Write externalized tool result files into the agent container."""
@@ -828,15 +828,15 @@ async def _load_tool_result_files(self, sandbox, tool_result_files: list[tuple[s
             await sandbox.write_file_from_text(content, container_path)
         logger.info(f"Loaded {len(tool_result_files)} tool result files into container")
 
-    async def _load_trajectory_into_container(self, sandbox_id: str, s3_uri: str) -> None:
-        """Load trajectory from S3 into an existing agent's container."""
+    async def _load_trajectory_into_container(self, sandbox_id: str, object_url: str) -> None:
+        """Load the trajectory from the object store into an existing agent's container."""
         from agent_env.providers.sandbox_providers.sandbox_provider import get_agent_sandbox_provider
         sandbox = await get_agent_sandbox_provider().get_sandbox(sandbox_id)
-        await sandbox.write_file_from_object(s3_uri, self.TRAJECTORY_CONTAINER_PATH)
+        await sandbox.write_file_from_object(object_url, self.TRAJECTORY_CONTAINER_PATH)
 
-    def _read_trajectory_text(self, s3_uri: str) -> str:
+    def _read_trajectory_text(self, object_url: str) -> str:
         from agent_env.config import get_config
-        return get_config().get_object_store_at(s3_uri).get(s3_uri).decode("utf-8", errors="replace")
+        return get_config().get_object_store_at(object_url).get(object_url).decode("utf-8", errors="replace")
 
     async def _load_per_turn_trajectories(
         self, sandbox, per_turn_uris: list[str], container_dir: str,
@@ -844,7 +844,7 @@ async def _load_per_turn_trajectories(
     ) -> None:
         """Load each turn into the judge container as turn_NN.json under ``container_dir``.
 
-        No filter → direct S3 pull per turn. DEFAULT filter → compact each turn in memory and
+        No filter → direct object-store pull per turn. DEFAULT filter → compact each turn in memory and
         write as text, its externalized tool-result files namespaced ``turn_NN_`` to avoid
         collisions, then loaded into the sandbox. The write methods create ``container_dir``
         inside the container, so no separate mkdir is needed.
@@ -871,7 +871,7 @@ def _merge_per_turn_text(
         self, per_turn_uris: list[str], trajectory_filter: Optional[TrajectoryFilter] = None,
     ) -> str:
         """Concatenate every turn's trajectory text, turn-marked, for the direct-LLM judge
-        (embedded inline; in-memory, no S3 write). A DEFAULT filter compacts each turn; its
+        (embedded inline; in-memory, no object-store write). A DEFAULT filter compacts each turn; its
         externalized files are dropped, as on the single-turn direct-judge path (no container)."""
         parts = []
         for i, uri in enumerate(per_turn_uris, start=1):
@@ -883,13 +883,13 @@ def _merge_per_turn_text(
         return "\n\n".join(parts)
 
     def _filter_trajectory_content(
-        self, s3_uri: str, trajectory_filter: TrajectoryFilter, result_file_prefix: str = "",
+        self, object_url: str, trajectory_filter: TrajectoryFilter, result_file_prefix: str = "",
     ) -> tuple[str, list[tuple[str, str]]]:
-        """DEFAULT-compact one trajectory in memory (no S3 re-upload), returning
+        """DEFAULT-compact one trajectory in memory (no re-upload), returning
         (compacted_json_text, tool_result_files). ``result_file_prefix`` namespaces the
         externalized filenames so per-turn files don't collide. In-memory sibling of
-        ``_filter_trajectory`` (which re-uploads to S3 for the single-file path)."""
-        spans = json.loads(self._read_trajectory_text(s3_uri))
+        ``_filter_trajectory`` (which re-uploads for the single-file path)."""
+        spans = json.loads(self._read_trajectory_text(object_url))
         filtered, tool_result_files = compact_otel_trajectory(
             spans, trajectory_filter, result_file_prefix=result_file_prefix,
         )
@@ -903,7 +903,7 @@ async def _prompt_llm_judge(
         context: TaskStepContext,
         image_blocks: Optional[list[dict]] = None,
     ) -> dict:
-        """Direct LiteLLM judge — no agent boot, no tools. Returns {response, trajectory_s3_uri}.
+        """Direct LiteLLM judge — no agent boot, no tools. Returns {response, trajectory_object_url}.
 
         When ``image_blocks`` are provided the user message becomes multimodal
         (text + images); otherwise it stays a plain string, identical to before."""
@@ -950,7 +950,7 @@ async def _prompt_llm_judge(
                     request_kwargs[ModelParam.RESPONSE_FORMAT] = response_format
                 response = await litellm.acompletion(**request_kwargs)
                 content = response.choices[0].message.content or ""
-                return {"response": content, "trajectory_s3_uri": None}
+                return {"response": content, "trajectory_object_url": None}
             except Exception as exc:
                 last_exc = exc
                 if image_blocks and use_response_format and _looks_like_response_format_error(exc):
@@ -1051,13 +1051,13 @@ async def _invoke_judge_a2a(
                 )
                 body = f"{body}\n{container_logs}" if body else container_logs
             raise RuntimeError(f"Judge A2A task failed ({detail}): {body}")
-        judge_trajectory_s3_uri = await self._fetch_judge_trajectory(
+        judge_trajectory_url = await self._fetch_judge_trajectory(
             judge_a2a_url=judge_a2a_url,
             judge_agent_card=judge_agent_card,
             a2a_server_task_id=task_id,
             sandbox_type=getattr(judge_agent, "sandbox_type", None),
         )
-        return {"response": tr.response_text, "trajectory_s3_uri": judge_trajectory_s3_uri}
+        return {"response": tr.response_text, "trajectory_object_url": judge_trajectory_url}
 
     async def _fetch_judge_trajectory(
         self,
@@ -1149,7 +1149,7 @@ async def _run_judge_with_output_retries(
         spec = self._spec()
         prompt = eval_prompt
         discrepancy_log: list[dict] = []
-        judge_trajectory_s3_uri: Optional[str] = None
+        judge_trajectory_url: Optional[str] = None
 
         if use_agent_judge:
             if not judge_a2a_url:
@@ -1179,12 +1179,12 @@ async def _run_judge_with_output_retries(
                 data = await self._prompt_llm_judge(prompt, model=model, context=context)
 
             logger.info(f"Judge response: {data['response'][:200]}...")
-            judge_trajectory_s3_uri = data.get("trajectory_s3_uri")
+            judge_trajectory_url = data.get("trajectory_object_url")
 
             results, discrepancy = spec.diagnose_response(data["response"], criteria)
             if discrepancy is None:
                 assert results is not None
-                return results, attempt - 1, discrepancy_log, judge_trajectory_s3_uri
+                return results, attempt - 1, discrepancy_log, judge_trajectory_url
 
             logger.warning(
                 "Judge rubric output discrepancy (attempt %d/%d, verifier_id=%s): %s",
diff --git a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
index bac7faf6..e89b4813 100644
--- a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
+++ b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py
@@ -38,7 +38,7 @@
     registered_sandbox_provider_classes,
 )
 from agent_env.store.ids import derive_id, is_local_id, key_segment, validate_local_id
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import TaskStepContext, dual_keyed
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 
 logger = logging.getLogger(__name__)
@@ -362,12 +362,12 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             "stdout_artifact": {
                 "id": stdout_artifact.id,
                 "version": stdout_artifact.version,
-                "s3_url": stdout_artifact.object_url,
+                **dual_keyed("s3_url", "object_url", stdout_artifact.object_url),
             },
             "stderr_artifact": {
                 "id": stderr_artifact.id,
                 "version": stderr_artifact.version,
-                "s3_url": stderr_artifact.object_url,
+                **dual_keyed("s3_url", "object_url", stderr_artifact.object_url),
             },
             "stdout_head": stdout[:_OUTPUT_PREVIEW_CHARS],
             "stderr_head": stderr[:_OUTPUT_PREVIEW_CHARS],
diff --git a/tst/integration/env/task_step/task_steps_test.py b/tst/integration/env/task_step/task_steps_test.py
index 340a9208..dba717c0 100644
--- a/tst/integration/env/task_step/task_steps_test.py
+++ b/tst/integration/env/task_step/task_steps_test.py
@@ -498,13 +498,13 @@ async def assert_universe_data_loaded(session, tools_result, tool_names):
     logger.info(f"PromptAgentTaskStep response: {prompt_response.response[:200]}")
 
     # 16. Verify trajectory was uploaded to the configured object store
-    assert prompt_response.agent_trajectory_s3_uri is not None, "Expected agent_trajectory_s3_uri to be set"
+    assert prompt_response.agent_trajectory_object_url is not None, "Expected agent_trajectory_object_url to be set"
     object_store = get_config().get_object_store()
-    assert _scheme(prompt_response.agent_trajectory_s3_uri) == _scheme(object_store.object_url("probe")), \
-        f"trajectory URI not on the configured object store: {prompt_response.agent_trajectory_s3_uri}"
-    logger.info(f"Trajectory URI: {prompt_response.agent_trajectory_s3_uri}")
+    assert _scheme(prompt_response.agent_trajectory_object_url) == _scheme(object_store.object_url("probe")), \
+        f"trajectory URI not on the configured object store: {prompt_response.agent_trajectory_object_url}"
+    logger.info(f"Trajectory URI: {prompt_response.agent_trajectory_object_url}")
 
-    trajectory = json.loads(object_store.get(prompt_response.agent_trajectory_s3_uri))
+    trajectory = json.loads(object_store.get(prompt_response.agent_trajectory_object_url))
     assert isinstance(trajectory, list), f"Expected trajectory to be a list, got {type(trajectory)}"
     assert len(trajectory) > 0, "Expected non-empty trajectory"
     logger.info(f"Trajectory has {len(trajectory)} messages")
@@ -779,7 +779,7 @@ async def test_an_agent_uploads_its_trajectory_through_a_local_grant(mcp_server_
         await PromptAgentTaskStep(
             id=f"grant-traj-prompt-{suffix}", version=None, prompt="hello through a grant", timeout_seconds=120,
         ).execute(context)
-        uri = context.prompt_responses[-1].agent_trajectory_s3_uri
+        uri = context.prompt_responses[-1].agent_trajectory_object_url
         assert uri, "the agent did not upload its trajectory through the grant"
         assert "hello through a grant" in store.get(uri).decode()
     finally:
@@ -869,9 +869,9 @@ async def exec_in(sandbox, wrapper_cmd, *cmd):
         logger.info(f"CLI e2e prompt response: {prompt_response.response[:300]}")
         assert prompt_response.tool_call_count and prompt_response.tool_call_count > 0, \
             f"Expected agent to make at least one tool call; got {prompt_response.tool_call_count}"
-        assert prompt_response.agent_trajectory_s3_uri, "Expected trajectory to be uploaded"
+        assert prompt_response.agent_trajectory_object_url, "Expected trajectory to be uploaded"
 
-        trajectory_json = get_config().get_object_store().get(prompt_response.agent_trajectory_s3_uri).decode()
+        trajectory_json = get_config().get_object_store().get(prompt_response.agent_trajectory_object_url).decode()
         assert installed["install_path"] in trajectory_json, (
             f"Expected install_path '{installed['install_path']}' to appear in agent trajectory "
             f"(would prove CLI was invoked); not found in {len(trajectory_json)} bytes of trajectory"
@@ -1039,10 +1039,10 @@ def verify_instance_on_step(i, total, step, context, duration):
     logger.info(f"Task prompt response: {prompt_response.response}")
 
     # Verify the trajectory exists in the configured object store
-    assert prompt_response.agent_trajectory_s3_uri is not None, "Expected agent_trajectory_s3_uri to be set"
-    trajectory = json.loads(get_config().get_object_store().get(prompt_response.agent_trajectory_s3_uri))
+    assert prompt_response.agent_trajectory_object_url is not None, "Expected agent_trajectory_object_url to be set"
+    trajectory = json.loads(get_config().get_object_store().get(prompt_response.agent_trajectory_object_url))
     assert isinstance(trajectory, list) and len(trajectory) > 0, "Expected non-empty trajectory list"
-    logger.info(f"Task trajectory: {len(trajectory)} messages at {prompt_response.agent_trajectory_s3_uri}")
+    logger.info(f"Task trajectory: {len(trajectory)} messages at {prompt_response.agent_trajectory_object_url}")
 
     # Verify both verifier results coexist in context.metadata["verifications"]
     verifications = context.metadata.get("verifications", {})
@@ -1632,7 +1632,7 @@ async def test_multi_turn_done_signal_terminates_early():
 
     assert context.prompt_responses, "expected at least one PromptResponse"
     pr = context.prompt_responses[-1]
-    traj_uris = pr.target_agent_per_turn_trajectory_s3_uris
+    traj_uris = pr.target_agent_per_turn_trajectory_object_urls
     prompts = pr.source_agent_per_turn_prompt_parts
     assert traj_uris is not None and len(traj_uris) >= 1
     assert prompts is not None and len(prompts) == len(traj_uris), \
@@ -1645,8 +1645,8 @@ async def test_multi_turn_done_signal_terminates_early():
                 f"trajectory URI not on the configured object store: {uri!r}"
 
     last_successful = next((u for u in reversed(traj_uris) if u is not None), None)
-    assert pr.agent_trajectory_s3_uri == last_successful, \
-        "agent_trajectory_s3_uri should equal the last successful per-turn URI"
+    assert pr.agent_trajectory_object_url == last_successful, \
+        "agent_trajectory_object_url should equal the last successful per-turn URI"
     assert pr.prompt_text == _MULTI_TURN_INITIAL_PROMPT
     assert pr.response, "expected non-empty final response"
 
@@ -1676,7 +1676,7 @@ async def test_multi_turn_max_turns_cap_terminates_loop():
     context = await task.run(instance_id=instance_id)
 
     pr = context.prompt_responses[-1]
-    traj_uris = pr.target_agent_per_turn_trajectory_s3_uris
+    traj_uris = pr.target_agent_per_turn_trajectory_object_urls
     prompts = pr.source_agent_per_turn_prompt_parts
     assert traj_uris is not None and len(traj_uris) == max_turns, \
         f"expected exactly {max_turns} turns (cap), got {len(traj_uris) if traj_uris else None}"
diff --git a/tst/unit/env/task_step/test_context_ops.py b/tst/unit/env/task_step/test_context_ops.py
index 64c49e10..5078e5fc 100644
--- a/tst/unit/env/task_step/test_context_ops.py
+++ b/tst/unit/env/task_step/test_context_ops.py
@@ -381,8 +381,8 @@ def test_in_place_mutation_of_appended_list_item_duplicates_in_doc():
     """Regression: mutating an item already present in a top-level $addToSet list
     re-adds it as a NEW element instead of updating in place.
 
-    Reproduces the production bug where ``rubrics_verifier`` set
-    ``compact_trajectory_s3_uri`` on a ``prompt_response`` that ``prompt_agent``
+    Reproduces the production bug where ``rubrics_verifier`` set the compacted
+    trajectory's URL on a ``prompt_response`` that ``prompt_agent``
     had already appended to ``context.prompt_responses``. The mutated item is
     BSON-unequal to the stored one, so the diff emits it via ``$addToSet`` and
     the instance ends up with two near-identical prompt_responses — the hub then
@@ -395,7 +395,7 @@ def test_in_place_mutation_of_appended_list_item_duplicates_in_doc():
 
     # Model the (now-fixed) in-place mutation: same logical item, one field changed.
     mutated = _response("p1")
-    mutated.compact_trajectory_s3_uri = "s3://bucket/compact.json"
+    mutated.tool_call_count = (mutated.tool_call_count or 0) + 1
     post = TaskStepContext(prompt_responses=[mutated])
 
     ops = build_context_update_ops(pre, post)
diff --git a/tst/unit/task_step/run_context_object_url_test.py b/tst/unit/task_step/run_context_object_url_test.py
new file mode 100644
index 00000000..63f12dd8
--- /dev/null
+++ b/tst/unit/task_step/run_context_object_url_test.py
@@ -0,0 +1,112 @@
+"""The S3-named run-context keys get neutral twins.
+
+Stored instances carry the S3-named keys, so they are read forever and written beside the neutral ones while workers
+on older versions read only them. The S3-named ``PromptResponse`` keywords still work, with a warning."""
+
+import dataclasses
+import logging
+import warnings
+
+import pytest
+
+from agent_env.cli.task.run import _format_prompt_response, _format_verifier_output
+from agent_env.task_step.context import PromptResponse, TaskStepContext
+from agent_env.task_step.context_ops import build_context_update_ops
+
+_URL, _PREFIX, _TURNS = "mem://b/t/2.json", "mem://b/t/", ["mem://b/t/1.json", None, "mem://b/t/2.json"]
+_PAIRS = {
+    "agent_trajectory_s3_uri": ("agent_trajectory_object_url", _URL),
+    "agent_trajectory_s3_prefix": ("agent_trajectory_object_prefix", _PREFIX),
+    "target_agent_per_turn_trajectory_s3_uris": ("target_agent_per_turn_trajectory_object_urls", _TURNS),
+}
+
+
+def _neutral(**kw) -> PromptResponse:
+    return PromptResponse(prompt_id="p1", response="done", agent_trajectory_object_url=_URL,
+                          agent_trajectory_object_prefix=_PREFIX, target_agent_per_turn_trajectory_object_urls=_TURNS,
+                          **kw)
+
+
+def test_a_response_built_with_the_neutral_keywords_carries_both_spellings_without_a_warning():
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        response = _neutral()
+    dumped = dataclasses.asdict(response)
+    assert all(dumped[legacy] == dumped[neutral] == value for legacy, (neutral, value) in _PAIRS.items())
+
+
+def test_a_persisted_response_carries_both_keys():
+    pre = TaskStepContext()
+    post = TaskStepContext(prompt_responses=[_neutral()])
+    [pushed] = build_context_update_ops(pre, post).add_to_sets["context.prompt_responses"]
+    assert all(pushed[legacy] == pushed[neutral] == value for legacy, (neutral, value) in _PAIRS.items())
+
+
+@pytest.mark.parametrize("spelling", ["legacy", "neutral"])
+def test_a_stored_response_loads_keyed_either_way(spelling):
+    keyed = {(legacy if spelling == "legacy" else neutral): value for legacy, (neutral, value) in _PAIRS.items()}
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        response = PromptResponse.from_dict({"prompt_id": "p1", "response": "done", **keyed})
+    assert dataclasses.asdict(response) == dataclasses.asdict(_neutral())
+
+
+def test_the_legacy_key_wins_when_the_two_disagree():
+    """Only a raw-doc writer outside the model can make them differ, and today's know only the legacy key."""
+    response = PromptResponse.from_dict({"prompt_id": "p1", "response": "done", "agent_trajectory_s3_uri": _URL,
+                                         "agent_trajectory_object_url": "mem://b/stale.json",
+                                         "target_agent_per_turn_trajectory_s3_uris": [],
+                                         "target_agent_per_turn_trajectory_object_urls": ["mem://b/stale.json"]})
+    assert response.agent_trajectory_object_url == _URL
+    assert response.target_agent_per_turn_trajectory_object_urls == []
+
+
+def test_a_legacy_key_holding_none_falls_through_to_the_neutral_one():
+    response = PromptResponse.from_dict({"prompt_id": "p1", "response": "done", "agent_trajectory_s3_uri": None,
+                                         "agent_trajectory_object_url": _URL})
+    assert response.agent_trajectory_object_url == response.agent_trajectory_s3_uri == _URL
+
+
+def test_the_legacy_keywords_still_work_and_are_counted(caplog):
+    with caplog.at_level(logging.WARNING, logger="agent_env.utils.deprecation"), \
+            pytest.warns(DeprecationWarning) as caught:
+        response = PromptResponse(prompt_id="p1", response="done", agent_trajectory_s3_uri=_URL,
+                                  agent_trajectory_s3_prefix=_PREFIX, target_agent_per_turn_trajectory_s3_uris=_TURNS)
+    assert dataclasses.asdict(response) == dataclasses.asdict(_neutral())
+    symbols = [f"PromptResponse({legacy}=)" for legacy in _PAIRS]
+    assert [str(w.message).split(" is deprecated")[0] for w in caught] == symbols
+    assert [r.deprecated_symbol for r in caplog.records if getattr(r, "event", None) == "agent_env_deprecated_symbol"] \
+        == symbols
+
+
+def test_the_two_spellings_disagreeing_is_an_error_and_agreeing_is_a_copy():
+    with pytest.raises(ValueError, match="agent_trajectory_s3_uri='mem://b/other.json' and agent_trajectory_object_url"):
+        _neutral(agent_trajectory_s3_uri="mem://b/other.json")
+    response = _neutral()
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", DeprecationWarning)
+        assert dataclasses.replace(response) == response
+        assert PromptResponse(**dataclasses.asdict(response)) == response
+
+
+def test_the_run_output_prints_the_neutral_labels():
+    printed = _format_prompt_response(_neutral())
+    assert f"agent_trajectory_object_url: {_URL}" in printed and f"agent_trajectory_object_prefix: {_PREFIX}" in printed
+    assert "s3" not in printed
+
+
+@pytest.mark.parametrize("spelling", ["legacy", "neutral"])
+def test_the_verifier_output_reads_metadata_keyed_either_way(spelling):
+    def keyed(legacy, neutral, value):
+        return {legacy if spelling == "legacy" else neutral: value}
+
+    vdata = {**keyed("compact_trajectory_s3_uri", "compact_trajectory_object_url", "mem://b/compact.json"),
+             **keyed("judge_trajectory_s3_uri", "judge_trajectory_object_url", "mem://b/judge.json"),
+             "stdout_artifact": {"id": "out", "version": 1, **keyed("s3_url", "object_url", "mem://b/stdout.txt")},
+             "stderr_artifact": {"id": "err", "version": 1, **keyed("s3_url", "object_url", "mem://b/stderr.txt")}}
+    assert _format_verifier_output("v", vdata) == [
+        "compact_trajectory_object_url (verifier_id=v): mem://b/compact.json",
+        "judge_trajectory_object_url (verifier_id=v): mem://b/judge.json",
+        "stdout_artifact (verifier_id=v): out v1 -> mem://b/stdout.txt",
+        "stderr_artifact (verifier_id=v): err v1 -> mem://b/stderr.txt",
+    ]
diff --git a/tst/unit/task_step/test_prompt_agent.py b/tst/unit/task_step/test_prompt_agent.py
index d5c53d0d..6f821ae7 100644
--- a/tst/unit/task_step/test_prompt_agent.py
+++ b/tst/unit/task_step/test_prompt_agent.py
@@ -263,7 +263,7 @@ def trajectory(request: httpx.Request) -> httpx.Response:
 
     result = await step.execute(context)
 
-    uri = result.prompt_responses[-1].agent_trajectory_s3_uri
+    uri = result.prompt_responses[-1].agent_trajectory_object_url
     assert trajectory_requests == [{"task_id": "task-1"}]
     assert store.get_object_key(uri).startswith("prompt_agent_trajectories/trajectory-")
     assert json.loads(store.get(uri)) == [{"type": "echo", "output": "Echo: hi"}]
diff --git a/tst/unit/task_step/test_prompt_agent_snapshots.py b/tst/unit/task_step/test_prompt_agent_snapshots.py
index a66287a3..cdc01624 100644
--- a/tst/unit/task_step/test_prompt_agent_snapshots.py
+++ b/tst/unit/task_step/test_prompt_agent_snapshots.py
@@ -154,7 +154,7 @@ def unexpected_upload(*args, **kwargs):
     row = await _series()._capture_never_raising(context(), is_final=False)
 
     assert row["capture_status"] == "ok"
-    assert row["trajectory_s3_uri"] == direct_url
+    assert row["trajectory_s3_uri"] == row["trajectory_object_url"] == direct_url
 
 
 async def _until(pred, timeout=2.0, what="condition"):
@@ -439,7 +439,7 @@ async def test_max_snapshots_bounds_interior_captures_but_reserves_the_final_one
     series = _series(interval_seconds=0.01, max_snapshots=2)
     ctx = context()
     ctx.prompt_responses.append(
-        PromptResponse(prompt_id="p1", response="ok", agent_trajectory_s3_uri="s3://b/t.json")
+        PromptResponse(prompt_id="p1", response="ok", agent_trajectory_object_url="s3://b/t.json")
     )
     series.start(ctx)
     await _until(
@@ -534,23 +534,23 @@ async def test_an_unsupported_trajectory_read_still_yields_a_gradable_row(monkey
     assert row["capture_status"] == "partial"
     assert row["capture_reason"] == "trajectory_context_unsupported"
     # Nulls are stripped on append, not here — this is the raw row.
-    assert row["bundle_object_url"] and row["trajectory_s3_uri"] is None
+    assert row["bundle_object_url"] and row["trajectory_s3_uri"] is row["trajectory_object_url"] is None
 
 
 @pytest.mark.asyncio
 async def test_the_final_row_reads_the_trajectory_cumulatively_like_the_others(monkeypatch):
-    """`PromptResponse.agent_trajectory_s3_uri` is only the LAST turn's, so using
+    """`PromptResponse.agent_trajectory_object_url` is only the LAST turn's, so using
     it here would make the final point narrower than its predecessors."""
     _install(monkeypatch)
     ctx = context()
     ctx.prompt_responses.append(
         PromptResponse(prompt_id="p1", response="done", a2a_context_id="cid-1",
-                       agent_trajectory_s3_uri="s3://b/last-turn-only.json")
+                       agent_trajectory_object_url="s3://b/last-turn-only.json")
     )
     await _series().finish(ctx)
 
     (row,) = _rows(ctx)
-    assert row["trajectory_s3_uri"] == f"{TRAJ_PREFIX}trajectory-x.json"
+    assert row["trajectory_s3_uri"] == row["trajectory_object_url"] == f"{TRAJ_PREFIX}trajectory-x.json"
     assert row["capture_status"] == "ok"
 
 
@@ -561,12 +561,12 @@ async def test_the_final_row_falls_back_to_the_recorded_uri(monkeypatch):
     ctx = context()
     ctx.prompt_responses.append(
         PromptResponse(prompt_id="p1", response="done", a2a_context_id="cid-1",
-                       agent_trajectory_s3_uri="s3://b/t.json")
+                       agent_trajectory_object_url="s3://b/t.json")
     )
     await _series().finish(ctx)
 
     (row,) = _rows(ctx)
-    assert row["trajectory_s3_uri"] == "s3://b/t.json"
+    assert row["trajectory_s3_uri"] == row["trajectory_object_url"] == "s3://b/t.json"
     assert row["capture_status"] == "partial"
     assert row["capture_reason"] == "trajectory_session_missing"
 
@@ -579,12 +579,12 @@ async def test_a_sibling_attempts_trajectory_is_never_borrowed(monkeypatch):
     ctx = context()
     ctx.prompt_responses.append(
         PromptResponse(prompt_id="p1", response="old", a2a_context_id="cid-PREVIOUS",
-                       agent_trajectory_s3_uri="s3://b/previous-attempt.json")
+                       agent_trajectory_object_url="s3://b/previous-attempt.json")
     )
     await _series().finish(ctx)
 
     (row,) = _rows(ctx)
-    assert "trajectory_s3_uri" not in row, "matched on prompt_id alone, not the conversation"
+    assert {"trajectory_s3_uri", "trajectory_object_url"}.isdisjoint(row), "matched on prompt_id alone, not the conversation"
     assert row["capture_reason"] == "trajectory_session_missing; trajectory_missing"
 
 
@@ -734,7 +734,7 @@ async def test_an_undeployed_env_id_is_a_reason_not_a_raise(monkeypatch):
     _install(monkeypatch)
     ctx = context()  # no deployed envs
     ctx.prompt_responses.append(
-        PromptResponse(prompt_id="p1", response="ok", agent_trajectory_s3_uri="s3://b/t.json")
+        PromptResponse(prompt_id="p1", response="ok", agent_trajectory_object_url="s3://b/t.json")
     )
     await _series(env_id="env-missing").finish(ctx)
 
@@ -836,7 +836,7 @@ async def test_teardown_outlasts_an_interior_capture_that_times_out(monkeypatch)
     ctx = context()
     ctx.prompt_responses.append(
         PromptResponse(prompt_id="p1", response="ok", a2a_context_id="cid-1",
-                       agent_trajectory_s3_uri="s3://b/t.json")
+                       agent_trajectory_object_url="s3://b/t.json")
     )
     series.start(ctx)
     await _until(lambda: series._lock.locked(), what="the interior capture")
@@ -1040,7 +1040,7 @@ async def exploding_conversation(self, context, *, a2a_context_id=None, series=N
     monkeypatch.setattr(PromptAgentTaskStep, "_execute_conversation", exploding_conversation)
     ctx = context()
     ctx.prompt_responses.append(
-        PromptResponse(prompt_id=step.prompt_id, response="", agent_trajectory_s3_uri="s3://b/t")
+        PromptResponse(prompt_id=step.prompt_id, response="", agent_trajectory_object_url="s3://b/t")
     )
 
     # A long run that died is the most interesting point for a curve, so the
diff --git a/tst/unit/task_step/verifiers/run_container_unit_tests_verifier_test.py b/tst/unit/task_step/verifiers/run_container_unit_tests_verifier_test.py
index 449b4d03..d022ee97 100644
--- a/tst/unit/task_step/verifiers/run_container_unit_tests_verifier_test.py
+++ b/tst/unit/task_step/verifiers/run_container_unit_tests_verifier_test.py
@@ -235,6 +235,10 @@ def record(text, artifact_id, description, object_url):
     outputs = get_config().get_object_store().object_url("fx/verifier-outputs/scrape/")
     assert [url.rsplit("/", 1)[-1] for url in uploaded] == ["stdout.txt", "stderr.txt"]
     assert all(url.startswith(outputs) for url in uploaded)
+    [recorded] = ctx.metadata["verifications"].values()
+    assert [(recorded[k]["s3_url"], recorded[k]["object_url"]) for k in ("stdout_artifact", "stderr_artifact")] == [
+        (url, url) for url in uploaded
+    ]
 
 
 def test_stdout_and_stderr_upload_off_the_event_loop(local_stores, monkeypatch):
diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_retry.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_retry.py
index 5ea6a73e..2f89bea2 100644
--- a/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_retry.py
+++ b/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_retry.py
@@ -41,12 +41,12 @@ async def test_retries_then_succeeds_on_second_attempt(monkeypatch):
     async def fake_prompt_llm_judge(prompt: str, *, model: str, context: TaskStepContext) -> dict:
         prompts.append(prompt)
         if len(prompts) == 1:
-            return {"response": bad, "trajectory_s3_uri": None}
-        return {"response": _valid_response(["c1", "c2"]), "trajectory_s3_uri": None}
+            return {"response": bad, "trajectory_object_url": "mem://judge/rejected.json"}
+        return {"response": _valid_response(["c1", "c2"]), "trajectory_object_url": "mem://judge/accepted.json"}
 
     monkeypatch.setattr(verifier, "_prompt_llm_judge", fake_prompt_llm_judge)
 
-    results, retries, discrepancies, _ = await verifier._run_judge_with_output_retries(
+    results, retries, discrepancies, judge_trajectory_url = await verifier._run_judge_with_output_retries(
         eval_prompt="Evaluate the agent.",
         criteria=verifier.criteria,
         context=TaskStepContext(),
@@ -61,6 +61,7 @@ async def fake_prompt_llm_judge(prompt: str, *, model: str, context: TaskStepCon
 
     assert len(results) == 2
     assert retries == 1
+    assert judge_trajectory_url == "mem://judge/accepted.json"
     assert len(discrepancies) == 1
     assert discrepancies[0]["missing_ids"] == ["c2"]
     assert len(prompts) == 2
@@ -77,7 +78,7 @@ async def test_raises_after_max_retries(monkeypatch):
     monkeypatch.setattr(
         verifier,
         "_prompt_llm_judge",
-        AsyncMock(return_value={"response": bad, "trajectory_s3_uri": None}),
+        AsyncMock(return_value={"response": bad, "trajectory_object_url": None}),
     )
 
     with pytest.raises(ValueError, match=r"expected 2 criteria"):
@@ -109,8 +110,8 @@ async def test_a2a_path_retries_with_correction(monkeypatch):
     async def fake_invoke(*, eval_prompt: str, judge_a2a_url: str, judge_agent_card: dict, judge_agent) -> dict:
         prompts.append(eval_prompt)
         if len(prompts) == 1:
-            return {"response": bad, "trajectory_s3_uri": None}
-        return {"response": _valid_response(["c1", "c2"]), "trajectory_s3_uri": None}
+            return {"response": bad, "trajectory_object_url": None}
+        return {"response": _valid_response(["c1", "c2"]), "trajectory_object_url": None}
 
     monkeypatch.setattr(verifier, "_invoke_judge_a2a", fake_invoke)
 
@@ -142,8 +143,8 @@ async def test_retries_on_malformed_json(monkeypatch):
     async def fake_prompt_llm_judge(prompt: str, *, model: str, context: TaskStepContext) -> dict:
         prompts.append(prompt)
         if len(prompts) == 1:
-            return {"response": "Sure! ```json\nnot valid", "trajectory_s3_uri": None}
-        return {"response": _valid_response(["c1", "c2"]), "trajectory_s3_uri": None}
+            return {"response": "Sure! ```json\nnot valid", "trajectory_object_url": None}
+        return {"response": _valid_response(["c1", "c2"]), "trajectory_object_url": None}
 
     monkeypatch.setattr(verifier, "_prompt_llm_judge", fake_prompt_llm_judge)
 
diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py
index b6605ebc..6a663fe1 100644
--- a/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py
+++ b/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py
@@ -421,5 +421,5 @@ async def fake_fetch(
 
     assert result == {
         "response": "ok",
-        "trajectory_s3_uri": "s3://bucket/judge_trajectories/verifier_id=verifier-test/trajectory-task-1.json",
+        "trajectory_object_url": "s3://bucket/judge_trajectories/verifier_id=verifier-test/trajectory-task-1.json",
     }
diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_mistakes_entry.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_mistakes_entry.py
index c7147eb2..cfdf70b4 100644
--- a/tst/unit/task_step/verifiers/test_rubrics_verifier_mistakes_entry.py
+++ b/tst/unit/task_step/verifiers/test_rubrics_verifier_mistakes_entry.py
@@ -65,7 +65,7 @@ def _verifier(**kw) -> RubricsVerifierTaskStep:
 
 
 def _ctx(**pr) -> TaskStepContext:
-    base = dict(prompt_id="p1", response="done", prompt_text="reserve", agent_trajectory_s3_uri="s3://b/raw.json")
+    base = dict(prompt_id="p1", response="done", prompt_text="reserve", agent_trajectory_object_url="s3://b/raw.json")
     base.update(pr)
     return TaskStepContext(prompt_responses=[PromptResponse(**base)])
 
@@ -192,8 +192,10 @@ async def test_execute_writes_the_mistakes_entry_from_the_same_call_under_the_de
     # The second entry: trajectory_mistakes shape, the same pointers, provenance.
     assert second["format"] == "trajectory_mistakes" and "judge_output_format" not in second
     assert second["source_verifier_id"] == "rubric"
-    assert second["judge_trajectory_s3_uri"] == main["judge_trajectory_s3_uri"] == "s3://judge/trajectory.json"
-    assert second["compact_trajectory_s3_uri"] == main["compact_trajectory_s3_uri"]
+    for legacy, neutral in (("judge_trajectory_s3_uri", "judge_trajectory_object_url"),
+                            ("compact_trajectory_s3_uri", "compact_trajectory_object_url")):
+        assert second[legacy] == second[neutral] == main[legacy] == main[neutral]
+    assert main["judge_trajectory_object_url"] == "s3://judge/trajectory.json"
     assert not {"reasoning", "checks"} & set(second)
 
     # ...its rows are exactly what the standalone parser builds for the same findings + derived coverage
diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_multiturn.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_multiturn.py
index 511a120f..16c86db5 100644
--- a/tst/unit/task_step/verifiers/test_rubrics_verifier_multiturn.py
+++ b/tst/unit/task_step/verifiers/test_rubrics_verifier_multiturn.py
@@ -1,7 +1,7 @@
 """Unit tests for multi-turn trajectory handling in rubrics_verifier.
 
 The verifier judges every turn of a multi-turn prompt_agent run, fed from the
-per-turn URIs already on the PromptResponse (target_agent_per_turn_trajectory_s3_uris):
+per-turn URIs already on the PromptResponse (target_agent_per_turn_trajectory_object_urls):
 
 - direct LLM judge  -> turns concatenated inline (``_merge_per_turn_text``)
 - agent judge       -> turns loaded into a container dir (``_load_per_turn_trajectories``)
@@ -82,8 +82,8 @@ async def test_direct_judge_multiturn_inlines_all_turns_marked(monkeypatch):
     monkeypatch.setattr(v, "_read_trajectory_text", lambda uri: fixtures[uri])
     captured = _capture_judge(monkeypatch, v)
     ctx = TaskStepContext(prompt_responses=[_pr(
-        target_agent_per_turn_trajectory_s3_uris=["s3://t/1.json", "s3://t/2.json"],
-        agent_trajectory_s3_uri="s3://t/2.json",
+        target_agent_per_turn_trajectory_object_urls=["s3://t/1.json", "s3://t/2.json"],
+        agent_trajectory_object_url="s3://t/2.json",
     )])
 
     await v.execute(ctx)
@@ -115,23 +115,23 @@ def test_merge_per_turn_text_default_filter_compacts_each_turn(monkeypatch):
 
 def test_compact_trajectory_prefers_agent_trajectory_uri():
     v = _verifier()
-    pr = _pr(agent_trajectory_s3_uri="s3://a/last.json",
-             target_agent_per_turn_trajectory_s3_uris=["s3://a/1.json", "s3://a/last.json"])
-    assert v._compact_trajectory(pr, None).container_s3_uri == "s3://a/last.json"
+    pr = _pr(agent_trajectory_object_url="s3://a/last.json",
+             target_agent_per_turn_trajectory_object_urls=["s3://a/1.json", "s3://a/last.json"])
+    assert v._compact_trajectory(pr, None).container_object_url == "s3://a/last.json"
 
 
 def test_compact_trajectory_falls_back_to_last_per_turn_uri():
     v = _verifier()
-    pr = _pr(agent_trajectory_s3_uri=None,
-             target_agent_per_turn_trajectory_s3_uris=["s3://a/1.json", "s3://a/2.json"])
-    assert v._compact_trajectory(pr, None).container_s3_uri == "s3://a/2.json"
+    pr = _pr(agent_trajectory_object_url=None,
+             target_agent_per_turn_trajectory_object_urls=["s3://a/1.json", "s3://a/2.json"])
+    assert v._compact_trajectory(pr, None).container_object_url == "s3://a/2.json"
 
 
 def test_compact_trajectory_fallback_skips_none_entries():
     v = _verifier()
-    pr = _pr(agent_trajectory_s3_uri=None,
-             target_agent_per_turn_trajectory_s3_uris=["s3://a/1.json", None])
-    assert v._compact_trajectory(pr, None).container_s3_uri == "s3://a/1.json"
+    pr = _pr(agent_trajectory_object_url=None,
+             target_agent_per_turn_trajectory_object_urls=["s3://a/1.json", None])
+    assert v._compact_trajectory(pr, None).container_object_url == "s3://a/1.json"
 
 
 # --- gate / guard fail-fast (via execute; raises before judge/sandbox) --------
@@ -141,8 +141,8 @@ async def test_screenshot_plus_multiturn_raises():
     v = _verifier(trajectory_filter=TrajectoryFilter(
         compaction_type=CompactionType.SCREENSHOT, screenshot_last_n=3))
     ctx = TaskStepContext(prompt_responses=[_pr(
-        target_agent_per_turn_trajectory_s3_uris=["s3://a/1.json", "s3://a/2.json"],
-        agent_trajectory_s3_uri="s3://a/2.json",
+        target_agent_per_turn_trajectory_object_urls=["s3://a/1.json", "s3://a/2.json"],
+        agent_trajectory_object_url="s3://a/2.json",
     )])
     with pytest.raises(RuntimeError, match="SCREENSHOT"):
         await v.execute(ctx)
@@ -152,7 +152,7 @@ async def test_screenshot_plus_multiturn_raises():
 async def test_guard_raises_when_all_trajectory_sources_empty():
     v = _verifier()
     ctx = TaskStepContext(prompt_responses=[_pr(
-        target_agent_per_turn_trajectory_s3_uris=[None], agent_trajectory_s3_uri=None,
+        target_agent_per_turn_trajectory_object_urls=[None], agent_trajectory_object_url=None,
     )])
     with pytest.raises(RuntimeError, match="No trajectory available"):
         await v.execute(ctx)
@@ -171,8 +171,8 @@ def _no_merge(*a, **k):
     monkeypatch.setattr(v, "_merge_per_turn_text", _no_merge)
     captured = _capture_judge(monkeypatch, v)
     ctx = TaskStepContext(prompt_responses=[_pr(
-        target_agent_per_turn_trajectory_s3_uris=["s3://t/only.json"],
-        agent_trajectory_s3_uri="s3://t/only.json",
+        target_agent_per_turn_trajectory_object_urls=["s3://t/only.json"],
+        agent_trajectory_object_url="s3://t/only.json",
     )])
 
     await v.execute(ctx)
@@ -234,7 +234,7 @@ def read(uri):
     monkeypatch.setattr(v, "_read_trajectory_text", read)
     _capture_judge(monkeypatch, v)
     uris = [f"s3://t/{i}.json" for i in range(1, turns + 1)]
-    ctx = TaskStepContext(prompt_responses=[_pr(target_agent_per_turn_trajectory_s3_uris=uris, agent_trajectory_s3_uri=uris[-1])])
+    ctx = TaskStepContext(prompt_responses=[_pr(target_agent_per_turn_trajectory_object_urls=uris, agent_trajectory_object_url=uris[-1])])
 
     await v.execute(ctx)
 
@@ -279,7 +279,7 @@ async def test_default_compaction_reads_compacts_and_rewrites_off_the_event_loop
     v = _verifier(trajectory_filter=TrajectoryFilter())
     _capture_judge(monkeypatch, v)
 
-    await v.execute(TaskStepContext(prompt_responses=[_pr(agent_trajectory_s3_uri="s3://t/raw.json")]))
+    await v.execute(TaskStepContext(prompt_responses=[_pr(agent_trajectory_object_url="s3://t/raw.json")]))
 
     assert len(store.on_loop) >= 2 and not any(store.on_loop)
     assert compactions == [False]
@@ -300,7 +300,7 @@ async def test_screenshot_compaction_reads_and_parses_off_the_event_loop(monkeyp
     v = _verifier(trajectory_filter=TrajectoryFilter(compaction_type=CompactionType.SCREENSHOT, screenshot_last_n=1))
     _capture_judge(monkeypatch, v)
 
-    await v.execute(TaskStepContext(prompt_responses=[_pr(agent_trajectory_s3_uri="s3://t/raw.json")]))
+    await v.execute(TaskStepContext(prompt_responses=[_pr(agent_trajectory_object_url="s3://t/raw.json")]))
 
     assert store.on_loop == [False] and parses == [False]
 
@@ -353,7 +353,7 @@ async def judge(*, eval_prompt, **kwargs):
 
     monkeypatch.setattr(v, "_run_judge_with_output_retries", judge)
 
-    await v.execute(TaskStepContext(prompt_responses=[_pr(agent_trajectory_s3_uri="s3://t/raw.json")]))
+    await v.execute(TaskStepContext(prompt_responses=[_pr(agent_trajectory_object_url="s3://t/raw.json")]))
 
     assert held and alive == [False]
 
diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_per_criterion.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_per_criterion.py
index d0c85d01..65fd9228 100644
--- a/tst/unit/task_step/verifiers/test_rubrics_verifier_per_criterion.py
+++ b/tst/unit/task_step/verifiers/test_rubrics_verifier_per_criterion.py
@@ -166,7 +166,7 @@ def _verifier(**kw) -> RubricsVerifierTaskStep:
 
 
 def _ctx(**pr) -> TaskStepContext:
-    base = dict(prompt_id="p1", response="done", prompt_text="reserve", agent_trajectory_s3_uri="s3://b/raw.json")
+    base = dict(prompt_id="p1", response="done", prompt_text="reserve", agent_trajectory_object_url="s3://b/raw.json")
     base.update(pr)
     return TaskStepContext(prompt_responses=[PromptResponse(**base)])
 
diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_screenshots.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_screenshots.py
index be2734da..9ffaf1e4 100644
--- a/tst/unit/task_step/verifiers/test_rubrics_verifier_screenshots.py
+++ b/tst/unit/task_step/verifiers/test_rubrics_verifier_screenshots.py
@@ -300,7 +300,7 @@ async def fake_run(*, eval_prompt, context, model, image_blocks=None, **kwargs):
 
     ctx = TaskStepContext(prompt_responses=[
         PromptResponse(prompt_id="prompt-1", response="done", prompt_text="do it",
-                       agent_trajectory_s3_uri=raw_uri),
+                       agent_trajectory_object_url=raw_uri),
     ])
 
     await verifier.execute(ctx)
@@ -341,7 +341,7 @@ async def fake_run(*, eval_prompt, context, model, image_blocks=None, **kwargs):
 
     ctx = TaskStepContext(prompt_responses=[
         PromptResponse(prompt_id="prompt-1", response="done", prompt_text="do it",
-                       agent_trajectory_s3_uri=raw_uri),
+                       agent_trajectory_object_url=raw_uri),
     ])
     await verifier.execute(ctx)
 
@@ -380,7 +380,7 @@ async def fake_run(*, eval_prompt, context, model, image_blocks=None, **kwargs):
 
     ctx = TaskStepContext(
         prompt_responses=[PromptResponse(prompt_id="prompt-1", response="done",
-                                         prompt_text="do it", agent_trajectory_s3_uri=raw_uri)],
+                                         prompt_text="do it", agent_trajectory_object_url=raw_uri)],
         metadata={"user_overrides": {"apply_trajectory_filter": False}},
     )
     await verifier.execute(ctx)
diff --git a/tst/unit/task_step/verifiers/test_trajectory_mistakes_format.py b/tst/unit/task_step/verifiers/test_trajectory_mistakes_format.py
index b75e11be..48700cad 100644
--- a/tst/unit/task_step/verifiers/test_trajectory_mistakes_format.py
+++ b/tst/unit/task_step/verifiers/test_trajectory_mistakes_format.py
@@ -226,7 +226,7 @@ async def fake_run(*, eval_prompt, criteria, **kw):
         return ([{**c, "score": 1.0, "result": True, "justification": "j"} for c in criteria], 0, [], None)
     monkeypatch.setattr(v, "_run_judge_with_output_retries", fake_run)
     ctx = TaskStepContext(prompt_responses=[PromptResponse(prompt_id="p1", response="done", prompt_text="reserve",
-                                                           agent_trajectory_s3_uri="s3://b/raw.json")])
+                                                           agent_trajectory_object_url="s3://b/raw.json")])
     await v.execute(ctx)
     text = captured["eval_prompt"]
     assert text.count("## Grading policy") == 1 and text.count(_AUTOCOMPLETE_POLICY) == 1