Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 39 additions & 4 deletions sdks/python/agenta/sdk/agents/adapters/vercel/stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,14 @@ async def _agent_run_to_vercel_parts_impl(
usage: Optional[Dict[str, Any]] = None
stop_reason: Optional[str] = None
content_parts_emitted = 0
# Whether a real `error` frame already went out this turn (a live-streamed provider/runner
# error, or one raised out of the run and caught below). The zero-content-parts backstop
# below must not pile a generic "produced no output" frame on top of a real error the user
# already saw -- that would bury the actionable message under a useless one (the runner's
# swallowed-provider-error recovery streams the real error live AND fails the terminal
# result, so both an `etype == "error"` event and the `except` branch can fire for the SAME
# underlying failure; either one must suppress the backstop).
error_emitted = False
# Tool-call ids already surfaced as a tool part. An approval request attaches
# to its tool part by id, so we synthesize one only when none preceded it.
seen_tool_calls: set = set()
Expand Down Expand Up @@ -285,6 +293,7 @@ async def _agent_run_to_vercel_parts_impl(
elif etype == "usage":
usage = _usage_metadata(data)
elif etype == "error":
error_emitted = True
yield {"type": "error", "errorText": data.get("message", "")}
elif etype == "done":
# Last non-null stop reason wins; see the routing-layer twin's `done` note.
Expand All @@ -294,7 +303,15 @@ async def _agent_run_to_vercel_parts_impl(
except Exception as exc:
# Sanitize — an unexpected exception's raw str() can carry a stack/path dump.
log.error("agent_run_to_vercel_parts: error mid-stream", exc_info=True)
yield {"type": "error", "errorText": sanitize_runner_error(exc)}
if not error_emitted:
# Only surface this as a NEW user-facing frame when no error already went out this
# turn. A swallowed-provider-error recovery streams the real error live (the
# `etype == "error"` branch above) and THEN fails the terminal result, so this
# 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)}
error_emitted = True
finally:
# Every exit path — including the raw exception above — must still drain to a
# finish frame, or a consumer waiting on it hangs. Mirrors the routing-layer twin.
Expand All @@ -312,8 +329,12 @@ async def _agent_run_to_vercel_parts_impl(
trace_id = result.trace_id

yield {"type": "finish-step"}
if content_parts_emitted == 0:
if content_parts_emitted == 0 and not error_emitted:
# An ok:true run with zero content parts would otherwise render as a blank bubble.
# Skip this when a real error already went out above -- appending a second, useless
# "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."}
finish: Dict[str, Any] = {"type": "finish"}
finish_reason = _map_finish_reason(stop_reason)
Expand Down Expand Up @@ -380,6 +401,9 @@ async def _agent_stream_to_vercel_stream_impl(
usage: Optional[Dict[str, Any]] = None
stop_reason: Optional[str] = None
content_parts_emitted = 0
# See the dev-twin's `error_emitted` note above: suppresses the zero-content backstop when
# a real error already went out this turn, live or raised.
error_emitted = False
seen_tool_calls: set = set()
tool_names_by_id: Dict[Any, Any] = {}

Expand Down Expand Up @@ -535,6 +559,7 @@ async def _agent_stream_to_vercel_stream_impl(
elif etype == "usage":
usage = _usage_metadata(data)
elif etype == "error":
error_emitted = True
yield {"type": "error", "errorText": data.get("message", "")}
elif etype == "done":
# Prefer the LAST non-null stop reason. The handler appends a corrective
Expand All @@ -547,13 +572,23 @@ async def _agent_stream_to_vercel_stream_impl(
stop_reason = reason
except Exception as exc:
log.error("agent_stream_to_vercel_stream: error mid-stream", exc_info=True)
yield {"type": "error", "errorText": sanitize_runner_error(exc)}
if not error_emitted:
# See the dev-twin's matching note: suppress this when a real error already went
# 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)}
error_emitted = True
finally:
# Every exit path — including the raw exception above — must still drain to a
# finish frame, or a consumer waiting on it hangs.
yield {"type": "finish-step"}
if content_parts_emitted == 0:
if content_parts_emitted == 0 and not error_emitted:
# An ok:true run with zero content parts would otherwise render as a blank bubble.
# Skip this when a real error already went out above (see the dev-twin's matching
# 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."}
finish: Dict[str, Any] = {"type": "finish"}
finish_reason = _map_finish_reason(stop_reason)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,88 @@ async def test_dropped_only_content_part_still_triggers_zero_content_guard_dev_t
)


@pytest.mark.asyncio
async def test_swallowed_provider_error_emits_exactly_one_error_frame() -> None:
"""F-5317-followup: a turn that recovers a swallowed provider error (out-of-credit, bad
key, ...) streams a real error live AND fails its terminal result -- the runner's
`findSwallowedPiError` path both `emitEvent({type:"error"})`s the recovered message and
returns `{ok:false, error:...}` for the SAME failure (`sandbox_agent.ts` around the
`swallowedError` branch). On the wire that means the live event surfaces as one `error`
frame here, and the failed terminal record raises out of the event iterator and is caught
by this adapter's `except Exception` as a second `error` frame.

Before the fix, the zero-content-parts backstop then piled a THIRD, generic frame on top
("The agent produced no output.") because neither error frame incremented
`content_parts_emitted` -- burying the actionable message under a useless one. QA observed
exactly this on a real out-of-credit run: frames were
``[error(real), error("Agent run failed: " + real), error("The agent produced no
output.")]`` and the UI showed only the last frame. This test pins that the generic
backstop no longer fires once a real error went out.
"""
real_error = (
"pi_core: the model provider account has insufficient credit "
"(check the project's OpenAI key)."
)

async def _events_with_uncaught_failure():
yield {"type": "error", "data": {"message": real_error}}
# Mirrors the terminal `ok:false` result raising out of the event iterator uncaught
# (streaming.py's AgentStream.__aiter__ -> result_from_wire -> RuntimeError, propagated
# through handler.py's agent_event_stream with no enclosing except).
raise RuntimeError(f"Agent run failed: {real_error}")

parts = [
part
async for part in agent_stream_to_vercel_stream(
_events_with_uncaught_failure(), trace_id="t-swallowed"
)
]
for part in parts:
assert_conforms(part)

error_parts = [p for p in parts if p["type"] == "error"]
assert len(error_parts) == 1, (
f"expected exactly one error frame, got {error_parts!r}"
)
assert error_parts[0]["errorText"] == real_error
assert not any(p.get("errorText") == "The agent produced no output." for p in parts)


@pytest.mark.asyncio
async def test_swallowed_provider_error_emits_exactly_one_error_frame_dev_twin() -> (
None
):
"""Dev-twin counterpart: the live error event AND the terminal `ok:false` both come off the
same ``AgentStream`` (`kind:"event"` then `kind:"result"`), matching the real runner's NDJSON
record shape (`server.ts` `liveEmit` then the terminal `{kind:"result"}` write).
"""
real_error = (
"pi_core: the model provider account has insufficient credit "
"(check the project's OpenAI key)."
)
# `Event.from_wire` keeps the raw record verbatim as `.data` (it does not unwrap a nested
# `data` key), so a live error event's `message` rides at the TOP level -- mirrors the
# runner's actual `run.emitEvent({type:"error", message: swallowedError})` shape
# (`sandbox_agent.ts`). The terminal result's `error` is the concise message UNPREFIXED
# (`{ok:false, error: swallowedError}`, same file) -- `result_from_wire` adds the
# "Agent run failed: " prefix itself when it raises.
records = [
{"kind": "event", "event": {"type": "error", "message": real_error}},
{"kind": "result", "result": {"ok": False, "error": real_error}},
]
run = AgentStream(_records(records))
parts = [part async for part in agent_run_to_vercel_parts(run)]
for part in parts:
assert_conforms(part)

error_parts = [p for p in parts if p["type"] == "error"]
assert len(error_parts) == 1, (
f"expected exactly one error frame, got {error_parts!r}"
)
assert error_parts[0]["errorText"] == real_error
assert not any(p.get("errorText") == "The agent produced no output." for p in parts)


def test_vendored_version_matches_package_pin() -> None:
# CI-grep-able tripwire: bump this const (and re-audit the shape above) whenever
# web/oss/package.json's "ai" pin changes.
Expand Down
Loading