[Refactor] Extract Harmony streaming SSE event builders into streaming_events.py

[sfeng33]

Purpose

serving.py is ~2800 lines. The _emit_* methods are pure functions (state + data → events) that don't use instance state, yet they live on the class. Extracting them:

1. Reduces serving.py by ~800 lines
2. Makes event builders independently testable
3. Prepares for StreamingParsableContext support — the upcoming streaming parsable implementation will reuse these same event builders instead of duplicating them (as _process_simple_streaming_events currently does)

Future plan

This is the first step toward unifying SSE event emission across all three streaming paths (Harmony, Simple, Parsable):

1. This PR: Extract event builders into shared module
2. Follow-up: Further refactor SSE events so that StreamingParsableContext and HarmonyContext share the same event processor loop

Test Plan

pytest tests/entrypoints/openai/responses/test_harmony.py

[gemini-code-assist]

Code Review

This pull request successfully refactors the Harmony streaming SSE event builders by extracting them from serving.py into a dedicated streaming_events.py module. This significantly improves the maintainability and testability of the code, and sets a solid foundation for unifying event emission across different streaming contexts. The refactor correctly handles the transition of state and dependencies (like tool_server). I have identified a robustness issue in the browser tool event builder that could lead to runtime crashes on unexpected model output.

[gemini-code-assist]

The emit_browser_tool_events function lacks robustness against unexpected model output. Specifically:

1. json.loads will raise a JSONDecodeError if the model produces malformed JSON.
2. Accessing parsed_args["query"] and parsed_args["pattern"] will raise a KeyError if those keys are missing.
3. The ValueError at line 658 will propagate and cause a 500 error for the streaming request.

Since this is an async generator context, these unhandled exceptions will crash the stream for the client. Consider implementing robust parsing with error handling similar to _parse_browser_tool_call in harmony_utils.py.

    try:
        parsed_args = json.loads(previous_item.content[0].text)
    except (json.JSONDecodeError, IndexError):
        return []

    if function_name == "search":
        query = parsed_args.get("query")
        if not query: return []
        action = response_function_web_search.ActionSearch(
            type="search",
            query=query,
        )
    elif function_name == "open":
        action = response_function_web_search.ActionOpenPage(
            type="open_page",
            # TODO: translate to url
            url=f"cursor:{parsed_args.get('cursor', '')}",
        )
    elif function_name == "find":
        pattern = parsed_args.get("pattern")
        if not pattern: return []
        action = response_function_web_search.ActionFind(
            type="find",
            pattern=pattern,
            # TODO: translate to url
            url=f"cursor:{parsed_args.get('cursor', '')}",
        )
    else:
        return []

[sfeng33]

This piece of code is copied over, I prefer to keep logic unchanged in this PR since it's for refactoring, but let me know if folks think otherwise.

[sfeng33]

PTAL: @qandrew @daniel-salib

[qandrew]

lgtm, thanks! cc @houseroad can we add "ready" / automerge?

[mgoin]

LGTM just two nits. They can wait if the goal is to make this purely a movement PR

[sfeng33]

Verified the entrypoints-integration-responses-api tests passed locally (test_mcp_tools, test_parsable_context, test_parsable_context). The other failed tests are unrelated.