fix(llm): handle subscription Responses streaming

[enyst]

Follow-up to #1682.

• Subscription mode requires for Responses; LiteLLM returns a streaming iterator.
• Drain the iterator and use the completed response so Agent/Conversation can proceed.
• Emit a one-time info log when subscription mode is active (helps confirm runtime path).

[all-hands-bot]

This PR successfully implements subscription mode streaming support for the Responses API. However, there are several issues that should be addressed, particularly around error handling, type validation, and exception usage. See inline comments for details.

[all-hands-bot]

🔴 Critical: Using assert for runtime type validation is problematic. Assertions can be disabled with Python's -O (optimize) flag, which would skip this check in production and could lead to silent failures.

Replace with an explicit check and proper exception:

Suggested change

	assert isinstance(completed_resp, ResponsesAPIResponse), (
	f"Expected ResponsesAPIResponse, got {type(completed_resp)}"
	)
	if not isinstance(completed_resp, ResponsesAPIResponse):
	raise TypeError(
	f"Expected ResponsesAPIResponse, got {type(completed_resp)}"
	)

[all-hands-bot]

🟠 Important: Raising AssertionError for validation failures is unconventional. AssertionError should typically be reserved for programmer errors and invariants, not runtime type validation. Consider using TypeError or a custom exception like LLMResponseError instead.

Suggested change

	if not isinstance(ret, SyncResponsesAPIStreamingIterator):
	raise AssertionError(
	f"Expected Responses stream iterator, got {type(ret)}"
	)
	if not isinstance(ret, SyncResponsesAPIStreamingIterator):
	raise TypeError(
	f"Expected Responses stream iterator, got {type(ret)}"
	)

[all-hands-bot]

🟠 Important: The stream draining loop has no timeout or error handling. If the stream:

• Hangs or yields indefinitely
• Raises an exception during iteration
• Takes an unusually long time

This could cause the entire request to hang or fail unexpectedly. Consider adding a timeout mechanism or wrapping in a try-except to provide better error messages if streaming fails.

[all-hands-bot]

🟠 Important: Same issue as above - using AssertionError for runtime validation. This should be a TypeError or similar exception.

Suggested change

	raise AssertionError(
	f"Expected ResponsesAPIResponse, got {type(ret)}"
	)
	raise TypeError(
	f"Expected ResponsesAPIResponse, got {type(ret)}"
	)

[all-hands-bot]

🟡 Suggestion: This import is performed inside the execution path. If the import fails (e.g., due to litellm version mismatch or API changes), the error message might not be immediately clear to developers.

Consider either:

1. Importing at module level if this is a stable API
2. Wrapping in try-except with a more descriptive error:

try:
    from litellm.responses.streaming_iterator import (
        SyncResponsesAPIStreamingIterator,
    )
except ImportError as e:
    raise ImportError(
        f"Failed to import SyncResponsesAPIStreamingIterator. "
        f"This may indicate an incompatible litellm version: {e}"
    ) from e

[all-hands-bot]

🟢 Nit: Consider adding parentheses for clarity, since the or operation happens before the bool() conversion:

Suggested change

	streaming_requested = bool(call_kwargs.get("stream", False) or self.stream)
	streaming_requested = bool(call_kwargs.get("stream", False)) or self.stream

Actually, self.stream is already a bool, so this could be simplified to:

streaming_requested = call_kwargs.get("stream", False) or self.stream