feat(context): object-store names for run-context trajectory keys

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:

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,17 +155,37 @@ 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:
         return cls(
             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"),

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,
         }

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,