refactor(a2a)!: remove the S3 transfer forms

packages/agentenv-protocol/README.md

@@ -568,20 +568,17 @@ 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.
 
 #### Staging
 

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,

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"))

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}")

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")