From fcc92cd00db35d65b394a6ab527874675097663b Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Mon, 5 Oct 2026 08:33:01 -0700
Subject: [PATCH 1/3] refactor(a2a)!: remove the S3 transfer forms

agent-env now moves a skill bundle, snapshot or changelog only through transfer
grants. The s3_prefix, presigned_post and skill_s3_url call shapes are gone, with
the agent-echoed prefix they relied on, and a trajectory answer that only names a
trajectory_s3_prefix is refused.

A call whose agent does not advertise the object form, or whose object store
issues no grants, fails before anything is sent. A snapshot captured in the
runtime's own layout cannot be restored. Inline skills and inline trajectories
are unchanged.

DeployAgentTaskStep drops agent_changelog_s3_prefix, changelog_apply_call drops
portable and skill_add_call drops forms; the validator stops probing the S3
forms.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 packages/agentenv-protocol/README.md          |  25 +--
 src/agent_env/a2a_agent/a2a_agent.py          |   4 +-
 src/agent_env/a2a_agent/object_transfer.py    | 200 +++++------------
 src/agent_env/a2a_agent/validator.py          |  12 +-
 src/agent_env/artifact/artifacts/skill.py     |   4 +-
 .../snapshot_utils/agent_state_capture.py     |  40 ++--
 .../verify_a2a_skill_config.py                |  38 +---
 .../verify_a2a_trajectory.py                  |  43 +---
 .../task_step/task_steps/add_skills.py        |   5 +-
 .../task_step/task_steps/deploy_agent.py      |  44 +---
 tst/unit/a2a_agent/add_skill_test.py          |  55 ++---
 tst/unit/a2a_agent/object_transfer_test.py    |  22 +-
 .../test_validator_changelog_objects.py       |  32 +--
 .../test_validator_skill_variants.py          |  21 +-
 .../validator_trajectory_dag_test.py          |  38 ++--
 tst/unit/task_step/capture_stubs.py           |  19 +-
 .../task_step/test_agent_state_capture.py     | 207 +++++-------------
 .../test_deploy_agent_changelog_objects.py    | 134 +++---------
 .../test_deploy_agent_snapshot_objects.py     |  82 ++-----
 tst/unit/task_step/test_prompt_agent.py       |  16 +-
 .../test_rubrics_verifier_judge_trajectory.py |   7 +-
 21 files changed, 296 insertions(+), 752 deletions(-)

diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md
index d5529b04..354ab6f7 100644
--- a/packages/agentenv-protocol/README.md
+++ b/packages/agentenv-protocol/README.md
@@ -568,17 +568,14 @@ that is by the uploading agent. AgentEnv accepts a trajectory upload response
 without reading the object back, and registers a snapshot only once both its
 objects are in the store.
 
-SDK agents advertise only these shapes. AgentEnv reads each Agent Card, sends
-the object variants when the agent advertises them and its object store issues
-grants (the S3 store does, and namespace grants for changelog capture only when
-it signs with long-term credentials), and keeps the older `s3_prefix`, `skill_s3_url` and
-`trajectory_s3_prefix` shapes for agents that advertise those instead. An SDK
-agent built on this protocol therefore needs an agent-env release that includes
-it: an older release sends the older shapes, which such an agent refuses apart
-from inline skills and trajectories. Roll out in this
-order: release agent-env and `agentenv-framework-protocol` together, move every service
-that embeds agent-env to that release, and only then build agents on the new
-SDK. A snapshot or changelog is restored in the form it was captured in: one
-captured as objects only through the object variants, an older one only
-through `s3_prefix`. Do not roll agent-env back once portable snapshots or
-changelogs exist, because older releases cannot load them.
+SDK agents advertise only these shapes, and AgentEnv sends no others. It moves
+a skill bundle, snapshot or changelog only through grants, so the call needs an
+agent that advertises the object variant and an object store that issues grants
+(the S3 store does, and namespace grants for changelog capture only when it
+signs with long-term credentials); otherwise it fails before anything is sent.
+Skills given as SKILL.md text and trajectories returned inline need no grants.
+A snapshot is restored only from the snapshot objects in the table above; one
+that holds the runtime's own files instead cannot be. An
+agent-env release without the object variants sends other shapes, which SDK
+agents refuse apart from inline skills and trajectories, and cannot load
+portable snapshots or changelogs, so do not run one alongside these agents.
diff --git a/src/agent_env/a2a_agent/a2a_agent.py b/src/agent_env/a2a_agent/a2a_agent.py
index a84ff315..156b763f 100644
--- a/src/agent_env/a2a_agent/a2a_agent.py
+++ b/src/agent_env/a2a_agent/a2a_agent.py
@@ -187,8 +187,8 @@ async def register_skill(
         skill_md: str | None = None,
         object_url: str | None = None,
     ) -> dict:
-        """Send one skill, as SKILL.md text or the objects under ``object_url``, in a form the
-        agent's card and the object store allow; returns the agent's answer."""
+        """Send one skill, as SKILL.md text inline or the objects under ``object_url`` as a
+        bundle of read grants; returns the agent's answer."""
         add_method, add_path = A2AAgent.operation(A2AAgent._skill_extension(deployed), "add")
         call = await asyncio.to_thread(
             skill_add_call,
diff --git a/src/agent_env/a2a_agent/object_transfer.py b/src/agent_env/a2a_agent/object_transfer.py
index bd158487..dd41dbf3 100644
--- a/src/agent_env/a2a_agent/object_transfer.py
+++ b/src/agent_env/a2a_agent/object_transfer.py
@@ -41,8 +41,8 @@
 
 logger = logging.getLogger(__name__)
 
-# "objects": the call carries grants and the agent moves the bytes. "legacy": every form that
-# carries none, the inline ones and the older S3 ones.
+# "objects": the call carries grants and the agent moves the bytes. "legacy": the inline forms,
+# which carry none.
 TransferMode = Literal["objects", "legacy"]
 _Response = TypeVar("_Response", bound=BaseModel)
 _MEDIA_TYPE = re.compile(r"[!#$&^_.+\-|~0-9a-z]+/[!#$&^_.+\-|~0-9a-z]+")
@@ -81,12 +81,6 @@ class ObjectLimits:
 }
 _OPAQUE = "application/octet-stream"
 
-# The S3 forms the SDK retired; older and hand-written Agent Cards still take them.
-_S3_PREFIX = ("s3_prefix",)
-_SKILL_S3_URL = ("name", "description", "skill_s3_url")
-_SNAPSHOT_SAVE_S3_PREFIX = ("context_id", "s3_prefix")
-
-
 def _fields(model: type[BaseModel]) -> tuple[str, ...]:
     return request_fields(model).required
 
@@ -123,6 +117,21 @@ def _accepts(method: Mapping[str, Any] | None, fields: Collection[str]) -> bool:
     return isinstance(request, Mapping) and card_request_accepts(request, fields)
 
 
+def _require_object_form(
+    method: Mapping[str, Any] | None,
+    model: type[BaseModel],
+    store: ObjectStore,
+    *,
+    operation: str,
+) -> None:
+    """Refuse ``operation`` unless the agent takes its object form and the store issues the
+    grants that form carries: agent-env moves objects no other way."""
+    if not _accepts(method, _fields(model)):
+        raise RuntimeError(f"{operation}: the agent does not advertise the object form")
+    if not store.supports_transfer_grants:
+        raise RuntimeError(f"{operation}: the object store does not issue transfer grants")
+
+
 @dataclass(frozen=True)
 class TransferCall:
     """One extension call ready to send: the form it takes and its body."""
@@ -249,29 +258,6 @@ def namespace_grant(
     )
 
 
-def bounded_echo(issued: str, echoed: str) -> str:
-    """Where a legacy agent says it wrote, clamped to the prefix agent-env issued.
-
-    Sidecars uploading with their own client nest under it and echo that; registering the
-    issued prefix instead points the artifact a level too high, which `put_existing`'s
-    recursive list hides until a restore cannot find its files.
-
-    Escaping is still refused — `put_existing` registers everything under what it is handed,
-    with the worker's credentials. Nesting is free to allow: the presigned POST's
-    `starts-with $key` already admits any depth under the issued prefix.
-    """
-    issued_norm, echoed_norm = issued.rstrip("/"), echoed.rstrip("/")
-    if echoed_norm == issued_norm:
-        return issued
-    if echoed_norm.startswith(issued_norm + "/"):
-        return echoed
-    logger.warning(
-        "the agent echoed a prefix outside the one issued; registering the "
-        "issued one (issued=%s returned=%s)", issued, echoed,
-    )
-    return issued
-
-
 def skill_bundle_request(
     store: ObjectStore, *, name: str, description: str, object_url: str
 ) -> BundleSkillRequest:
@@ -297,24 +283,12 @@ def skill_add_call(
     description: str,
     skill_md: str | None = None,
     object_url: str | None = None,
-    forms: Collection[TransferMode] = ("objects", "legacy"),
 ) -> TransferCall:
-    """The skill ``add`` call for a skill given as SKILL.md text or as the objects under
-    ``object_url``, in a form the agent and the store allow; ``forms`` narrows the choice."""
+    """The skill ``add`` call for a skill given as SKILL.md text, sent inline, or as the objects
+    under ``object_url``, sent as a bundle of read grants."""
     base = {"name": name, "description": description}
     if object_url is not None:
-        mode = choose_transfer(
-            method,
-            objects=_fields(BundleSkillRequest) if "objects" in forms else None,
-            legacy=_SKILL_S3_URL if "legacy" in forms else None,
-            store=store,
-        )
-        if mode is None:
-            raise RuntimeError(
-                "Agent advertises no object-backed skill form this object store can serve"
-            )
-        if mode == "legacy":
-            return TransferCall("legacy", {**base, "skill_s3_url": object_url})
+        _require_object_form(method, BundleSkillRequest, store, operation="skill add")
         request = skill_bundle_request(
             store, name=name, description=description, object_url=object_url
         )
@@ -335,30 +309,12 @@ def snapshot_save_call(
     capture_prefix: str,
 ) -> TransferCall:
     """The snapshot ``save`` call that writes one capture below ``capture_prefix``."""
-    mode = choose_transfer(
+    _require_object_form(
         method,
-        objects=_fields(ObjectSnapshotSaveRequest),
-        legacy=_SNAPSHOT_SAVE_S3_PREFIX,
-        store=store,
+        ObjectSnapshotSaveRequest,
+        store,
+        operation=f"snapshot save on agent '{agent_name}'",
     )
-    if mode is None:
-        raise RuntimeError(
-            f"Agent '{agent_name}' advertises no snapshot save form this object store can serve"
-        )
-    if mode == "legacy":
-        # Signed with the step's fresh credentials: the sidecar's are the deployer's STS
-        # session, frozen at deploy and expired on a long run.
-        presigned_post = store.signed_post(capture_prefix)
-        return TransferCall(
-            "legacy",
-            {
-                "context_id": context_id,
-                "s3_prefix": capture_prefix,
-                # Absent, not null, when the backend cannot sign: the sidecar then uploads
-                # with its own client.
-                **({"presigned_post": presigned_post} if presigned_post else {}),
-            },
-        )
     objects = {
         name: write_object(store, url, media_type=_OPAQUE, max_bytes=max_bytes)
         for name, url, max_bytes in _snapshot_objects(store, capture_prefix)
@@ -378,24 +334,14 @@ def snapshot_load_call(
     file_names: Collection[str],
     target_context_id: str | None,
 ) -> TransferCall:
-    """The snapshot ``load`` call that restores the capture at ``bundle_url`` in the form it was
-    captured in. An older capture is sent as ``bundle_url`` itself, whatever form it is stored in."""
-    portable = is_portable_snapshot(file_names)
-    mode = (
-        choose_transfer(method, objects=_fields(ObjectSnapshotLoadRequest), store=store)
-        if portable
-        else choose_transfer(method, legacy=_S3_PREFIX, store=store)
-    )
-    if mode is None:
+    """The snapshot ``load`` call that restores the capture at ``bundle_url``. Only a portable
+    capture can be restored; an older one holds the runtime's own files."""
+    operation = f"snapshot load on agent '{agent_name}'"
+    if not is_portable_snapshot(file_names):
         raise RuntimeError(
-            f"Agent '{agent_name}' cannot load the configured "
-            f"{'portable' if portable else 'legacy'} snapshot"
+            f"{operation}: {bundle_url} is not a portable snapshot, so it cannot be restored"
         )
-    if mode == "legacy":
-        payload: dict[str, Any] = {"s3_prefix": bundle_url}
-        if target_context_id:
-            payload["target_context_id"] = target_context_id
-        return TransferCall("legacy", payload)
+    _require_object_form(method, ObjectSnapshotLoadRequest, store, operation=operation)
     objects = {
         name: read_object(store, url, media_type=_OPAQUE, max_bytes=max_bytes)
         for name, url, max_bytes in _snapshot_objects(store, bundle_url)
@@ -423,37 +369,20 @@ def changelog_enable_call(
     namespace_url: str,
     expires_in: int,
 ) -> TransferCall:
-    """The ``enable-changelog`` call that captures below ``namespace_url``. When the store cannot
-    sign a namespace grant that lasts ``expires_in``, an agent that also takes ``s3_prefix``
-    gets that form; one that does not fails with GrantUnavailableError."""
-    mode = choose_transfer(
-        method, objects=_fields(NamespaceChangelogEnableRequest), legacy=_S3_PREFIX, store=store
-    )
-    if mode is None:
-        raise RuntimeError(
-            f"Agent '{agent_name}' advertises no changelog enable form this object store can serve"
+    """The ``enable-changelog`` call that captures below ``namespace_url``. Raises
+    GrantUnavailableError when the store cannot sign a namespace grant that lasts ``expires_in``."""
+    operation = f"changelog enable on agent '{agent_name}'"
+    _require_object_form(method, NamespaceChangelogEnableRequest, store, operation=operation)
+    try:
+        grant = namespace_grant(
+            store, namespace_url, limits=CHANGELOG_LIMITS, expires_in=expires_in
         )
-    if mode == "objects":
-        try:
-            grant = namespace_grant(
-                store, namespace_url, limits=CHANGELOG_LIMITS, expires_in=expires_in
-            )
-        except GrantUnavailableError as exc:
-            if choose_transfer(method, legacy=_S3_PREFIX, store=store) is None:
-                raise GrantUnavailableError(
-                    f"Agent '{agent_name}' takes only the object changelog enable form, and "
-                    f"the object store cannot issue its namespace grant: {exc}"
-                ) from exc
-            logger.warning(
-                "Portable changelog grant is unavailable for agent '%s' (%s); "
-                "using its legacy s3_prefix variant",
-                agent_name,
-                exc,
-            )
-        else:
-            request = NamespaceChangelogEnableRequest(write_namespace=grant)
-            return TransferCall("objects", request.model_dump(mode="json", exclude_none=True))
-    return TransferCall("legacy", {"s3_prefix": namespace_url})
+    except GrantUnavailableError as exc:
+        raise GrantUnavailableError(
+            f"{operation}: the object store cannot issue its namespace grant: {exc}"
+        ) from exc
+    request = NamespaceChangelogEnableRequest(write_namespace=grant)
+    return TransferCall("objects", request.model_dump(mode="json", exclude_none=True))
 
 
 def changelog_apply_call(
@@ -462,33 +391,19 @@ def changelog_apply_call(
     *,
     agent_name: str,
     source_url: str,
-    portable: bool,
     up_to_tool_call_exclusive: int | None = None,
     resume_conversation: bool = False,
     target_context_id: str | None = None,
 ) -> TransferCall:
-    """The ``apply-changelog`` call that replays the capture at ``source_url`` in the form it was
-    captured in: a portable one sends read grants for its increments before the cutoff, in
-    sequence order, and none when the cutoff precedes the first tool call."""
-    mode = (
-        choose_transfer(method, objects=_fields(ObjectChangelogApplyRequest), store=store)
-        if portable
-        else choose_transfer(method, legacy=_S3_PREFIX, store=store)
+    """The ``apply-changelog`` call that replays the capture at ``source_url``: read grants for
+    its increments before the cutoff, in sequence order, and none when the cutoff precedes the
+    first tool call."""
+    _require_object_form(
+        method,
+        ObjectChangelogApplyRequest,
+        store,
+        operation=f"changelog apply on agent '{agent_name}'",
     )
-    if mode is None:
-        raise RuntimeError(
-            f"Agent '{agent_name}' cannot apply the configured "
-            f"{'portable' if portable else 'legacy'} changelog"
-        )
-    if mode == "legacy":
-        payload: dict[str, Any] = {"s3_prefix": source_url}
-        if resume_conversation:
-            payload["resume_conversation"] = True
-        if up_to_tool_call_exclusive is not None:
-            payload["up_to_tool_call_exclusive"] = up_to_tool_call_exclusive
-        if target_context_id:
-            payload["target_context_id"] = target_context_id
-        return TransferCall("legacy", payload)
     cutoff = math.inf if up_to_tool_call_exclusive is None else up_to_tool_call_exclusive
     described = read_objects_under(
         store,
@@ -568,11 +483,9 @@ def to(cls, store: ObjectStore, object_url: str) -> TrajectoryUpload:
 @dataclass(frozen=True)
 class FetchedTrajectory:
     """Where a trajectory ``get`` left the trajectory: uploaded by the agent through the grant
-    (``object_url``), below a prefix a legacy agent uploaded to itself (``legacy_prefix``), or in
-    the answer (``inline``). All None: the agent returned none."""
+    (``object_url``) or in the answer (``inline``). Both None: the agent returned none."""
 
     object_url: str | None = None
-    legacy_prefix: str | None = None
     inline: Any = None
 
 
@@ -604,6 +517,9 @@ async def fetch_trajectory(
         return FetchedTrajectory(object_url=upload.object_url)
     if not isinstance(body, Mapping):
         return FetchedTrajectory()
-    return FetchedTrajectory(
-        legacy_prefix=body.get("trajectory_s3_prefix") or None, inline=body.get("trajectory")
-    )
+    if body.get("trajectory") is None and body.get("trajectory_s3_prefix"):
+        raise RuntimeError(
+            "trajectory get answered with trajectory_s3_prefix, which agent-env does not read; "
+            "the agent must return the trajectory inline or take the object form"
+        )
+    return FetchedTrajectory(inline=body.get("trajectory"))
diff --git a/src/agent_env/a2a_agent/validator.py b/src/agent_env/a2a_agent/validator.py
index 985800ba..0a642d95 100644
--- a/src/agent_env/a2a_agent/validator.py
+++ b/src/agent_env/a2a_agent/validator.py
@@ -196,13 +196,6 @@ async def validate(
             description="Validator probe for the portable skill bundle variant.",
             body="Portable skill bundle validation fixture.",
         )
-        skill_s3_probe_url = await asyncio.to_thread(
-            A2AAgentValidator._upload_skill_fixture,
-            agent,
-            name="validator-probe-s3",
-            description="Validator probe for the legacy S3 skill variant.",
-            body="Legacy S3 skill validation fixture.",
-        )
         fixtures = await asyncio.to_thread(A2AAgentValidator._upload_probe_fixtures, agent, skill_object_url)
 
         # ── Build modality probe steps + matching grading list ───────────────
@@ -286,7 +279,6 @@ async def validate(
                 a2a_agent_id=agent.id,
                 rubric_verifier_id=skill_verifier_id,
                 skill_bundle_object_url=skill_bundle_probe_url,
-                skill_s3_url=skill_s3_probe_url,
                 depends_on=[TaskStepDependency(task_step_id=skill_rubric_id)],
                 fail_task_on_error=False,
             ),
@@ -665,7 +657,6 @@ def record(*, supported, advertised, save_ok, apply_ok, roundtrip_ok, note=""):
                 get_config().get_object_store(),
                 agent_name=apply_agent_name,
                 source_url=capture_source,
-                portable=capture.get("transfer_mode") == "objects",
             )
         except RuntimeError as e:
             record(supported=False, advertised=advertised, save_ok=save_ok,
@@ -685,8 +676,7 @@ def record(*, supported, advertised, save_ok, apply_ok, roundtrip_ok, note=""):
                 timeout=TRANSFER_TIMEOUT_SECONDS,
                 response_model=ObjectChangelogApplyResponse,
             )
-            if call.mode == "objects":
-                check_changelog_applied(answer, call, agent_name=apply_agent_name)
+            check_changelog_applied(answer, call, agent_name=apply_agent_name)
         except Exception as e:  # noqa: BLE001
             record(supported=False, advertised=advertised, save_ok=save_ok,
                    apply_ok=False, roundtrip_ok=False, note=f"apply request failed: {e}")
diff --git a/src/agent_env/artifact/artifacts/skill.py b/src/agent_env/artifact/artifacts/skill.py
index 1de81504..b9275794 100644
--- a/src/agent_env/artifact/artifacts/skill.py
+++ b/src/agent_env/artifact/artifacts/skill.py
@@ -3,7 +3,7 @@
 A SkillArtifact references one FileArtifactUniverse bundling every file in the skill
 (SKILL.md + scripts/, references/, assets/, …) colocated under a single object-store prefix
 (`artifacts/skill/<key_segment(id)>/<version>/`). That prefix IS the bundle — no duplication, and
-the prefix is handed to deployed A2A agents as `skill_s3_url`.
+deployed A2A agents get its files through read grants.
 """
 
 from __future__ import annotations
@@ -38,7 +38,7 @@ class SkillArtifact(Artifact):
     type: Literal["skill"] = "skill"
 
     skill_files_id: str = Field(description="ID of the FileArtifactUniverse bundling every file in the skill")
-    skill_object_url: str = Field(alias="skill_s3_url", description="Object-store prefix where the skill bundle lives (handed to agents)")
+    skill_object_url: str = Field(alias="skill_s3_url", description="Object-store prefix where the skill bundle lives")
     agent_skills_spec_version: str = Field(description="agentskills.io spec revision this skill was validated under")
 
     skill_name: str = Field(description="Skill name from SKILL.md frontmatter; equals the artifact id")
diff --git a/src/agent_env/task_step/snapshot_utils/agent_state_capture.py b/src/agent_env/task_step/snapshot_utils/agent_state_capture.py
index 3e969d42..d284aba1 100644
--- a/src/agent_env/task_step/snapshot_utils/agent_state_capture.py
+++ b/src/agent_env/task_step/snapshot_utils/agent_state_capture.py
@@ -22,7 +22,6 @@
     SNAPSHOT_WORKSPACE_OBJECT_NAME,
     FetchedTrajectory,
     TrajectoryUpload,
-    bounded_echo,
     fetch_trajectory,
     invoke_transfer,
     snapshot_save_call,
@@ -39,10 +38,9 @@ class WorkspaceCapture:
 
     universe_id: str
     universe_version: Optional[int]
-    # Where the files landed; what restores.
+    # What restores.
     bundle_object_url: str
-    # What was presigned — not always where the agent wrote. Per-service state,
-    # captured alongside, still goes here.
+    # Where the agent wrote it. Per-service state, captured alongside, goes here too.
     capture_prefix: str
 
 
@@ -64,7 +62,7 @@ async def capture_workspace(
     artifact_id: str,
     timeout_seconds: float,
 ) -> WorkspaceCapture:
-    """Tar the agent's workspace to a fresh S3 prefix and wrap it as a universe.
+    """Capture the agent's workspace below a fresh prefix and wrap it as a universe.
 
     Raises on any failure — without the tar there is nothing to grade."""
     from agent_env.artifact.store import get_artifact_store
@@ -93,7 +91,7 @@ async def capture_workspace(
         context_id=a2a_context_id,
         capture_prefix=capture_prefix,
     )
-    save_body = await invoke_transfer(
+    await invoke_transfer(
         a2a_url + save_path,
         call,
         verb="POST",
@@ -102,35 +100,28 @@ async def capture_workspace(
         response_model=ObjectSnapshotSaveResponse,
     )
 
-    if call.mode == "objects":
-        # The agent's answer is only a claim: a snapshot missing either object cannot be restored.
-        listed = await asyncio.to_thread(store.list_at, capture_prefix)
-        stored = {url.rsplit("/", 1)[-1] for url in listed}
-        missing = {SNAPSHOT_TRAJECTORY_OBJECT_NAME, SNAPSHOT_WORKSPACE_OBJECT_NAME} - stored
-        if missing:
-            raise RuntimeError(
-                f"snapshot save from agent '{agent_name}' left {sorted(missing)} out of the object store"
-            )
-        registered_prefix = capture_prefix
-    else:
-        echoed = save_body.get("s3_prefix")
-        if not echoed:
-            raise RuntimeError(f"snapshot save response missing 's3_prefix': {save_body}")
-        registered_prefix = bounded_echo(capture_prefix, echoed)
+    # The agent's answer is only a claim: a snapshot missing either object cannot be restored.
+    listed = await asyncio.to_thread(store.list_at, capture_prefix)
+    stored = {url.rsplit("/", 1)[-1] for url in listed}
+    missing = {SNAPSHOT_TRAJECTORY_OBJECT_NAME, SNAPSHOT_WORKSPACE_OBJECT_NAME} - stored
+    if missing:
+        raise RuntimeError(
+            f"snapshot save from agent '{agent_name}' left {sorted(missing)} out of the object store"
+        )
 
     from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse
 
     # Unserialized — a concurrent version race on one artifact id is
     # `ArtifactStore.put_document`'s retry to absorb.
     universe = await asyncio.to_thread(
-        FileArtifactUniverse.put_existing, id=artifact_id, s3_url=registered_prefix
+        FileArtifactUniverse.put_existing, id=artifact_id, s3_url=capture_prefix
     )
     # `put_existing` always sets it, but the field is Optional on the artifact, and a
     # row carrying None here would read as an ungradable capture rather than an error.
     if not universe.bundle_object_url:
         raise RuntimeError(
             f"FileArtifactUniverse {universe.id} v{universe.version} registered no "
-            f"bundle url for {registered_prefix}"
+            f"bundle url for {capture_prefix}"
         )
     logger.info(
         "Snapshot captured: agent=%s context_id=%s -> FileArtifactUniverse %s v%s at %s",
@@ -242,9 +233,6 @@ def store_trajectory(
     """The URL of a fetched trajectory, storing it first when the agent returned it inline."""
     if fetched.object_url is not None:
         return fetched.object_url
-    if fetched.legacy_prefix:
-        urls = get_config().get_object_store().list_at(fetched.legacy_prefix)
-        return urls[0] if urls else None
     if fetched.inline is not None:
         return upload_trajectory(fetched.inline, trajectory_output_prefix, name=name)
     return None
diff --git a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py
index eca46ad5..78283265 100644
--- a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py
+++ b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py
@@ -12,7 +12,6 @@
 from agent_env.task_step.context import TaskStepContext
 from agent_env.a2a_agent.object_transfer import (
     TRANSFER_TIMEOUT_SECONDS,
-    TransferMode,
     invoke_transfer,
     parse_response,
     skill_add_call,
@@ -37,20 +36,17 @@ def __init__(
         depends_on: Optional[list[TaskStepDependency]] = None,
         fail_task_on_error: bool = True,
         skill_bundle_object_url: Optional[str] = None,
-        skill_s3_url: Optional[str] = None,
     ):
         super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
         self.a2a_agent_id = a2a_agent_id
         self.rubric_verifier_id = rubric_verifier_id
         self.skill_bundle_object_url = skill_bundle_object_url
-        self.skill_s3_url = skill_s3_url
 
     def to_dict(self) -> dict:
         base = super().to_dict()
         base["a2a_agent_id"] = self.a2a_agent_id
         base["rubric_verifier_id"] = self.rubric_verifier_id
         base["skill_bundle_object_url"] = self.skill_bundle_object_url
-        base["skill_s3_url"] = self.skill_s3_url
         return base
 
     @classmethod
@@ -60,7 +56,6 @@ def from_dict(cls, data: dict) -> VerifyA2ASkillConfigStep:
             a2a_agent_id=data["a2a_agent_id"],
             rubric_verifier_id=data["rubric_verifier_id"],
             skill_bundle_object_url=data.get("skill_bundle_object_url"),
-            skill_s3_url=data.get("skill_s3_url"),
         )
 
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
@@ -90,8 +85,8 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         store = get_config().get_object_store()
         a2a_url = deployed_agent.a2a_url or deployed_agent.api_url
 
-        async def probe_add(form: TransferMode, name: str, description: str, object_url: str | None) -> bool:
-            """Whether the agent registers an object-backed skill sent in exactly ``form``."""
+        async def probe_bundle(name: str, description: str, object_url: str | None) -> bool:
+            """Whether the agent registers an object-backed skill sent as a bundle."""
             if object_url is None:
                 return False
             try:
@@ -102,49 +97,40 @@ async def probe_add(form: TransferMode, name: str, description: str, object_url:
                     name=name,
                     description=description,
                     object_url=object_url,
-                    forms=(form,),
                 )
             except RuntimeError as exc:
-                logger.info("Skill %s request: not sent (%s)", form, exc)
+                logger.info("Skill bundle request: not sent (%s)", exc)
                 return False
             except Exception as exc:
-                logger.warning("Skill %s request: failed (%s)", form, exc)
+                logger.warning("Skill bundle request: failed (%s)", exc)
                 return False
             try:
                 answer = await invoke_transfer(
                     a2a_url + add_path,
                     call,
                     verb="POST",
-                    operation=f"skill add ({form})",
+                    operation="skill add (bundle)",
                     timeout=TRANSFER_TIMEOUT_SECONDS,
                 )
-                result = parse_response(SkillAddResponse, answer, operation=f"skill add ({form})")
+                result = parse_response(SkillAddResponse, answer, operation="skill add (bundle)")
                 if result.name != name:
                     raise ValueError("response name does not match the requested skill")
-                logger.info("Skill %s request: OK", form)
+                logger.info("Skill bundle request: OK")
                 return True
             except Exception as exc:
-                logger.warning("Skill %s request: failed (%s)", form, exc)
+                logger.warning("Skill bundle request: failed (%s)", exc)
                 return False
 
-        bundle_passed = await probe_add(
-            "objects",
+        bundle_passed = await probe_bundle(
             "validator-probe-bundle",
             "Validator probe for the portable skill bundle variant.",
             self.skill_bundle_object_url,
         )
-        s3_passed = await probe_add(
-            "legacy",
-            "validator-probe-s3",
-            "Validator probe for the legacy S3 skill variant.",
-            self.skill_s3_url,
-        )
         logger.info(
-            "Skill validation '%s': inline=%s bundle=%s s3=%s",
+            "Skill validation '%s': inline=%s bundle=%s",
             self.rubric_verifier_id,
             inline_passed,
             bundle_passed,
-            s3_passed,
         )
 
         # List skills via GET endpoint
@@ -161,7 +147,7 @@ async def probe_add(form: TransferMode, name: str, description: str, object_url:
             logger.warning(f"Skill list failed: {e}")
 
         # Build validated entry from observed behavior for each request variant.
-        add_supported = inline_passed or bundle_passed or s3_passed
+        add_supported = inline_passed or bundle_passed
         skill_entry = {
             "supported": add_supported or list_ok,
             "methods": {
@@ -173,7 +159,6 @@ async def probe_add(form: TransferMode, name: str, description: str, object_url:
                         "skill": {"supported": inline_passed},
                         "skill_md": {"supported": inline_passed},
                         "skill_bundle": {"supported": bundle_passed},
-                        "skill_s3_url": {"supported": s3_passed},
                     },
                 },
                 "list": {"supported": list_ok},
@@ -188,7 +173,6 @@ async def probe_add(form: TransferMode, name: str, description: str, object_url:
         context.metadata.setdefault("verifications", {})["a2a_skill_config"] = {
             "inline": inline_passed,
             "bundle": bundle_passed,
-            "s3": s3_passed,
             "list": list_ok,
         }
         return context
diff --git a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py
index 7e6f03a3..cb9efe1e 100644
--- a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py
+++ b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py
@@ -8,12 +8,8 @@
 
 from agent_env.task_step.context import TaskStepContext
 from agent_env.a2a_agent.object_transfer import (
-    REPLY_TIMEOUT_SECONDS,
-    TransferCall,
     TrajectoryUpload,
-    choose_transfer,
     fetch_trajectory,
-    invoke_transfer,
     trajectory_mode,
 )
 from agent_env.task_step.snapshot_utils.agent_state_capture import trajectory_object_url
@@ -105,47 +101,17 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             except Exception as e:
                 logger.warning(f"Object trajectory: failed ({e})")
 
-        s3_ok = False
-        if choose_transfer(
-            get_method, legacy=("task_id", "trajectory_s3_prefix"), store=store
-        ) == "legacy":
-            s3_prefix = store.object_url(probe_prefix)
-            try:
-                data = await invoke_transfer(
-                    endpoint,
-                    TransferCall(
-                        "legacy", {"task_id": task_id, "trajectory_s3_prefix": s3_prefix}
-                    ),
-                    verb="POST",
-                    operation="legacy trajectory get",
-                    timeout=REPLY_TIMEOUT_SECONDS,
-                )
-                if data.get("trajectory_s3_prefix"):
-                    s3_ok = True
-                    logger.info(f"S3 trajectory: OK ({data['trajectory_s3_prefix']})")
-                else:
-                    logger.warning(
-                        "S3 trajectory: response missing 'trajectory_s3_prefix' key or empty"
-                    )
-            except Exception as e:
-                logger.warning(f"S3 trajectory: failed ({e})")
-
-        logger.info(
-            f"Trajectory validation: inline={inline_ok} objects={objects_ok} s3={s3_ok}"
-        )
+        logger.info(f"Trajectory validation: inline={inline_ok} objects={objects_ok}")
 
         # Build validated_a2a_extensions entry
         traj_entry = {
-            "supported": inline_ok or objects_ok or s3_ok,
+            "supported": inline_ok or objects_ok,
             "methods": {
                 "get": {
-                    "supported": inline_ok or objects_ok or s3_ok,
+                    "supported": inline_ok or objects_ok,
                     "options": {
-                        "task_id": {
-                            "supported": inline_ok or objects_ok or s3_ok
-                        },
+                        "task_id": {"supported": inline_ok or objects_ok},
                         "objects": {"supported": objects_ok},
-                        "trajectory_s3_prefix": {"supported": s3_ok},
                     },
                 },
             },
@@ -159,6 +125,5 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         context.metadata.setdefault("verifications", {})["a2a_trajectory"] = {
             "inline": inline_ok,
             "objects": objects_ok,
-            "s3": s3_ok,
         }
         return context
diff --git a/src/agent_env/task_step/task_steps/add_skills.py b/src/agent_env/task_step/task_steps/add_skills.py
index 3b5320eb..aa3c3ac6 100644
--- a/src/agent_env/task_step/task_steps/add_skills.py
+++ b/src/agent_env/task_step/task_steps/add_skills.py
@@ -22,8 +22,9 @@ class Skill:
 
     A skill can come from three mutually-exclusive sources:
     - **inline**: body + frontmatter fields, rendered into SKILL.md at send time
-    - **s3_url**: S3 prefix containing a skill directory with its own SKILL.md
-    - **skill_artifact_id**: a SkillArtifact in the store; name/description/skill_s3_url
+    - **s3_url**: object-store prefix containing a skill directory with its own SKILL.md,
+      sent as a bundle of read grants
+    - **skill_artifact_id**: a SkillArtifact in the store; name/description/skill_object_url
       are resolved from it at send time
     """
     name: Optional[str] = None
diff --git a/src/agent_env/task_step/task_steps/deploy_agent.py b/src/agent_env/task_step/task_steps/deploy_agent.py
index 5b4eb49d..6dc17122 100644
--- a/src/agent_env/task_step/task_steps/deploy_agent.py
+++ b/src/agent_env/task_step/task_steps/deploy_agent.py
@@ -16,7 +16,6 @@
 from agent_env.a2a_agent.object_transfer import (
     REPLY_TIMEOUT_SECONDS,
     TRANSFER_TIMEOUT_SECONDS,
-    bounded_echo,
     changelog_apply_call,
     changelog_enable_call,
     check_changelog_applied,
@@ -119,7 +118,6 @@ def __init__(
         network_policy: Optional[dict] = None,
         role: Optional[str] = None,
         enable_agent_changelog: bool = False,
-        agent_changelog_s3_prefix: Optional[str] = None,
         agent_changelog_toolcall_position_exclusive: Optional[int] = None,
         sandbox_name: Optional[str] = None,
         enable_docker: bool = False,
@@ -152,18 +150,13 @@ def __init__(
         self.role = role
         self.enable_agent_changelog = enable_agent_changelog
         self.agent_changelog_object_url = agent_changelog_object_url
-        self.agent_changelog_s3_prefix = agent_changelog_s3_prefix
         self.agent_changelog_toolcall_position_exclusive = agent_changelog_toolcall_position_exclusive
-        if self.agent_changelog_object_url and self.agent_changelog_s3_prefix:
-            raise ValueError(
-                "set only one of agent_changelog_object_url and "
-                "agent_changelog_s3_prefix"
-            )
-        if self.agent_changelog_toolcall_position_exclusive is not None and not (
-            self.agent_changelog_object_url or self.agent_changelog_s3_prefix
+        if (
+            self.agent_changelog_toolcall_position_exclusive is not None
+            and not self.agent_changelog_object_url
         ):
             raise ValueError(
-                "agent_changelog_toolcall_position_exclusive requires a changelog source"
+                "agent_changelog_toolcall_position_exclusive requires agent_changelog_object_url"
             )
         self.sandbox_name = sandbox_name
         # Rootless Docker-in-Docker for the agent (not the VM socket); VM mode only.
@@ -194,7 +187,6 @@ def to_dict(self) -> dict:
         base["role"] = self.role
         base["enable_agent_changelog"] = self.enable_agent_changelog
         base["agent_changelog_object_url"] = self.agent_changelog_object_url
-        base["agent_changelog_s3_prefix"] = self.agent_changelog_s3_prefix
         base["agent_changelog_toolcall_position_exclusive"] = self.agent_changelog_toolcall_position_exclusive
         base["sandbox_name"] = self.sandbox_name
         base["enable_docker"] = self.enable_docker
@@ -227,7 +219,6 @@ def from_dict(cls, data: dict) -> DeployAgentTaskStep:
             role=data.get("role"),
             enable_agent_changelog=data.get("enable_agent_changelog", False),
             agent_changelog_object_url=data.get("agent_changelog_object_url"),
-            agent_changelog_s3_prefix=data.get("agent_changelog_s3_prefix"),
             agent_changelog_toolcall_position_exclusive=data.get("agent_changelog_toolcall_position_exclusive"),
             sandbox_name=data.get("sandbox_name"),
             enable_docker=data.get("enable_docker", False),
@@ -395,7 +386,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         if self.agent_snapshot_files_artifact_id:
             await self._load_snapshot(a2a_url, card, a2a_agent_id, context)
 
-        if self.agent_changelog_object_url or self.agent_changelog_s3_prefix:
+        if self.agent_changelog_object_url:
             await self._apply_agent_changelog(a2a_url, card, context)
 
         if self.enable_agent_changelog:
@@ -488,18 +479,13 @@ async def _configure_agent_changelog(
             timeout=REPLY_TIMEOUT_SECONDS,
             response_model=NamespaceChangelogEnableResponse,
         )
-        if call.mode == "objects":
-            roots, object_url = answer.roots, namespace_url
-        else:
-            roots = answer.get("roots")
-            object_url = bounded_echo(namespace_url, answer.get("s3_prefix") or namespace_url)
         context.metadata.setdefault("agent_changelog", []).append({
             "agent_name": self.agent_name,
-            "roots": roots,
+            "roots": answer.roots,
             "transfer_mode": call.mode,
-            "object_url": object_url,
+            "object_url": namespace_url,
         })
-        logger.info(f"agent-changelog capture enabled on '{self.agent_name}': {object_url}")
+        logger.info(f"agent-changelog capture enabled on '{self.agent_name}': {namespace_url}")
 
     async def _apply_agent_changelog(self, a2a_url: str, card: dict, context: TaskStepContext) -> None:
         """Rewind this fresh agent to a point in a source changelog (fs + conversation) via the snapshot apply-changelog method and resume."""
@@ -513,14 +499,13 @@ async def _apply_agent_changelog(self, a2a_url: str, card: dict, context: TaskSt
                 f"'{A2AAgent.SNAPSHOT_METHOD_APPLY_CHANGELOG}' method; cannot apply agent changelog"
             )
         store = get_config().get_object_store()
-        source_url = self.agent_changelog_object_url or self.agent_changelog_s3_prefix
+        source_url = self.agent_changelog_object_url
         call = await asyncio.to_thread(
             changelog_apply_call,
             method,
             store,
             agent_name=self.agent_name,
             source_url=source_url,
-            portable=bool(self.agent_changelog_object_url),
             up_to_tool_call_exclusive=self.agent_changelog_toolcall_position_exclusive,
             resume_conversation=True,
             target_context_id=self.agent_snapshot_target_context_id,
@@ -533,11 +518,8 @@ async def _apply_agent_changelog(self, a2a_url: str, card: dict, context: TaskSt
             timeout=TRANSFER_TIMEOUT_SECONDS,
             response_model=ObjectChangelogApplyResponse,
         )
-        if call.mode == "objects":
-            check_changelog_applied(answer, call, agent_name=self.agent_name)
-            context_id = answer.context_id
-        else:
-            context_id = answer.get("context_id")
+        check_changelog_applied(answer, call, agent_name=self.agent_name)
+        context_id = answer.context_id
         context.metadata.setdefault("agent_changelog_rewinds", []).append({
             "agent_name": self.agent_name,
             "context_id": context_id,
@@ -594,9 +576,7 @@ async def _load_snapshot(
             timeout=TRANSFER_TIMEOUT_SECONDS,
             response_model=ObjectSnapshotLoadResponse,
         )
-        loaded_context_id = (
-            answer.context_id if call.mode == "objects" else answer.get("context_id")
-        )
+        loaded_context_id = answer.context_id
         logger.info(
             f"Loaded snapshot universe={universe.id} v{universe.version} "
             f"into agent '{self.agent_name}' as context_id={loaded_context_id}"
diff --git a/tst/unit/a2a_agent/add_skill_test.py b/tst/unit/a2a_agent/add_skill_test.py
index 44f15442..dba9977c 100644
--- a/tst/unit/a2a_agent/add_skill_test.py
+++ b/tst/unit/a2a_agent/add_skill_test.py
@@ -98,7 +98,7 @@ async def post(self, url, *, json, timeout):
         body = (
             {"status": "added", "name": json["name"], "source": "s3"}
             if agent.status_code < 400
-            else {"detail": "skill_s3_url must be s3://bucket/key"}
+            else {"detail": "skill_md has no frontmatter"}
         )
         return httpx.Response(agent.status_code, json=body, request=httpx.Request("POST", url))
 
@@ -166,50 +166,29 @@ async def test_artifact_skill_prefers_portable_bundle_when_advertised(
 
 
 @pytest.mark.asyncio
-async def test_s3_skill_keeps_legacy_shape_when_bundle_is_not_advertised(
+async def test_an_object_skill_is_refused_by_an_agent_without_the_bundle_form(
     store: GrantingObjectStore, prefix: str, agent: _Agent
 ) -> None:
-    await A2AAgent.add_skill(
-        _deployed(bundle=False),
-        Skill(name="review", description="Review work", s3_url=prefix),
-    )
-
-    assert agent.requests[0]["json"] == {
-        "name": "review",
-        "description": "Review work",
-        "skill_s3_url": prefix,
-    }
-    assert store.granted == []
-
-
-@pytest.mark.asyncio
-async def test_skill_uses_the_advertised_legacy_shape_on_a_store_without_grants(
-    store: GrantingObjectStore, prefix: str, agent: _Agent
-) -> None:
-    store.supports_transfer_grants = False
-
-    await A2AAgent.add_skill(
-        _deployed(bundle=True, legacy=True),
-        Skill(name="review", description="Review work", s3_url=prefix),
-    )
+    with pytest.raises(RuntimeError, match="skill add: the agent does not advertise the object form"):
+        await A2AAgent.add_skill(
+            _deployed(bundle=False),
+            Skill(name="review", description="Review work", s3_url=prefix),
+        )
 
-    assert agent.requests[0]["json"] == {
-        "name": "review",
-        "description": "Review work",
-        "skill_s3_url": prefix,
-    }
+    assert agent.requests == []
     assert store.granted == []
 
 
 @pytest.mark.asyncio
-async def test_bundle_only_agent_on_a_store_without_grants_is_refused_before_any_request(
-    store: GrantingObjectStore, prefix: str, agent: _Agent
+@pytest.mark.parametrize("legacy", [True, False], ids=["dual", "bundle-only"])
+async def test_an_object_skill_on_a_store_without_grants_is_refused_before_any_request(
+    store: GrantingObjectStore, prefix: str, agent: _Agent, legacy: bool
 ) -> None:
     store.supports_transfer_grants = False
 
-    with pytest.raises(RuntimeError, match="no object-backed skill form"):
+    with pytest.raises(RuntimeError, match="does not issue transfer grants"):
         await A2AAgent.add_skill(
-            _deployed(bundle=True),
+            _deployed(bundle=True, legacy=legacy),
             Skill(name="review", description="Review work", s3_url=prefix),
         )
 
@@ -298,15 +277,15 @@ async def test_deploy_skill_dicts_keep_their_established_shapes(
 
 
 @pytest.mark.asyncio
-async def test_a_legacy_skill_error_keeps_the_agents_detail(
+async def test_an_inline_skill_error_keeps_the_agents_detail(
     store: GrantingObjectStore, agent: _Agent
 ) -> None:
     agent.status_code = 400
 
-    with pytest.raises(httpx.HTTPStatusError, match="skill_s3_url must be s3://bucket/key"):
+    with pytest.raises(httpx.HTTPStatusError, match="skill_md has no frontmatter"):
         await A2AAgent.add_skill(
             _deployed(bundle=False),
-            Skill(name="review", description="Review work", s3_url="file:///skill/"),
+            Skill(name="review", description="Review work", body="Check it."),
         )
 
 
@@ -386,6 +365,6 @@ async def post(self, url, *, json, timeout):
 
     monkeypatch.setattr(httpx.AsyncClient, "post", post)
 
-    await A2AAgent.add_skill(_deployed(bundle=False), Skill(name="review", description="Review work", s3_url=prefix))
+    await A2AAgent.add_skill(_deployed(bundle=True), Skill(name="review", description="Review work", s3_url=prefix))
 
     assert on_loop and not any(on_loop)
diff --git a/tst/unit/a2a_agent/object_transfer_test.py b/tst/unit/a2a_agent/object_transfer_test.py
index 8ac5158f..9f686707 100644
--- a/tst/unit/a2a_agent/object_transfer_test.py
+++ b/tst/unit/a2a_agent/object_transfer_test.py
@@ -219,23 +219,33 @@ def answer(request: httpx.Request) -> httpx.Response:
     ]
 
 
-@pytest.mark.asyncio
-async def test_fetch_trajectory_reports_a_legacy_agents_own_upload(monkeypatch):
+def _answering(monkeypatch, body: dict) -> None:
     real_client = httpx.AsyncClient
     monkeypatch.setattr(
         object_transfer.httpx,
         "AsyncClient",
         lambda **kwargs: real_client(
-            transport=httpx.MockTransport(
-                lambda request: httpx.Response(200, json={"trajectory_s3_prefix": "s3://b/t/"})
-            ),
+            transport=httpx.MockTransport(lambda request: httpx.Response(200, json=body)),
             **kwargs,
         ),
     )
 
+
+@pytest.mark.asyncio
+async def test_fetch_trajectory_refuses_an_answer_that_only_names_an_s3_prefix(monkeypatch):
+    _answering(monkeypatch, {"trajectory_s3_prefix": "s3://b/t/"})
+
+    with pytest.raises(RuntimeError, match="answered with trajectory_s3_prefix"):
+        await fetch_trajectory("https://agent.test/ext/trajectory", {"task_id": "t"})
+
+
+@pytest.mark.asyncio
+async def test_fetch_trajectory_reads_an_inline_trajectory_beside_a_prefix(monkeypatch):
+    _answering(monkeypatch, {"trajectory": [{"role": "user"}], "trajectory_s3_prefix": "s3://b/t/"})
+
     fetched = await fetch_trajectory("https://agent.test/ext/trajectory", {"task_id": "t"})
 
-    assert fetched == FetchedTrajectory(legacy_prefix="s3://b/t/")
+    assert fetched == FetchedTrajectory(inline=[{"role": "user"}])
 
 
 def test_the_transfer_time_budget_nests():
diff --git a/tst/unit/a2a_agent/test_validator_changelog_objects.py b/tst/unit/a2a_agent/test_validator_changelog_objects.py
index 0c7488ab..a225f57a 100644
--- a/tst/unit/a2a_agent/test_validator_changelog_objects.py
+++ b/tst/unit/a2a_agent/test_validator_changelog_objects.py
@@ -120,35 +120,11 @@ async def test_validator_applies_a_captured_portable_changelog(store, requests):
 
 
 @pytest.mark.asyncio
-async def test_validator_sends_a_legacy_capture_as_its_prefix(store, requests):
+async def test_validator_records_an_apply_agent_without_the_object_form(store, requests):
+    store.put("changelog/run-1/000000.tar", b"increment")
     context = _context(
-        {
-            "object_url": "s3://artifact-bucket/agent_changelog/run-1/capture",
-            "transfer_mode": "legacy",
-        },
+        {"object_url": store.object_url("changelog/run-1"), "transfer_mode": "objects"},
         ["s3_prefix"],
-        ["increments"],
-    )
-
-    verification = await _validate(context)
-
-    assert requests[0]["json"] == {
-        "s3_prefix": "s3://artifact-bucket/agent_changelog/run-1/capture"
-    }
-    assert verification["roundtrip"] is True
-    assert store.granted == []
-
-
-@pytest.mark.asyncio
-async def test_validator_never_signs_a_foreign_namespace_an_agent_echoed(store, requests):
-    for name in ("000000.tar", "000001.tar"):
-        store.put(f"agent_changelog/other-run/solver/{name}", b"someone else's increment")
-    context = _context(
-        {
-            "object_url": store.object_url("agent_changelog/other-run/solver"),
-            "transfer_mode": "legacy",
-        },
-        ["increments"],
     )
 
     verification = await _validate(context)
@@ -156,7 +132,7 @@ async def test_validator_never_signs_a_foreign_namespace_an_agent_echoed(store,
     assert not requests
     assert store.granted == []
     assert verification["apply"] is False
-    assert "cannot apply the configured legacy changelog" in verification["note"]
+    assert "does not advertise the object form" in verification["note"]
 
 
 @pytest.mark.asyncio
diff --git a/tst/unit/a2a_agent/test_validator_skill_variants.py b/tst/unit/a2a_agent/test_validator_skill_variants.py
index 41ea39e2..7239e958 100644
--- a/tst/unit/a2a_agent/test_validator_skill_variants.py
+++ b/tst/unit/a2a_agent/test_validator_skill_variants.py
@@ -43,7 +43,7 @@ async def get(self, url, *, timeout):
         )
 
     async def post(self, url, *, json, timeout):
-        variant = "bundle" if "skill_bundle" in json else "s3"
+        variant = "bundle" if "skill_bundle" in json else "other"
         self.posts.append({"variant": variant, "json": json})
         status_code = 422 if variant in self.rejected_variants else 200
         return httpx.Response(
@@ -91,7 +91,6 @@ def _context(*, bundle: bool) -> TaskStepContext:
         "rubric": {
             "results": [
                 {"id": "secret_code_inline", "score": 1.0},
-                {"id": "secret_code_s3", "score": 1.0},
             ]
         }
     }
@@ -101,10 +100,11 @@ def _context(*, bundle: bool) -> TaskStepContext:
 @pytest.mark.parametrize(
     ("bundle", "rejected_variants", "expected"),
     [
-        (True, set(), {"bundle": True, "s3": True}),
-        (False, set(), {"bundle": False, "s3": True}),
-        (True, {"bundle"}, {"bundle": False, "s3": True}),
+        (True, set(), {"bundle": True}),
+        (False, set(), {"bundle": False}),
+        (True, {"bundle"}, {"bundle": False}),
     ],
+    ids=["bundle", "no-bundle-form", "bundle-rejected"],
 )
 def test_validator_reports_the_artifact_variant_actually_used(
     monkeypatch, tmp_path, bundle, rejected_variants, expected
@@ -137,19 +137,16 @@ def test_validator_reports_the_artifact_variant_actually_used(
             a2a_agent_id="agent-1",
             rubric_verifier_id="rubric",
             skill_bundle_object_url="s3://bucket/bundle/",
-            skill_s3_url="s3://bucket/legacy/",
         ).execute(_context(bundle=bundle))
     )
 
     verification = result.metadata["verifications"]["a2a_skill_config"]
-    assert {name: verification[name] for name in expected} == expected
+    assert verification == {"inline": True, "list": True, **expected}
     options = agent.metadata["validated_a2a_extensions"][
         A2AAgent.EXT_SKILL_CONFIG
     ]["methods"]["add"]["options"]
     assert options["skill_bundle"] == {"supported": expected["bundle"]}
-    assert options["skill_s3_url"] == {"supported": expected["s3"]}
+    assert "skill_s3_url" not in options
     assert options["skill_md"] == {"supported": True}
-    expected_variants = ["s3"]
-    if bundle:
-        expected_variants.insert(0, "bundle")
-    assert [post["variant"] for post in posts] == expected_variants
+    # An agent without the bundle form is never sent the object-backed probe at all.
+    assert [post["variant"] for post in posts] == (["bundle"] if bundle else [])
diff --git a/tst/unit/a2a_agent/validator_trajectory_dag_test.py b/tst/unit/a2a_agent/validator_trajectory_dag_test.py
index 1daa75c7..093469be 100644
--- a/tst/unit/a2a_agent/validator_trajectory_dag_test.py
+++ b/tst/unit/a2a_agent/validator_trajectory_dag_test.py
@@ -19,7 +19,7 @@
 
 from agent_env.a2a_agent import A2AAgent, object_transfer
 from agent_env.a2a_agent.validator import A2AAgentValidator
-from agent_env.config import Config, configure, get_config
+from agent_env.config import Config, configure
 from agent_env.task import Task
 from agent_env.task_step.context import DeployedAgent, TaskStepContext
 from agent_env.task_step.task_step import TaskStep
@@ -95,8 +95,6 @@ async def post(self, _url: str, json: dict, timeout: int) -> _StubResponse:  # n
                     }
                 }
             )
-        if "trajectory_s3_prefix" in json:
-            return _StubResponse({"trajectory_s3_prefix": json["trajectory_s3_prefix"]})
         return _StubResponse({"trajectory": [{"type": "tool_call"}]})
 
 
@@ -123,7 +121,7 @@ def test_trajectory_step_fails_loudly_without_the_mcp_task_id():
 
 
 def test_trajectory_step_probes_with_the_task_id_from_the_mcp_verification(monkeypatch):
-    """The MCP step's `task_id` is load-bearing — it is what both probes carry.
+    """The MCP step's `task_id` is load-bearing — it is what the probes carry.
 
     Renaming the `verifications["a2a_agent_mcp"]["task_id"]` key, or severing
     the DAG edge that populates it, breaks the check.
@@ -138,11 +136,10 @@ def test_trajectory_step_probes_with_the_task_id_from_the_mcp_verification(monke
         step.execute(_context_with_trajectory_agent({"task_id": "task-abc"}))
     )
 
-    assert [p["task_id"] for p in _StubAsyncClient.posts] == ["task-abc", "task-abc"]
+    assert [p["task_id"] for p in _StubAsyncClient.posts] == ["task-abc"]
     assert result.metadata["verifications"]["a2a_trajectory"] == {
         "inline": True,
         "objects": False,
-        "s3": True,
     }
     assert (
         fake_agent.metadata["validated_a2a_extensions"][A2AAgent.EXT_TRAJECTORY][
@@ -152,7 +149,7 @@ def test_trajectory_step_probes_with_the_task_id_from_the_mcp_verification(monke
     )
 
 
-def test_trajectory_step_treats_method_without_request_as_legacy(monkeypatch):
+def test_trajectory_step_treats_method_without_request_as_inline_only(monkeypatch):
     fake_agent = _FakeAgent()
     _StubAsyncClient.posts = []
     monkeypatch.setattr(object_transfer.httpx, "AsyncClient", _StubAsyncClient)
@@ -168,11 +165,11 @@ def test_trajectory_step_treats_method_without_request_as_legacy(monkeypatch):
         )
     )
 
-    assert [set(post) for post in _StubAsyncClient.posts] == [
-        {"task_id"},
-        {"task_id", "trajectory_s3_prefix"},
-    ]
-    assert result.metadata["verifications"]["a2a_trajectory"]["s3"] is True
+    assert [set(post) for post in _StubAsyncClient.posts] == [{"task_id"}]
+    assert result.metadata["verifications"]["a2a_trajectory"] == {
+        "inline": True,
+        "objects": False,
+    }
 
 
 def test_trajectory_step_records_failed_object_url_construction(monkeypatch):
@@ -208,7 +205,6 @@ def object_url(self, _key: str) -> str:
     assert result.metadata["verifications"]["a2a_trajectory"] == {
         "inline": True,
         "objects": False,
-        "s3": False,
     }
 
 
@@ -263,17 +259,15 @@ def test_trajectory_step_probes_the_advertised_object_variant(monkeypatch):
     assert [set(post) for post in _StubAsyncClient.posts] == [
         {"task_id"},
         {"task_id", "objects"},
-        {"task_id", "trajectory_s3_prefix"},
     ]
     assert result.metadata["verifications"]["a2a_trajectory"] == {
         "inline": True,
         "objects": True,
-        "s3": True,
     }
     options = fake_agent.metadata["validated_a2a_extensions"][
         A2AAgent.EXT_TRAJECTORY
     ]["methods"]["get"]["options"]
-    assert options["objects"] == {"supported": True}
+    assert options == {"task_id": {"supported": True}, "objects": {"supported": True}}
 
 
 def test_trajectory_step_skips_the_object_variant_on_a_store_without_grants(monkeypatch):
@@ -359,19 +353,21 @@ async def noop(*_args, **_kwargs) -> None:
 def test_the_trajectory_probe_prefix_is_under_the_fixture_prefix(local_stores, monkeypatch):
     monkeypatch.setenv("AGENT_ENV_FIXTURE_PREFIX", "fx")
     configure()
+    store = _ObjectStore()
     _StubAsyncClient.posts = []
     monkeypatch.setattr(object_transfer.httpx, "AsyncClient", _StubAsyncClient)
     monkeypatch.setattr(A2AAgent, "get", lambda *_a, **_kw: _FakeAgent())
+    monkeypatch.setattr(Config, "get_object_store", lambda self: store)
+    get_method = {"request": {"required": ["task_id"], "oneOf": [{}, {"required": ["objects"]}]}}
 
     asyncio.run(
         VerifyA2ATrajectoryStep(id="t", version=None, a2a_agent_id="agent-x").execute(
-            _context_with_trajectory_agent({"task_id": "task-abc"})
+            _context_with_trajectory_agent({"task_id": "task-abc"}, get_method=get_method)
         )
     )
 
-    (legacy,) = [post for post in _StubAsyncClient.posts if "trajectory_s3_prefix" in post]
-    store = get_config().get_object_store()
-    assert store.get_object_key(legacy["trajectory_s3_prefix"]).startswith("fx/a2a_validator_trajectories/agent-x")
+    (granted,) = store.write_grants
+    assert store.get_object_key(granted).startswith("fx/a2a_validator_trajectories/agent-x/")
 
 
 @pytest.mark.asyncio
@@ -403,7 +399,7 @@ def upload(*_args, **_kwargs):
 
     await A2AAgentValidator.validate(FakeAgent())
 
-    assert on_loop == [False, False, False]
+    assert on_loop == [False, False]
 
 
 @pytest.mark.asyncio
diff --git a/tst/unit/task_step/capture_stubs.py b/tst/unit/task_step/capture_stubs.py
index bb0dff4f..3edfa2ca 100644
--- a/tst/unit/task_step/capture_stubs.py
+++ b/tst/unit/task_step/capture_stubs.py
@@ -1,7 +1,7 @@
 """Shared stubs for the agent-state capture tests.
 
 Follows the ``tst.unit.store.fakes`` convention: the capture path touches an
-artifact store, an object store (including its presigner) and the agent sidecar, and
+artifact store, an object store (including its grants) and the agent sidecar, and
 both ``test_agent_state_capture`` and ``test_prompt_agent_periodic_snapshot`` need
 the same fakes for all three.
 """
@@ -62,7 +62,6 @@ class _StubObjectStore:
 
     def __init__(self):
         self.puts: list[tuple[str, bytes]] = []
-        self.signed_posts: list[dict] = []
         self.write_grants: list[dict] = []
         self.withheld: set[str] = set()
 
@@ -93,13 +92,6 @@ def issue_write_grant(self, object_url, *, media_type, max_bytes, expires_in):
             expires_at=datetime.now(UTC) + timedelta(minutes=5),
         )
 
-    def signed_post(self, url_prefix, *, expires_in=3600, max_bytes=None):
-        self.signed_posts.append(
-            {"url_prefix": url_prefix, "expires_in": expires_in, "max_bytes": max_bytes}
-        )
-        key = f"{self.get_object_key(url_prefix)}${{filename}}"
-        return {"url": f"https://{BUCKET}.s3.amazonaws.com/", "fields": {"key": key}}
-
     def get_object_key(self, url: str) -> str:
         prefix = f"s3://{BUCKET}/"
         if not url.startswith(prefix):
@@ -181,11 +173,12 @@ async def fake_request(self, method, url, *, json=None, timeout=None, **kwargs):
             {"method": method, "url": url, "json": json, "timeout": timeout}
         )
         body = rec.save_body
-        # Callable form: echo a prefix relative to the issued one, which is random.
-        if callable(body):
-            body = body(json or {})
         if body is None:
-            body = {"s3_prefix": (json or {}).get("s3_prefix")}
+            request_json = json or {}
+            body = {
+                "context_id": request_json.get("context_id"),
+                "objects": {name: {"size_bytes": 1} for name in request_json.get("objects", {})},
+            }
         return httpx.Response(rec.save_status, json=body, request=request)
 
     monkeypatch.setattr(httpx.AsyncClient, "request", fake_request)
diff --git a/tst/unit/task_step/test_agent_state_capture.py b/tst/unit/task_step/test_agent_state_capture.py
index c7396276..58fb8b6c 100644
--- a/tst/unit/task_step/test_agent_state_capture.py
+++ b/tst/unit/task_step/test_agent_state_capture.py
@@ -23,10 +23,26 @@
 )
 
 
+def _save_card(*variants: list[str], **params) -> dict:
+    return agent_card(
+        snapshot={
+            "methods": {
+                "save": {
+                    "request": {
+                        "required": ["context_id"],
+                        "oneOf": [{"required": fields} for fields in variants],
+                    }
+                }
+            }
+        },
+        **params,
+    )
+
+
 async def _capture(**overrides):
     kwargs = dict(
         a2a_url="https://agent",
-        a2a_card=agent_card(),
+        a2a_card=_save_card(["objects"]),
         agent_name="solver",
         a2a_context_id="ctx-1",
         artifact_id="wsp",
@@ -36,51 +52,41 @@ async def _capture(**overrides):
     return await mod.capture_workspace(**kwargs)
 
 
+def _granted(rec) -> dict[str, str]:
+    return {
+        grant["object_url"].rsplit("/", 1)[-1]: grant["object_url"]
+        for grant in rec.object_store.write_grants
+    }
+
+
 # ---- capture_workspace ------------------------------------------------------
 
 @pytest.mark.asyncio
-async def test_posts_context_id_and_the_issued_prefix(monkeypatch):
+async def test_posts_context_id_and_a_write_grant_for_each_snapshot_object(monkeypatch):
     rec = install_capture_stubs(monkeypatch)
     result = await _capture()
 
     (sent,) = rec.save_requests
     assert sent["url"] == "https://agent/ext/snapshot"
+    assert set(sent["json"]) == {"context_id", "objects"}
     assert sent["json"]["context_id"] == "ctx-1"
-    assert sent["json"]["s3_prefix"].startswith(f"s3://{BUCKET}/agent_snapshots/wsp/")
-    assert "presigned_post" in sent["json"]
-    # The bundle url is what the sidecar confirmed, wrapped as a universe.
-    assert result.bundle_object_url == sent["json"]["s3_prefix"]
+    assert set(sent["json"]["objects"]) == {"trajectory", "workspace"}
+    assert result.capture_prefix.startswith(f"s3://{BUCKET}/agent_snapshots/wsp/")
+    assert _granted(rec) == {
+        name: f"{result.capture_prefix}{name}" for name in ("trajectory", "workspace")
+    }
+    # Registered where the grants pointed, wrapped as a universe.
+    assert result.bundle_object_url == result.capture_prefix
     assert result.universe_id == "wsp"
-    assert result.capture_prefix == sent["json"]["s3_prefix"]
 
 
 @pytest.mark.asyncio
 async def test_honours_a_card_pinned_endpoint(monkeypatch):
     rec = install_capture_stubs(monkeypatch)
-    await _capture(a2a_card=agent_card(snapshot_endpoint="/custom/snap"))
+    await _capture(a2a_card=_save_card(["objects"], snapshot_endpoint="/custom/snap"))
     assert rec.save_requests[0]["url"] == "https://agent/custom/snap"
 
 
-@pytest.mark.asyncio
-async def test_signs_the_capture_prefix_and_forwards_the_whole_grant(monkeypatch):
-    rec = install_capture_stubs(monkeypatch)
-    await _capture()
-
-    (signed,) = rec.object_store.signed_posts
-    sent = rec.save_requests[0]["json"]
-    assert signed["url_prefix"] == sent["s3_prefix"]
-    assert set(sent["presigned_post"]) == {"url", "fields"}
-
-
-@pytest.mark.asyncio
-async def test_a_backend_that_cannot_sign_omits_the_grant(monkeypatch):
-    rec = install_capture_stubs(monkeypatch)
-    monkeypatch.setattr(rec.object_store, "signed_post", lambda *a, **k: None)
-    await _capture()
-
-    assert "presigned_post" not in rec.save_requests[0]["json"]
-
-
 @pytest.mark.asyncio
 async def test_each_capture_gets_its_own_prefix(monkeypatch):
     """The random suffix is what isolates two captures, NOT the version.
@@ -113,31 +119,9 @@ async def test_raises_on_an_http_error(monkeypatch):
     rec = install_capture_stubs(monkeypatch)
     rec.save_status = 500
     rec.save_body = {"detail": "boom"}
-    with pytest.raises(httpx.HTTPStatusError, match="snapshot save failed with HTTP 500: .*boom"):
+    with pytest.raises(httpx.HTTPStatusError, match="snapshot save failed with HTTP 500"):
         await _capture()
-
-
-@pytest.mark.asyncio
-async def test_raises_when_the_response_omits_the_prefix(monkeypatch):
-    rec = install_capture_stubs(monkeypatch)
-    rec.save_body = {"file_count": 3}
-    with pytest.raises(RuntimeError, match="missing 's3_prefix'"):
-        await _capture()
-
-
-@pytest.mark.asyncio
-async def test_a_prefix_outside_the_issued_one_is_never_registered(monkeypatch, caplog):
-    """`put_existing` lists the prefix and registers every object under it with the
-    worker's credentials, so honouring an agent-chosen prefix would let a
-    compromised sidecar pull any prefix that role can read into this snapshot."""
-    rec = install_capture_stubs(monkeypatch)
-    rec.save_body = {"s3_prefix": f"s3://{BUCKET}/someone/elses/run/"}
-    result = await _capture()
-
-    issued = rec.save_requests[0]["json"]["s3_prefix"]
-    assert result.bundle_object_url == issued
-    assert "someone/elses/run" not in result.bundle_object_url
-    assert "echoed a prefix outside the one issued" in caplog.text
+    assert not rec.universes
 
 
 @pytest.mark.asyncio
@@ -145,72 +129,18 @@ async def test_a_capture_is_issued_under_the_fixture_prefix(monkeypatch):
     rec = install_capture_stubs(monkeypatch)
     rec.key_prefix = "fx/"
     result = await _capture()
-    assert rec.save_requests[0]["json"]["s3_prefix"].startswith(f"s3://{BUCKET}/fx/agent_snapshots/wsp/")
     assert result.capture_prefix.startswith(f"s3://{BUCKET}/fx/agent_snapshots/wsp/")
+    assert all(url.startswith(result.capture_prefix) for url in _granted(rec).values())
 
 
 @pytest.mark.asyncio
-async def test_a_sidecar_that_nests_under_the_issued_prefix_is_registered_there(monkeypatch):
-    """Registering the issued prefix for a nester fails silently — the recursive list
-    still finds the object, and only a later restore cannot."""
+async def test_a_card_offering_both_save_forms_gets_the_object_one(monkeypatch):
     rec = install_capture_stubs(monkeypatch)
-    stamp = "2026-08-05T00:00:00.000Z"
-    rec.save_body = lambda req: {"s3_prefix": f"{req['s3_prefix']}{stamp}/"}
-    result = await _capture()
-
-    issued = rec.save_requests[0]["json"]["s3_prefix"]
-    assert result.bundle_object_url == f"{issued}{stamp}/"
-    # The presigned prefix is unchanged — per-service state still lands beside it.
-    assert result.capture_prefix.startswith(f"s3://{BUCKET}/agent_snapshots/wsp/")
 
+    result = await _capture(a2a_card=_save_card(["s3_prefix"], ["objects"]))
 
-@pytest.mark.asyncio
-async def test_a_flat_echo_registers_the_issued_prefix_unchanged(monkeypatch, caplog):
-    """A sidecar honouring the presigned POST writes flat and echoes it back."""
-    rec = install_capture_stubs(monkeypatch)
-    result = await _capture()
-
-    issued = rec.save_requests[0]["json"]["s3_prefix"]
-    assert result.bundle_object_url == issued
-    assert "outside the one issued" not in caplog.text
-
-
-@pytest.mark.asyncio
-async def test_snapshot_save_prefers_advertised_object_mode(monkeypatch):
-    rec = install_capture_stubs(monkeypatch)
-    rec.save_body = {
-        "context_id": "ctx-1",
-        "objects": {
-            "trajectory": {"size_bytes": 12},
-            "workspace": {"size_bytes": 34},
-        },
-    }
-    card = agent_card(
-        snapshot={
-            "methods": {
-                "save": {
-                    "request": {
-                        "required": ["context_id"],
-                        "oneOf": [
-                            {"required": ["s3_prefix"]},
-                            {"required": ["objects"]},
-                        ],
-                    }
-                }
-            }
-        }
-    )
-
-    result = await _capture(a2a_card=card)
-
-    sent = rec.save_requests[0]["json"]
-    assert set(sent) == {"context_id", "objects"}
-    assert set(sent["objects"]) == {"trajectory", "workspace"}
-    assert not rec.object_store.signed_posts
-    assert {
-        grant["object_url"].rsplit("/", 1)[-1]
-        for grant in rec.object_store.write_grants
-    } == {"trajectory", "workspace"}
+    assert set(rec.save_requests[0]["json"]) == {"context_id", "objects"}
+    assert set(_granted(rec)) == {"trajectory", "workspace"}
     assert result.bundle_object_url == result.capture_prefix
 
 
@@ -218,61 +148,36 @@ async def test_snapshot_save_prefers_advertised_object_mode(monkeypatch):
 async def test_snapshot_save_requires_every_supplied_object(monkeypatch):
     rec = install_capture_stubs(monkeypatch)
     rec.object_store.withheld = {"workspace"}
-    rec.save_body = {
-        "context_id": "ctx-1",
-        "objects": {
-            "trajectory": {"size_bytes": 12},
-            "workspace": {"size_bytes": 34},
-        },
-    }
-    card = agent_card(
-        snapshot={
-            "methods": {
-                "save": {
-                    "request": {"required": ["context_id", "objects"]}
-                }
-            }
-        }
-    )
 
     with pytest.raises(RuntimeError, match=r"left \['workspace'\] out of the object store"):
-        await _capture(a2a_card=card)
+        await _capture()
     assert not rec.universes
 
 
-def _save_card(*variants: list[str]) -> dict:
-    return agent_card(
-        snapshot={
-            "methods": {
-                "save": {
-                    "request": {
-                        "required": ["context_id"],
-                        "oneOf": [{"required": fields} for fields in variants],
-                    }
-                }
-            }
-        }
-    )
-
-
 @pytest.mark.asyncio
-async def test_snapshot_save_takes_the_legacy_shape_on_a_store_without_grants(monkeypatch):
+@pytest.mark.parametrize(
+    "card",
+    [_save_card(["s3_prefix"]), agent_card()],
+    ids=["s3-prefix-only", "no-declared-request"],
+)
+async def test_an_agent_without_the_object_save_form_is_refused_before_any_request(
+    monkeypatch, card
+):
     rec = install_capture_stubs(monkeypatch)
-    monkeypatch.setattr(rec.object_store, "supports_transfer_grants", False)
 
-    await _capture(a2a_card=_save_card(["s3_prefix"], ["objects"]))
+    with pytest.raises(RuntimeError, match="does not advertise the object form"):
+        await _capture(a2a_card=card)
 
-    assert "s3_prefix" in rec.save_requests[0]["json"]
-    assert rec.object_store.write_grants == []
+    assert (rec.save_requests, rec.object_store.write_grants, rec.universes) == ([], [], [])
 
 
 @pytest.mark.asyncio
-async def test_an_objects_only_save_on_a_store_without_grants_is_refused(monkeypatch):
+async def test_a_save_on_a_store_without_grants_is_refused(monkeypatch):
     rec = install_capture_stubs(monkeypatch)
     monkeypatch.setattr(rec.object_store, "supports_transfer_grants", False)
 
-    with pytest.raises(RuntimeError, match="no snapshot save form"):
-        await _capture(a2a_card=_save_card(["objects"]))
+    with pytest.raises(RuntimeError, match="does not issue transfer grants"):
+        await _capture()
 
     assert rec.save_requests == []
 
diff --git a/tst/unit/task_step/test_deploy_agent_changelog_objects.py b/tst/unit/task_step/test_deploy_agent_changelog_objects.py
index 6d915302..e65115ed 100644
--- a/tst/unit/task_step/test_deploy_agent_changelog_objects.py
+++ b/tst/unit/task_step/test_deploy_agent_changelog_objects.py
@@ -155,27 +155,9 @@ async def test_the_changelog_namespace_is_under_the_fixture_prefix(monkeypatch,
 
 
 @pytest.mark.asyncio
-async def test_enable_uses_legacy_variant_when_the_store_cannot_grant_the_lifetime(
-    monkeypatch, store
-):
-    def unavailable(*args, **kwargs):
-        raise GrantUnavailableError("credentials expire first")
-
-    monkeypatch.setattr(store, "issue_upload_policy", unavailable)
-    requests = _install(monkeypatch, {"s3_prefix": store.object_url(NAMESPACE_KEY)})
-    context = _run_context()
-
-    await _step(enable_agent_changelog=True)._configure_agent_changelog(
-        "https://agent", _card(), context, expires_in=28_800
-    )
-
-    assert requests[0]["json"] == {"s3_prefix": store.object_url(NAMESPACE_KEY)}
-    assert context.metadata["agent_changelog"][0]["transfer_mode"] == "legacy"
-
-
-@pytest.mark.asyncio
-async def test_enable_fails_for_a_portable_only_agent_when_the_store_cannot_grant(
-    monkeypatch, store
+@pytest.mark.parametrize("enable_legacy", [True, False], ids=["dual", "portable-only"])
+async def test_enable_fails_when_the_store_cannot_grant_the_lifetime(
+    monkeypatch, store, enable_legacy
 ):
     def unavailable(*args, **kwargs):
         raise GrantUnavailableError("credentials expire first")
@@ -184,11 +166,12 @@ def unavailable(*args, **kwargs):
     requests = _install(monkeypatch, {"roots": ["workspace"]})
 
     with pytest.raises(
-        GrantUnavailableError, match="'solver' takes only the object.*credentials expire first"
+        GrantUnavailableError,
+        match="'solver': the object store cannot issue its namespace grant: credentials expire first",
     ):
         await _step(enable_agent_changelog=True)._configure_agent_changelog(
             "https://agent",
-            _card(enable_legacy=False),
+            _card(enable_legacy=enable_legacy),
             _run_context(),
             expires_in=28_800,
         )
@@ -197,61 +180,31 @@ def unavailable(*args, **kwargs):
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "enable_legacy, expected", [(True, "legacy"), (False, None)], ids=["dual", "portable-only"]
-)
-async def test_enable_on_a_store_without_grants_asks_for_no_grant(
-    monkeypatch, store, enable_legacy, expected
-):
+@pytest.mark.parametrize("enable_legacy", [True, False], ids=["dual", "portable-only"])
+async def test_enable_on_a_store_without_grants_is_refused(monkeypatch, store, enable_legacy):
     store.supports_transfer_grants = False
-    requests = _install(monkeypatch, {"s3_prefix": store.object_url(NAMESPACE_KEY)})
-    context = _run_context()
-    step = _step(enable_agent_changelog=True)
-
-    if expected is None:
-        with pytest.raises(RuntimeError, match="no changelog enable form"):
-            await step._configure_agent_changelog(
-                "https://agent", _card(enable_legacy=enable_legacy), context, expires_in=7_200
-            )
-        assert not requests
-    else:
-        await step._configure_agent_changelog(
-            "https://agent", _card(enable_legacy=enable_legacy), context, expires_in=7_200
-        )
-        assert context.metadata["agent_changelog"][0]["transfer_mode"] == expected
-    assert store.granted == []
-
-
-@pytest.mark.asyncio
-async def test_enable_records_the_issued_prefix_when_the_agent_echoes_a_foreign_one(
-    monkeypatch, store
-):
-    _install(
-        monkeypatch,
-        {"s3_prefix": "s3://other-bucket/agent_changelog/other-run/solver", "roots": ["/"]},
-    )
-    context = _run_context()
+    requests = _install(monkeypatch, {"roots": ["workspace"]})
 
-    await _step(enable_agent_changelog=True)._configure_agent_changelog(
-        "https://agent", _card(enable_objects=False), context, expires_in=7_200
-    )
+    with pytest.raises(RuntimeError, match="does not issue transfer grants"):
+        await _step(enable_agent_changelog=True)._configure_agent_changelog(
+            "https://agent", _card(enable_legacy=enable_legacy), _run_context(), expires_in=7_200
+        )
 
-    (entry,) = context.metadata["agent_changelog"]
-    assert entry["object_url"] == store.object_url(NAMESPACE_KEY)
-    assert entry["transfer_mode"] == "legacy"
+    assert not requests
+    assert store.granted == []
 
 
 @pytest.mark.asyncio
-async def test_a_legacy_enable_error_keeps_the_agents_detail(monkeypatch, store):
-    _install(
-        monkeypatch, {"detail": "s3_prefix must be s3://bucket/key"}, status_code=400
-    )
+async def test_enable_is_refused_for_an_agent_without_the_object_form(monkeypatch, store):
+    requests = _install(monkeypatch, {"roots": ["workspace"]})
 
-    with pytest.raises(httpx.HTTPStatusError, match="s3_prefix must be s3://bucket/key"):
+    with pytest.raises(RuntimeError, match="does not advertise the object form"):
         await _step(enable_agent_changelog=True)._configure_agent_changelog(
             "https://agent", _card(enable_objects=False), _run_context(), expires_in=7_200
         )
 
+    assert not requests
+
 
 @pytest.mark.asyncio
 async def test_apply_lists_validates_and_sends_ordered_read_grants(monkeypatch, store):
@@ -367,7 +320,7 @@ async def test_apply_sends_the_empty_baseline_when_no_increment_is_selected(
 
 def _apply_call(store, source_url):
     return changelog_apply_call(
-        _APPLY_OBJECTS, store, agent_name="solver", source_url=source_url, portable=True
+        _APPLY_OBJECTS, store, agent_name="solver", source_url=source_url
     )
 
 
@@ -408,39 +361,11 @@ def test_apply_counts_empty_increments_as_capture_does(monkeypatch, store):
 
 
 @pytest.mark.asyncio
-async def test_apply_uses_legacy_variant_for_a_legacy_source(monkeypatch, store):
-    requests = _install(monkeypatch, {"ok": True, "count": 2})
-    step = _step(
-        agent_changelog_s3_prefix="s3://artifact-bucket/agent_changelog/run-1/solver",
-        agent_changelog_toolcall_position_exclusive=1,
-    )
-    context = TaskStepContext()
-
-    await step._apply_agent_changelog("https://agent", _card(), context)
-
-    assert requests[0]["json"] == {
-        "s3_prefix": "s3://artifact-bucket/agent_changelog/run-1/solver",
-        "resume_conversation": True,
-        "up_to_tool_call_exclusive": 1,
-    }
-    assert store.granted == []
-    assert context.metadata["agent_changelog_rewinds"] == [
-        {
-            "agent_name": "solver",
-            "context_id": None,
-            "up_to_tool_call_exclusive": 1,
-            "transfer_mode": "legacy",
-            "object_url": "s3://artifact-bucket/agent_changelog/run-1/solver",
-        }
-    ]
-
-
-@pytest.mark.asyncio
-async def test_a_portable_source_is_not_sent_to_a_legacy_only_agent(monkeypatch, store):
+async def test_a_source_is_not_sent_to_an_agent_without_the_object_apply_form(monkeypatch, store):
     namespace = _increments(store, "000000.tar")
     requests = _install(monkeypatch, {"ok": True, "count": 1})
 
-    with pytest.raises(RuntimeError, match="cannot apply the configured portable changelog"):
+    with pytest.raises(RuntimeError, match="does not advertise the object form"):
         await _step(agent_changelog_object_url=namespace)._apply_agent_changelog(
             "https://agent", _card(apply_objects=False), TaskStepContext()
         )
@@ -449,7 +374,7 @@ async def test_a_portable_source_is_not_sent_to_a_legacy_only_agent(monkeypatch,
     assert store.granted == []
 
 
-def test_portable_changelog_source_round_trips_without_reusing_s3_field():
+def test_a_changelog_source_round_trips():
     step = _step(
         agent_changelog_object_url="s3://artifact-bucket/agent_changelog/run-1/solver",
         agent_changelog_toolcall_position_exclusive=2,
@@ -458,12 +383,9 @@ def test_portable_changelog_source_round_trips_without_reusing_s3_field():
     restored = DeployAgentTaskStep.from_dict(step.to_dict())
 
     assert restored.agent_changelog_object_url == step.agent_changelog_object_url
-    assert restored.agent_changelog_s3_prefix is None
+    assert restored.agent_changelog_toolcall_position_exclusive == 2
 
 
-def test_changelog_sources_are_mutually_exclusive():
-    with pytest.raises(ValueError, match="set only one"):
-        _step(
-            agent_changelog_object_url="s3://artifact-bucket/portable",
-            agent_changelog_s3_prefix="s3://artifact-bucket/legacy",
-        )
+def test_a_cutoff_needs_a_changelog_source():
+    with pytest.raises(ValueError, match="requires agent_changelog_object_url"):
+        _step(agent_changelog_toolcall_position_exclusive=2)
diff --git a/tst/unit/task_step/test_deploy_agent_snapshot_objects.py b/tst/unit/task_step/test_deploy_agent_snapshot_objects.py
index 434df771..d58f9ec6 100644
--- a/tst/unit/task_step/test_deploy_agent_snapshot_objects.py
+++ b/tst/unit/task_step/test_deploy_agent_snapshot_objects.py
@@ -141,36 +141,24 @@ def recording(url):
     assert loop_threads == [False, False]
 
 
-EXPIRED_SIGNED_URL = (
-    "https://objects.s3.region.example.test/agent_snapshots/snapshot-1/"
-    "?X-Amz-Date=20000101T000000Z&X-Amz-Expires=3600&X-Amz-Signature=0"
-)
-
-
 @pytest.mark.asyncio
-@pytest.mark.parametrize("signed", [False, True], ids=["store-url", "expired-signed-url"])
-async def test_load_sends_a_legacy_snapshot_whole_as_its_prefix(monkeypatch, store, signed):
-    universe = _Universe(store, LEGACY)
-    if signed:
-        universe.bundle_object_url = EXPIRED_SIGNED_URL
-    requests = _install(monkeypatch, universe, {"ok": True, "context_id": "legacy-context"})
-    context = TaskStepContext()
+@pytest.mark.parametrize(
+    "card, files",
+    [
+        (_card(), LEGACY),
+        (_card(objects=False), LEGACY),
+        (_card(legacy=False), ("workspace.tar.gz",)),
+    ],
+    ids=["dual-agent", "legacy-only-agent", "objects-only-agent"],
+)
+async def test_a_legacy_snapshot_cannot_be_restored(monkeypatch, store, card, files):
+    requests = _install(monkeypatch, _Universe(store, files), {"context_id": "wrong"})
 
-    await _step()._load_snapshot("https://agent", _card(), "agent-1", context)
+    with pytest.raises(RuntimeError, match="is not a portable snapshot, so it cannot be restored"):
+        await _step()._load_snapshot("https://agent", card, "agent-1", TaskStepContext())
 
-    assert requests[0]["json"] == {
-        "s3_prefix": universe.bundle_object_url,
-        "target_context_id": "restored-context",
-    }
     assert not store.granted
-    assert context.metadata["agent_loaded_snapshots"] == [
-        {
-            "agent_name": "solver",
-            "context_id": "legacy-context",
-            "source_artifact_id": "snapshot-1",
-            "source_artifact_version": 3,
-        }
-    ]
+    assert not requests
 
 
 @pytest.mark.asyncio
@@ -201,49 +189,21 @@ async def test_load_ignores_response_fields_it_does_not_know(monkeypatch, store)
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    "card, grants",
-    [(_card(objects=False), True), (_card(), False)],
+    "card, grants, refusal",
+    [
+        (_card(objects=False), True, "does not advertise the object form"),
+        (_card(), False, "does not issue transfer grants"),
+    ],
     ids=["legacy-only-agent", "store-without-grants"],
 )
 async def test_a_portable_snapshot_loads_through_objects_or_not_at_all(
-    monkeypatch, store, card, grants
+    monkeypatch, store, card, grants, refusal
 ):
     store.supports_transfer_grants = grants
     requests = _install(monkeypatch, _Universe(store, PORTABLE[:1]), {"context_id": "wrong"})
 
-    with pytest.raises(RuntimeError, match="cannot load the configured portable"):
+    with pytest.raises(RuntimeError, match=refusal):
         await _step()._load_snapshot("https://agent", card, "agent-1", TaskStepContext())
 
     assert not store.granted
     assert not requests
-
-
-@pytest.mark.asyncio
-async def test_legacy_snapshot_is_not_sent_to_an_objects_only_agent(monkeypatch, store):
-    requests = _install(
-        monkeypatch, _Universe(store, ("workspace.tar.gz",)), {"context_id": "wrong"}
-    )
-
-    with pytest.raises(RuntimeError, match="cannot load the configured legacy"):
-        await _step()._load_snapshot(
-            "https://agent",
-            _card(legacy=False),
-            "agent-1",
-            TaskStepContext(),
-        )
-
-    assert not store.granted
-    assert not requests
-
-
-@pytest.mark.asyncio
-async def test_a_legacy_load_error_keeps_the_agents_detail(monkeypatch, store):
-    _install(
-        monkeypatch,
-        _Universe(store, LEGACY),
-        {"detail": "cannot load a workspace while an agent turn is running"},
-        status_code=409,
-    )
-
-    with pytest.raises(httpx.HTTPStatusError, match="while an agent turn is running"):
-        await _step()._load_snapshot("https://agent", _card(), "agent-1", TaskStepContext())
diff --git a/tst/unit/task_step/test_prompt_agent.py b/tst/unit/task_step/test_prompt_agent.py
index ae0aeacd..177c8943 100644
--- a/tst/unit/task_step/test_prompt_agent.py
+++ b/tst/unit/task_step/test_prompt_agent.py
@@ -22,7 +22,7 @@
 from agentenv_protocol.transfers import HttpPutGrant
 
 from agent_env.a2a_agent.object_transfer import DEFAULT_TRAJECTORY_MAX_BYTES
-from agent_env.config import configure, set_object_store
+from agent_env.config import configure
 from agent_env.store.object_store import LocalFilesystemObjectStore
 from agent_env.task_step.context import DeployedAgent, TaskStepContext
 from agent_env.task_step.snapshot_utils import agent_state_capture
@@ -272,17 +272,3 @@ def agent(request: httpx.Request) -> httpx.Response:
     assert json.loads(store.get(uri)) == [{"type": "echo", "output": "Echo: hi"}]
 
 
-def test_agent_written_trajectory_is_found_on_the_local_store(monkeypatch, tmp_path):
-    store = LocalFilesystemObjectStore(str(tmp_path))
-    set_object_store(store)
-    step = PromptAgentTaskStep(id="p", version=None, prompt="hi", prompt_id="pid", agent_name="a")
-    prefix = step._trajectory_prefix()
-    written = store.put(f"{store.get_object_key(prefix)}/trajectory-1.json", b"{}")
-    client = MagicMock()
-    client.__aenter__.return_value.post = AsyncMock(return_value=SimpleNamespace(
-        raise_for_status=lambda: None, json=lambda: {"trajectory_s3_prefix": prefix},
-    ))
-    monkeypatch.setattr(pa.httpx, "AsyncClient", lambda *a, **k: client)
-
-    uri = asyncio.run(step._fetch_trajectory("http://a", {"params": {"endpoint": "/t"}}, "SERVER-ID", prefix, "CLIENT-ID"))
-    assert uri == written
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 7386ffbf..73d67f53 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
@@ -162,7 +162,7 @@ def fake_upload(trajectory, prefix, *, name=None):
 
 
 @pytest.mark.asyncio
-async def test_server_side_s3_prefix_is_listed_for_the_object_url(monkeypatch):
+async def test_a_judge_naming_its_own_trajectory_prefix_yields_no_trajectory(monkeypatch):
     verifier = _verifier()
 
     async def fake_request(self, method, url, **kwargs):
@@ -180,8 +180,7 @@ def object_url(self, key):
             return f"s3://bucket/{key}"
 
         def list_at(self, prefix):
-            assert prefix == "s3://bucket/pre/"
-            return ["s3://bucket/pre/trajectory-t1.json"]
+            raise AssertionError("an agent-named prefix is never listed")
 
     _use_store(monkeypatch, FakeStore())
 
@@ -190,7 +189,7 @@ def list_at(self, prefix):
         judge_agent_card=_card_with_trajectory_ext(),
         a2a_server_task_id="t1",
     )
-    assert uri == "s3://bucket/pre/trajectory-t1.json"
+    assert uri is None
 
 
 @pytest.mark.asyncio

From b5197c70acdef50d4250944925569e777993fde7 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Mon, 5 Oct 2026 08:34:02 -0700
Subject: [PATCH 2/3] test(a2a): stored steps with a removed S3-form field
 still load

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../task_step/stored_s3_form_fields_test.py   | 21 +++++++++++++++++++
 1 file changed, 21 insertions(+)
 create mode 100644 tst/unit/task_step/stored_s3_form_fields_test.py

diff --git a/tst/unit/task_step/stored_s3_form_fields_test.py b/tst/unit/task_step/stored_s3_form_fields_test.py
new file mode 100644
index 00000000..f071f4af
--- /dev/null
+++ b/tst/unit/task_step/stored_s3_form_fields_test.py
@@ -0,0 +1,21 @@
+"""Stored step documents written before the S3 transfer forms were removed still load, and drop
+the key."""
+
+import pytest
+
+from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_skill_config import (
+    VerifyA2ASkillConfigStep,
+)
+from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
+
+
+@pytest.mark.parametrize("step, key", [
+    (DeployAgentTaskStep(id="d", version=None), "agent_changelog_s3_prefix"),
+    (
+        VerifyA2ASkillConfigStep(id="v", version=None, a2a_agent_id="a", rubric_verifier_id="r"),
+        "skill_s3_url",
+    ),
+], ids=["deploy_agent", "verify_a2a_skill_config"])
+def test_a_stored_step_with_an_s3_form_field_still_loads(step, key):
+    document = step.to_dict()
+    assert type(step).from_dict({**document, key: "s3://bucket/prefix/"}).to_dict() == document

From 3caecf25f157889012da3c18d2cb7a937624ce8c Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Mon, 5 Oct 2026 10:31:37 -0700
Subject: [PATCH 3/3] fix(task-step): refuse a stored deploy_agent step with an
 S3-form changelog source

Dropping agent_changelog_s3_prefix on load left the step loadable but skipped the rewind it asked
for. A stored step that still names one is now refused; an empty field, which every stored step
carries, still loads.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../task_step/task_steps/deploy_agent.py      |  5 +++
 .../task_step/stored_s3_form_fields_test.py   | 33 ++++++++++++-------
 2 files changed, 27 insertions(+), 11 deletions(-)

diff --git a/src/agent_env/task_step/task_steps/deploy_agent.py b/src/agent_env/task_step/task_steps/deploy_agent.py
index 6dc17122..40824353 100644
--- a/src/agent_env/task_step/task_steps/deploy_agent.py
+++ b/src/agent_env/task_step/task_steps/deploy_agent.py
@@ -194,6 +194,11 @@ def to_dict(self) -> dict:
 
     @classmethod
     def from_dict(cls, data: dict) -> DeployAgentTaskStep:
+        if data.get("agent_changelog_s3_prefix"):
+            raise ValueError(
+                "agent_changelog_s3_prefix is no longer supported: an S3-form changelog cannot "
+                "be applied; capture it again and set agent_changelog_object_url"
+            )
         return cls(
             **cls._base_from_dict(data),
             env_ids=data.get("env_ids"),
diff --git a/tst/unit/task_step/stored_s3_form_fields_test.py b/tst/unit/task_step/stored_s3_form_fields_test.py
index f071f4af..0541a6ea 100644
--- a/tst/unit/task_step/stored_s3_form_fields_test.py
+++ b/tst/unit/task_step/stored_s3_form_fields_test.py
@@ -1,5 +1,5 @@
-"""Stored step documents written before the S3 transfer forms were removed still load, and drop
-the key."""
+"""Stored step documents written before the S3 transfer forms were removed: a field that only
+named an S3 location is dropped, and an S3-form changelog source is refused rather than skipped."""
 
 import pytest
 
@@ -9,13 +9,24 @@
 from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
 
 
-@pytest.mark.parametrize("step, key", [
-    (DeployAgentTaskStep(id="d", version=None), "agent_changelog_s3_prefix"),
-    (
-        VerifyA2ASkillConfigStep(id="v", version=None, a2a_agent_id="a", rubric_verifier_id="r"),
-        "skill_s3_url",
-    ),
-], ids=["deploy_agent", "verify_a2a_skill_config"])
-def test_a_stored_step_with_an_s3_form_field_still_loads(step, key):
+def test_a_stored_skill_config_check_with_an_s3_probe_url_still_loads():
+    step = VerifyA2ASkillConfigStep(id="v", version=None, a2a_agent_id="a", rubric_verifier_id="r")
     document = step.to_dict()
-    assert type(step).from_dict({**document, key: "s3://bucket/prefix/"}).to_dict() == document
+    assert VerifyA2ASkillConfigStep.from_dict({**document, "skill_s3_url": "s3://bucket/prefix/"}).to_dict() == document
+
+
+@pytest.mark.parametrize("cutoff", [None, 3], ids=["whole-changelog", "with-cutoff"])
+def test_a_stored_deploy_step_with_an_s3_changelog_source_is_refused(cutoff):
+    document = {
+        **DeployAgentTaskStep(id="d", version=None).to_dict(),
+        "agent_changelog_s3_prefix": "s3://bucket/agent_changelog/run-1/solver",
+        "agent_changelog_toolcall_position_exclusive": cutoff,
+    }
+    with pytest.raises(ValueError, match="agent_changelog_s3_prefix is no longer supported"):
+        DeployAgentTaskStep.from_dict(document)
+
+
+def test_a_stored_deploy_step_with_an_empty_s3_changelog_field_still_loads():
+    step = DeployAgentTaskStep(id="d", version=None)
+    document = step.to_dict()
+    assert DeployAgentTaskStep.from_dict({**document, "agent_changelog_s3_prefix": None}).to_dict() == document