feat(sdk): attachment block ingress, model capabilities, and the delivery chain

docs/design/agent-workflows/projects/agent-multi-modality/protocols/stage-1.md

@@ -253,3 +253,58 @@ unrelated to this stage.
   `attachment_delivery` record, because the legacy path has no attachment id to key the event.
   Honest visibility for pasted images arrives with WP4, when the front end switches them to real
   attachments.
+
+## WP3: the SDK and agent-service producer
+
+### Implementation decisions worth knowing (beyond the plan)
+
+- The stable failure code is SDK-owned (the runner result carries only a string today), named
+  `failure_code` because the errors module already uses `code` for an integer HTTP status.
+- `attachment_delivery` events parse through the generic event path; the plan's dedicated parser
+  branch would have duplicated what the generic path already preserves.
+- The model catalog's docstring now states that its `modalities` field feeds the runtime delivery
+  gate through the connection resolver; the catalog itself still gates nothing.
+
+### The review, and what it changed
+
+The adversarial review's headline finding was verified empirically against the pinned AI SDK:
+the first implementation put the stable error code on the Vercel error frame, and the SDK
+validates that frame with a strict schema that rejects unknown keys, so every error-carrying
+stream would have aborted with an opaque parse failure instead of rendering the sanitized
+message. The code now travels in a `data-agent-error` part emitted before the standard two-key
+error frame, with per-site codes (`runner_error`, `no_output`, the exception's own code, or the
+default).
+
+Second finding: the pinned SDK's file part has no top-level `size` field and validation strips
+extras, so the golden was pinning a value production could never send; `size` moved into the
+`providerMetadata.agenta` envelope beside the attachment id.
+
+Third, the review settled the plan's open key-space question with an end-to-end trace: the Pi
+path resolves for the common case (`provider/model` ids match the catalog keys exactly), the
+Claude picker aliases resolve, but bare dated Anthropic ids missed, which would have silently
+gated every dated-id Claude run's uploads to workspace-only. Closed by falling back to the Pi
+catalog's `anthropic/<id>` entry, a second read of the same sourced fact, not a guess.
+
+Fourth, a semantics defect at the WP2 seam, fixed on the WP2 lane: both catalogs only ever
+enumerate text and image, so the runner's gate treating a kind's absence as "unsupported"
+asserted a false negative for documents; absence now reads as unknown (workspace-only with the
+unknown reason), and the unsupported code is reserved for a catalog that can genuinely state
+negatives.
+
+### The stack, restructured again for the same reason
+
+The typed-failure work and the contract-test additions build on the sandbox-slug rename and the
+Pi-builtins changes in the same file regions, so those two parallel lanes were linearized into
+the train below WP3 (the same dependent-hunks refusal as WP2's case, caught the same way: the
+tool's partial-commit warning plus a tree-versus-tip diff). Consequence: PR #5597 now merges in
+the train after WP2; the sandbox-slug content already merged independently as #5585 and its lane
+dissolves on the next rebase.
+
+### Forced routes to double-check
+
+- **A catalog miss means workspace-only for that model's uploads.** The honesty rule's cost:
+  a model absent from both catalogs delivers attachments to the workspace with a notice until
+  the catalog data learns it. Closing a miss is a data addition, not a code change.
+- **The catalogs cannot express document or audio support today**, so native document delivery
+  (Stage 2) will need the catalog schema to grow before the gate can ever say yes; the gate's
+  absence-means-unknown rule is what keeps that honest in the meantime.

sdks/python/agenta/sdk/agents/__init__.py

@@ -83,6 +83,7 @@
     to_messages,
 )
 from .errors import (
+    AgentRunFailed,
     AgentRunnerConfigurationError,
     LocalSandboxNotAllowedError,
     SandboxNotAllowedError,
@@ -262,6 +263,7 @@
     "Environment",
     "Harness",
     # Errors
+    "AgentRunFailed",
     "AgentRunnerConfigurationError",
     "SandboxNotAllowedError",
     "LocalSandboxNotAllowedError",

sdks/python/agenta/sdk/agents/adapters/vercel/messages.py

@@ -3,10 +3,14 @@
 This adapter translates between the Vercel AI SDK ``UIMessage`` parts shape and the
 neutral agent runtime ``Message`` / ``ContentBlock`` types. The neutral DTOs stay the port;
 Vercel-specific part names live here.
+
+Attachment references are ingress-only. A neutral ``attachment`` block has no URL, so
+``_block_to_parts`` cannot reconstruct a Vercel ``FileUIPart`` from it.
 """
 
 from __future__ import annotations
 
+import re
 from typing import Any, Dict, List, Optional
 
 from agenta.sdk.utils.logging import get_module_logger
@@ -29,6 +33,10 @@
     TOOL_OUTPUT_ERROR,
     TOOL_OUTPUT_DENIED,
 }
+# Lowercase-only is deliberate and pinned by the invalid-metadata test.
+_CANONICAL_UUID = re.compile(
+    r"[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}"
+)
 
 
 def vercel_messages_to_agenta_messages(raw: Optional[List[Any]]) -> List[Message]:
@@ -68,17 +76,17 @@ def _part_to_blocks(part: Any) -> List[ContentBlock]:
 
     Agenta's ``ContentBlock`` model is canonical. This mapper only ever produces
     ``ContentBlock`` types Agenta already defines internally (``text``, ``image``,
-    ``resource``, ``tool_call``, ``tool_result``). To support a new Vercel part kind,
+    ``resource``, ``attachment``, ``tool_call``, ``tool_result``). To support a new Vercel part kind,
     first add first-class support for it in the Agenta ``ContentBlock`` model, then map
     it here — never fabricate an adapter-specific block type or pass an unmapped kind
     through opaquely. A part kind Agenta does not define is dropped (observably via a
     debug log), not fabricated.
 
     Keep this channel symmetric: a kind must be handled in both directions or neither. If
     something is mapped inbound it must map outbound too (and vice versa); a kind dropped
-    here must also be absent in ``_block_to_parts`` — never add one side alone. The only
-    exception is a direction that explicitly cannot occur (e.g. the one-way live event
-    stream in ``stream.py``, which has no inbound counterpart by design).
+    here must also be absent in ``_block_to_parts`` — never add one side alone. Attachment
+    references are the module-documented ingress-only exception because their neutral block
+    has no URL.
 
     Reasoning is such a stream-only concept: the live event stream maps the ``thought``
     event to Vercel ``reasoning`` frames. Stored ``UIMessage`` conversion has no reasoning
@@ -96,6 +104,32 @@ def _part_to_blocks(part: Any) -> List[ContentBlock]:
 
     if ptype == "file":
         media = part.get("mediaType") or part.get("mimeType")
+        provider_metadata = part.get("providerMetadata")
+        agenta_metadata = (
+            provider_metadata.get("agenta")
+            if isinstance(provider_metadata, dict)
+            else None
+        )
+        attachment_id = (
+            agenta_metadata.get("attachmentId")
+            if isinstance(agenta_metadata, dict)
+            else None
+        )
+        if isinstance(attachment_id, str) and _CANONICAL_UUID.fullmatch(attachment_id):
+            size = (
+                agenta_metadata.get("size")
+                if isinstance(agenta_metadata, dict)
+                else None
+            )
+            return [
+                ContentBlock(
+                    type="attachment",
+                    attachment_id=attachment_id,
+                    filename=part.get("filename"),
+                    mime_type=media,
+                    size=size,
+                )
+            ]
         kind = (
             "image"
             if isinstance(media, str) and media.startswith("image/")
@@ -302,8 +336,8 @@ def _block_to_parts(block: ContentBlock) -> List[Dict[str, Any]]:
     Keep this channel symmetric: a kind must be handled in both directions or neither. If
     something is mapped here outbound it must map inbound too (and vice versa); a kind
     dropped in ``_part_to_blocks`` must also be absent here — never add one side alone. The
-    only exception is a direction that explicitly cannot occur (e.g. the one-way live event
-    stream in ``stream.py``, which has no inbound counterpart by design).
+    module documents the ingress-only ``attachment`` case; its neutral block has no URL to
+    render here.
     """
     if block.type == "text":
         return [{"type": "text", "text": block.text or ""}]
@@ -334,6 +368,11 @@ def _block_to_parts(block: ContentBlock) -> List[Dict[str, Any]]:
                 "output": block.output,
             }
         ]
+    if block.type == "attachment":
+        log.debug(
+            "vercel adapter: dropping outbound attachment block with no FileUIPart URL: %r",
+            block.attachment_id,
+        )
     return []
 
 

sdks/python/agenta/sdk/agents/adapters/vercel/stream.py

@@ -9,6 +9,7 @@
 from agenta.sdk.utils.logging import get_module_logger
 
 from ...dtos import AgentResult
+from ...errors import AgentRunFailed
 from ...streaming import AgentStream
 from ...utils.wire import sanitize_runner_error
 from .messages import TOOL_APPROVAL_REQUEST
@@ -304,11 +305,17 @@ async def _agent_run_to_vercel_parts_impl(
                 if _conform(file_part) is not None:
                     content_parts_emitted += 1
                 yield file_part
+            elif etype == "attachment_delivery":
+                content_parts_emitted += 1
+                yield _attachment_delivery_part(data)
             elif etype == "usage":
                 usage = _usage_metadata(data)
             elif etype == "error":
                 error_emitted = True
-                yield {"type": "error", "errorText": data.get("message", "")}
+                for part in _error_parts(
+                    data.get("message", ""), failure_code="runner_error"
+                ):
+                    yield part
             elif etype == "done":
                 # Last non-null stop reason wins; see the routing-layer twin's `done` note.
                 reason = data.get("stopReason")
@@ -324,7 +331,8 @@ async def _agent_run_to_vercel_parts_impl(
             # exception is very often just that same failure resurfacing as a raised
             # `RuntimeError` (`result_from_wire`) -- yielding it too would duplicate the message
             # the user already saw under a second, "Agent run failed: ..."-prefixed frame.
-            yield {"type": "error", "errorText": sanitize_runner_error(exc)}
+            for part in _error_parts(sanitize_runner_error(exc), error=exc):
+                yield part
         error_emitted = True
     finally:
         # Every exit path — including the raw exception above — must still drain to a
@@ -349,7 +357,10 @@ async def _agent_run_to_vercel_parts_impl(
             # "no output" frame on top of it would bury the actionable message (the swallowed-
             # provider-error path both streams a live error event AND fails the terminal result,
             # so this backstop must not double up on it).
-            yield {"type": "error", "errorText": "The agent produced no output."}
+            for part in _error_parts(
+                "The agent produced no output.", failure_code="no_output"
+            ):
+                yield part
         finish: Dict[str, Any] = {"type": "finish"}
         finish_reason = _map_finish_reason(stop_reason)
         if finish_reason is not None:
@@ -584,11 +595,17 @@ async def _agent_stream_to_vercel_stream_impl(
                 if _conform(file_part) is not None:
                     content_parts_emitted += 1
                 yield file_part
+            elif etype == "attachment_delivery":
+                content_parts_emitted += 1
+                yield _attachment_delivery_part(data)
             elif etype == "usage":
                 usage = _usage_metadata(data)
             elif etype == "error":
                 error_emitted = True
-                yield {"type": "error", "errorText": data.get("message", "")}
+                for part in _error_parts(
+                    data.get("message", ""), failure_code="runner_error"
+                ):
+                    yield part
             elif etype == "done":
                 # Prefer the LAST non-null stop reason. The handler appends a corrective
                 # terminal `done` after the runner's `done` when the authoritative result
@@ -605,7 +622,8 @@ async def _agent_stream_to_vercel_stream_impl(
             # out live this turn, so a swallowed-provider-error recovery (live error event, then
             # a failed terminal result raised as this same exception) doesn't duplicate the
             # user-facing message under a second, differently-worded frame.
-            yield {"type": "error", "errorText": sanitize_runner_error(exc)}
+            for part in _error_parts(sanitize_runner_error(exc), error=exc):
+                yield part
         error_emitted = True
     finally:
         # Every exit path — including the raw exception above — must still drain to a
@@ -617,7 +635,10 @@ async def _agent_stream_to_vercel_stream_impl(
             # note) -- a swallowed-provider-error turn both streams a live error event and fails
             # the terminal result, so this backstop must not double up on it and bury the real
             # message under "The agent produced no output."
-            yield {"type": "error", "errorText": "The agent produced no output."}
+            for part in _error_parts(
+                "The agent produced no output.", failure_code="no_output"
+            ):
+                yield part
         finish: Dict[str, Any] = {"type": "finish"}
         finish_reason = _map_finish_reason(stop_reason)
         if finish_reason is not None:
@@ -850,6 +871,32 @@ def _as_text(value: Any) -> str:
     return value if isinstance(value, str) else str(value)
 
 
+def _attachment_delivery_part(data: Dict[str, Any]) -> Dict[str, Any]:
+    delivery = {
+        key: data[key]
+        for key in ("attachmentId", "outcome", "reasonCode", "workingPath")
+        if data.get(key) is not None
+    }
+    return {"type": "data-attachment-delivery", "data": delivery}
+
+
+def _error_parts(
+    error_text: Any,
+    *,
+    failure_code: Optional[str] = None,
+    error: Optional[BaseException] = None,
+) -> Iterator[Dict[str, Any]]:
+    resolved_code = failure_code or getattr(error, "failure_code", None)
+    if not isinstance(resolved_code, str) or not resolved_code:
+        resolved_code = AgentRunFailed.failure_code
+    resolved_text = _as_text(error_text)
+    yield {
+        "type": "data-agent-error",
+        "data": {"code": resolved_code, "errorText": resolved_text},
+    }
+    yield {"type": "error", "errorText": resolved_text}
+
+
 def _safe_result(run: AgentStream) -> Optional[AgentResult]:
     try:
         return run.result()

sdks/python/agenta/sdk/agents/connections/models.py

@@ -16,7 +16,7 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional
 from uuid import UUID
 
 from pydantic import BaseModel, Field, field_serializer, model_validator
@@ -181,6 +181,7 @@ class ResolvedConnection(BaseModel):
         default_factory=dict, repr=False
     )  # the ONLY secret channel
     endpoint: Optional[Endpoint] = None  # NON-secret connection config only
+    input_modalities: Optional[List[str]] = None
 
     @field_serializer("env", when_used="always")
     def _mask_env(self, env: Dict[str, str]) -> Dict[str, str]:
@@ -204,6 +205,8 @@ def to_wire(self) -> Dict[str, Any]:
             endpoint_wire = self.endpoint.to_wire()
             if endpoint_wire:
                 wire["endpoint"] = endpoint_wire
+        if self.input_modalities is not None:
+            wire["modelCapabilities"] = {"inputModalities": list(self.input_modalities)}
         return wire
 
 

sdks/python/agenta/sdk/agents/connections/resolver.py

@@ -18,6 +18,7 @@
 from typing import Any, Dict, Optional
 
 from ..capabilities import PROVIDER_ENV_VARS
+from ..model_catalog import model_input_modalities
 from .errors import UnsupportedProviderError
 from .models import (
     Endpoint,
@@ -30,6 +31,13 @@
 _PROVIDER_ENV_VARS: Dict[str, str] = PROVIDER_ENV_VARS
 
 
+def _input_modalities(
+    context: RuntimeAuthContext, *, provider: str, model: str
+) -> Optional[list[str]]:
+    # A miss means workspace-only downstream; do not guess.
+    return model_input_modalities(context.harness, model, provider=provider or None)
+
+
 class EnvConnectionResolver:
     """Read the requested provider's api key from the current process environment.
 
@@ -55,11 +63,15 @@ async def resolve(
         context: RuntimeAuthContext,
     ) -> ResolvedConnection:
         if model.connection.mode == "self_managed":
+            provider = model.provider or ""
             return ResolvedConnection(
-                provider=model.provider or "",
+                provider=provider,
                 model=model.model,
                 credential_mode="runtime_provided",
                 env={},
+                input_modalities=_input_modalities(
+                    context, provider=provider, model=model.model
+                ),
             )
 
         provider = model.provider
@@ -77,13 +89,19 @@ async def resolve(
                 model=model.model,
                 credential_mode="env",
                 env={env_var: key},
+                input_modalities=_input_modalities(
+                    context, provider=provider, model=model.model
+                ),
             )
         # Absence is valid: inject nothing and let the harness use its own login/OAuth.
         return ResolvedConnection(
             provider=provider,
             model=model.model,
             credential_mode="runtime_provided",
             env={},
+            input_modalities=_input_modalities(
+                context, provider=provider, model=model.model
+            ),
         )
 
 
@@ -141,4 +159,7 @@ async def resolve(
             credential_mode="env" if env else "runtime_provided",
             env=env,
             endpoint=endpoint,
+            input_modalities=_input_modalities(
+                context, provider=provider, model=model.model
+            ),
         )