← pod_r3_v1e

fastapi_15588

failed WRONG_FIX wrong_fix · 27 tool calls · 280 s · fastapi/fastapi

Task input

♻️ Validate Server Sent Event fields to avoid applications from sending broken data

## Pull Request

<!--
Please start with a GitHub Discussion.

Once a team member asks you to open a PR, create it and link the discussion here.

Obvious typo fixes can be made in a PR without starting a discussion.
-->

Discussion: <!-- Link to the GitHub Discussion -->

## Description

♻️ Validate Server Sent Event fields to avoid applications from sending broken data

This was reported multiple times as a "vulnerability" by users probably using automated AI tools, with the argument that if input was taken from the request and passed directly to an event unfiltered, it could break the client.

If an app sets an `id` or `event` with multi-line content, I would consider that app broken, and probably the app itself vulnerable to something, not FastAPI. With this change, if any app sets invalid multi-line values (sourced from unfiltered input or in any other way), FastAPI will make it a server error (which it is, as the app is sending invalid data).

I don't think the code before this PR has a bug, less a vulnerability. This change is mainly a quality of life improvement to prevent badly written apps from shooting themselves in the foot by passing unsanitized data directly to a SSE.

<!-- Write the description of your PR here -->

## AI Disclaimer

<!-- If using AI, write here the prompt and model used -->

Codex with GPT-5.5

<details>
<summary>AI transcript</summary>

<!-- Paste here the entire AI transcript -->

</details>

## Checklist

- [ ] This PR is an obvious typo fix, or it links to a GitHub Discussion for the proposed code change.
- [x] I added tests for the change.
- [x] The new or updated tests fail on the main branch and pass on this PR.
- [x] Coverage stays at 100%.
- [x] The documentation explains the change if needed.

Tool calls (28)

#ToolArgumentsResult
1run_command
{
 "command": "grep -rn \"EventSourceResponse\" . | head -20"
}
{"status": "ok", "stdout": "./tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n./tests/test_sse.py:29:@app.get(\"/items/stream\", response
full
{"status": "ok", "stdout": "./tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n./tests/test_sse.py:29:@app.get(\"/items/stream\", response_class=EventSourceResponse)\n./tests/test_sse.py:35:@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\n./tests/test_sse.py:40:@app.get(\"/items/stream-no-annotation\", response_class=EventSourceResponse)\n./tests/test_sse.py:46:@app.get(\"/items/stream-sync-no-annotation\", response_class=EventSourceResponse)\n./tests/test_sse.py:51:@app.get(\"/items/stream-dict\", response_class=EventSourceResponse)\n./tests/test_sse.py:57:@app.get(\"/items/stream-sse-event\", response_class=EventSourceResponse)\n./tests/test_sse.py:65:@app.get(\"/items/stream-mixed\", response_class=EventSourceResponse)\n./tests/test_sse.py:72:@app.get(\"/items/stream-string\", response_class=EventSourceResponse)\n./tests/test_sse.py:77:@app.post(\"/items/stream-post\", response_class=EventSourceResponse)\n./tests/test_sse.py:83:@app.get(\"/items/stream-raw\", response_class=EventSourceResponse)\n./tests/test_sse.py:93:@router.get(\"/events\", response_class=EventSourceResponse)\n./tests/test_sse.py:274:@keepalive_app.get(\"/slow-async\", response_class=EventSourceResponse)\n./tests/test_sse.py:283:@keepalive_app.get(\"/slow-sync\", response_class=EventSourceResponse)\n./docs/zh-hant/docs/tutorial/server-sent-events.md:38:\u8981\u5728 FastAPI \u4e2d\u4e32\u6d41 SSE\uff0c\u8acb\u5728\u4f60\u7684\u8def\u5f91\u64cd\u4f5c\u51fd\u5f0f\uff08path operation function\uff09\u4e2d\u4f7f\u7528 `yield`\uff0c\u4e26\u8a2d\u5b9a `response_class=EventSourceResponse`\u3002\n./docs/zh-hant/docs/tutorial/server-sent-events.md:40:\u5f9e `fastapi.sse` \u532f\u5165 `EventSourceResponse`\uff1a\n./docs/ja/docs/tutorial/server-sent-events.md:38:FastAPI \u3067 SSE \u3092\u30b9\u30c8\u30ea\u30fc\u30df\u30f3\u30b0\u3059\u308b\u306b\u306f\u3001*path operation \u95a2\u6570*\u3067 `yield` \u3092\u4f7f\u3044\u3001`response_class=EventSourceResponse` \u30
2run_command
{
 "command": "grep -rn \"EventSourceResponse\" fastapi | head -20"
}
{"status": "ok", "stdout": "fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse # noqa\nfastapi/.agents/skills/fastapi/ref
full
{"status": "ok", "stdout": "fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse  # noqa\nfastapi/.agents/skills/fastapi/references/streaming.md:16:To stream Server-Sent Events, use `response_class=EventSourceResponse` and `yield` items from the endpoint.\nfastapi/.agents/skills/fastapi/references/streaming.md:24:from fastapi.sse import EventSourceResponse\nfastapi/.agents/skills/fastapi/references/streaming.md:35:@app.get(\"/items/stream\", response_class=EventSourceResponse)\nfastapi/.agents/skills/fastapi/references/streaming.md:47:from fastapi.sse import EventSourceResponse, ServerSentEvent\nfastapi/.agents/skills/fastapi/references/streaming.md:52:@app.get(\"/events\", response_class=EventSourceResponse)\nfastapi/.agents/skills/fastapi/SKILL.md:333:See [the streaming reference](references/streaming.md) for JSON Lines, Server-Sent Events (`EventSourceResponse`, `ServerSentEvent`), and byte streaming (`StreamingResponse`) patterns.\nfastapi/sse.py:20:class EventSourceResponse(StreamingResponse):\nfastapi/sse.py:23:    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\nfastapi/sse.py:46:    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\nfastapi/routing.py:64:    EventSourceResponse,\nfastapi/routing.py:376:    is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse)\nfastapi/routing.py:856:                    # EventSourceResponse (SSE).\nfastapi/routing.py:862:                        or lenient_issubclass(response_class, EventSourceResponse)\nfastapi/routing.py:971:            response_class, EventSourceResponse\n", "stderr": "", "exit_code": 0}
3run_command
{
 "command": "cat fastapi/sse.py"
}
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_val
full
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string
4run_command
{
 "command": "sed -n '1,120p' fastapi/sse.py"
}
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_val
full
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string
5run_command
{
 "command": "sed -n '121,160p' fastapi/sse.py"
}
{"status": "ok", "stdout": " ] = None\n comment: Annotated[\n str | None,\n Doc(\n \"\"\"\n Optional comment line(s).\n\n Comment lines start with `:` in the SS
full
{"status": "ok", "stdout": "    ] = None\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s).\n\n            Comment lines start with `:` in the SSE wire format and are ignored by\n            `EventSource` clients. Useful for keep-alive pings to prevent\n            proxy/load-balancer timeouts.\n            \"\"\"\n        ),\n    ] = None\n\n    @model_validator(mode=\"after\")\n    def _check_data_exclusive(self) -> \"ServerSentEvent\":\n        if self.data is not None and self.raw_data is not None:\n            raise ValueError(\n                \"Cannot set both 'data' and 'raw_data' on the same \"\n                \"ServerSentEvent. Use 'data' for JSON-serialized payloads \"\n                \"or 'raw_data' for pre-formatted strings.\"\n            )\n        return self\n\n\ndef format_sse_event(\n    *,\n    data_str: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n", "stderr": "", "exit_code": 0}
6run_command
{
 "command": "sed -n '161,200p' fastapi/sse.py"
}
{"status": "ok", "stdout": " \"\"\"\n ),\n ] = None,\n id: Annotated[\n str | None,\n Doc(\n \"\"\"\n Optional event ID (`id:` field).\n \"\"\"\n ),\n ] = None,
full
{"status": "ok", "stdout": "            \"\"\"\n        ),\n    ] = None,\n    id: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event ID (`id:` field).\n            \"\"\"\n        ),\n    ] = None,\n    retry: Annotated[\n        int | None,\n        Doc(\n            \"\"\"\n            Optional reconnection time in milliseconds (`retry:` field).\n            \"\"\"\n        ),\n    ] = None,\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s) (`:` prefix).\n            \"\"\"\n        ),\n    ] = None,\n) -> bytes:\n    \"\"\"Build SSE wire-format bytes from **pre-serialized** data.\n\n    The result always ends with `\\n\\n` (the event terminator).\n    \"\"\"\n    lines: list[str] = []\n\n    if comment is not None:\n        for line in comment.splitlines():\n            lines.append(f\": {line}\")\n\n    if event is not None:\n        lines.append(f\"event: {event}\")\n", "stderr": "", "exit_code": 0}
7run_command
{
 "command": "sed -n '201,220p' fastapi/sse.py"
}
{"status": "ok", "stdout": "\n if data_str is not None:\n for line in data_str.splitlines():\n lines.append(f\"data: {line}\")\n\n if id is not None:\n lines.ap
full
{"status": "ok", "stdout": "\n    if data_str is not None:\n        for line in data_str.splitlines():\n            lines.append(f\"data: {line}\")\n\n    if id is not None:\n        lines.append(f\"id: {id}\")\n\n    if retry is not None:\n        lines.append(f\"retry: {retry}\")\n\n    lines.append(\"\")\n    lines.append(\"\")\n    return \"\\n\".join(lines).encode(\"utf-8\")\n\n\n# Keep-alive comment, per the SSE spec recommendation\nKEEPALIVE_COMMENT = b\": ping\\n\\n\"\n\n# Seconds between keep-alive pings when a generator is idle.\n", "stderr": "", "exit_code": 0}
8run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q -x"
}
{"status": "ok", "stdout": ".................. [100%]\n=============================== warnings summary ===============================\n../../../../../../../Us
full
{"status": "ok", "stdout": "..................                                                       [100%]\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n18 passed, 1 warning in 1.11s\n", "stderr": "", "exit_code": 0}
9run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi import FastAPI\nfrom fastapi.sse import EventSourceResponse, ServerSentEvent\nfrom fastapi.testclient import TestClient\nimport pytest\n\napp = FastAPI()\n\n@app.get(\"/sse\")\nasync def sse_endpoint():\n    yield ServerSentEvent(event=\"my-event\\nbreak\", data=\"some data\")\n\nclient = TestClient(app)\n\ndef test_sse_multiline_event():\n    with pytest.raises(ValueError) as excinfo:\n        # We need to actually iterate the response to trigger the generator\n        response = client.get(\"/sse\")\n        # The generator is executed during response iteration\n        # However, EventSourceResponse is a StreamingResponse, so we need to read it\n        _ = response.content \n    assert \"SSE 'event' must not contain newline characters\" in str(excinfo.value)\n\ndef test_sse_multiline_id():\n    @app.get(\"/sse-id\")\n    async def sse_id_endpoint():\n        yield ServerSentEvent(id=\"my-id\\nbreak\", data=\"some data\")\n    \n    with pytest.raises(ValueError) as excinfo:\n        response = client.get(\"/sse-id\")\n        _ = response.content\n    assert \"SSE 'id' must not contain newline characters\" in str(excinfo.value)\n\nif __name__ == \"__main__\":\n    # This is a simple script to test, not a full pytest suite\n    try:\n        # Test multiline event\n        print(\"Testing multiline event...\")\n        # Since we are not in a pytest environment here, we manually call the logic\n        # that would be triggered.\n        # In FastAPI, the generator is consumed by the server.\n        # Let's just test the ServerSentEvent model validation if we add it there.\n        pass\n    except Exception as e:\n        print(e)\nPY\n"
}
{"status": "ok", "stdout": "Testing multiline event...\n", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "Testing multiline event...\n", "stderr": "", "exit_code": 0}
10run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    print(\"Testing multiline event...\")\n    ServerSentEvent(event=\"line1\\nline2\", data=\"data\")\n    print(\"Failed: multiline event did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\n\ntry:\n    print(\"\\nTesting multiline id...\")\n    ServerSentEvent(id=\"id1\\nid2\", data=\"data\")\n    print(\"Failed: multiline id did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\nPY\n"
}
{"status": "ok", "stdout": "Testing multiline event...\nFailed: multiline event did not raise ValueError\n\nTesting multiline id...\nFailed: multiline id did no
full
{"status": "ok", "stdout": "Testing multiline event...\nFailed: multiline event did not raise ValueError\n\nTesting multiline id...\nFailed: multiline id did not raise ValueError\n", "stderr": "", "exit_code": 0}
11run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# Define the validator for newlines\nnewline_validator_code = \"\"\"\ndef _check_no_newlines(v: str | None, field_name: str) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(f\"SSE '{field_name}' must not contain newline characters\")\n    return v\n\"\"\"\n\n# We need to wrap it to use it in AfterValidator, which takes a callable.\n# Since AfterValidator takes a function, we can't easily pass field_name.\n# We can use a closure or separate functions.\n\n# Let's use separate functions for simplicity or a helper.\n# Actually, let's just define two functions.\n\ns = s.replace(\n    'def _check_id_no_null(v: str | None) -> str | None:',\n    'def _check_id_no_null(v: str | None) -> str | None:\\n    if v is not None and \"\\\\n\" in v:\\n        raise ValueError(\"SSE \\\\'id\\\\' must not contain newline characters\")\\n    if v is not None and \"\\\\0\" in v:\\n        raise ValueError(\"SSE \\\\'id\\\\' must not contain null characters\")\\n    return v'\n)\n\n# Now for the event field.\n# Find the event field definition and add AfterValidator.\n# The event field is:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n#             \"\"\"\n#             Optional event type name.\n#\n#             Maps to `addEventListener(event, ...)` on the browser. When omitted,\n#             the browser dispatches on the generic `message` event.\n#             \"\"\"\n#         ),\n#     ] = None\n\n# Let's define a new validator function for event.\n# I'll insert it before ServerSentEvent class.\n\n# First, let's add the _check_event_no_newlines function.\n# I will insert it after _check_id_no_null.\n\n# Let's do this more carefully.\nPY\n"
}
{"status": "error", "error_type": "CommandError", "error_message": " File \"<stdin>\", line 22\n 'def _check_id_no_null(v: str | None) -> str | None:\\n if v is
full
{"status": "error", "error_type": "CommandError", "error_message": "  File \"<stdin>\", line 22\n    'def _check_id_no_null(v: str | None) -> str | None:\\n    if v is not None and \"\\\\n\" in v:\\n        raise ValueError(\"SSE \\\\'id\\\\' must not contain newline characters\")\\n    if v is not None and \"\\\\0\" in v:\\n        raise ValueError(\"SSE \\\\'id\\\\' must not contain null characters\")\\n    return v'\n                                                                                                                                    ^\nSyntaxError: unexpected character after line continuation character\n", "details": {"stdout": "", "stderr": "  File \"<stdin>\", line 22\n    'def _check_id_no_null(v: str | None) -> str | None:\\n    if v is not None and \"\\\\n\" in v:\\n        raise ValueError(\"SSE \\\\'id\\\\' must not contain newline characters\")\\n    if v is not None and \"\\\\0\" in v:\\n        raise ValueError(\"SSE \\\\'id\\\\' must not contain null characters\")\\n    return v'\n                                                                                                                                    ^\nSyntaxError: unexpected character after line continuation character\n", "exit_code": 1}}
12run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# 1. Update _check_id_no_null to also check for newlines\nold_id_check = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\"\"\"\n\nnew_id_check = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\"\"\"\n\ns = s.replace(old_id_check, new_id_check)\n\n# 2. Add _check_event_no_newlines function\n# We'll insert it before the ServerSentEvent class.\n# Find the line where class ServerSentEvent(BaseModel): starts.\n# We can use a marker.\n\n# Let's find the line number of class ServerSentEvent(BaseModel):\nlines = s.splitlines()\nfor i, line in enumerate(lines):\n    if line.strip() == \"class ServerSentEvent(BaseModel):\":\n        insert_pos = i\n        break\n\nevent_check_code = \"\"\"\ndef _check_event_no_newlines(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'event' must not contain newline characters\")\n    return v\n\"\"\"\n# Insert it before the class definition.\nlines.insert(insert_pos, \"\")\nlines.insert(insert_pos, event_check_code.strip())\n\n# Now update the event field to use AfterValidator(_check_event_no_newlines)\n# The event field is:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n#             \"\"\"\n#             Optional event type name.\n#\n#             Maps to `addEventListener(event, ...)` on the browser. When omitted,\n#             the browser dispatches on the generic `message` event.\n#             \"\"\"\n#         ),\n#     ] = None\n\n# We need to find the exact block and replace it.\n# Since it's a multi-line string, let's use a more robust way.\n\n# Let's find the start and end of the event field definition.\n# We'll look for 'event: Annotated[' and the closing ' ] = None'\n# but we must be careful not to match other fields.\n\n# Let's just use a simple replace for the specific part.\n# The current code is:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n#             \"\"\"\n#             Optional event type name.\n#\n#             Maps to `addEventListener(event, ...)` on the browser. When omitted,\n#             the browser dispatches on the generic `message` event.\n#             \"\"\"\n#         ),\n#     ] = None\n\n# We want to change it to:\n#     event: Annotated[\n#         str | None,\n#         AfterValidator(_check_event_no_newlines),\n#         Doc(\n#             \"\"\"\n#             Optional event type name.\n#\n#             Maps to `addEventListener(event, ...)` on the browser. When omitted,\n#             the browser dispatches on the generic `message` event.\n#             \"\"\"\n#         ),\n#     ] = None\n\n# Let's use a string replacement on the whole content.\n# We'll use the exact text from the file.\n\nold_event_field = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\n            \\\"\\\"\\\"\n            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n            the browser dispatches on the generic `message` event.\n            \\\"\\\"\\\"\n        ),\n    ] = None\"\"\"\n\nnew_event_field = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_event_no_newlines),\n        Doc(\n            \\\"\\\"\\\"\n            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n            the browser dispatches on the generic `message` event.\n            \\\"\\\"\\\"\n        ),\n    ] = None\"\"\"\n\n# Note: the original fi
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
13run_command
{
 "command": "sed -n '78,90p' fastapi/sse.py"
}
{"status": "ok", "stdout": "\n Use this when you need to send pre-formatted text, HTML fragments,\n CSV lines, or any non-JSON payload. The string is placed dir
full
{"status": "ok", "stdout": "\n            Use this when you need to send pre-formatted text, HTML fragments,\n            CSV lines, or any non-JSON payload. The string is placed directly\n            into the `data:` field as-is.\n\n            Mutually exclusive with `data`.\n            \"\"\"\n        ),\n    ] = None\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n", "stderr": "", "exit_code": 0}
14run_command
{
 "command": "sed -n '91,100p' fastapi/sse.py"
}
{"status": "ok", "stdout": " Optional event type name.\n\n Maps to `addEventListener(event, ...)` on the browser. When omitted,\n the browser dispatches on the
full
{"status": "ok", "stdout": "            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n            the browser dispatches on the generic `message` event.\n            \"\"\"\n        ),\n    ] = None\n    id: Annotated[\n        str | None,\n        AfterValidator(_check_id_no_null),\n", "stderr": "", "exit_code": 0}
15run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# 1. Update _check_id_no_null\nold_id_check = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\"\"\"\n\nnew_id_check = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\"\"\"\n\ns = s.replace(old_id_check, new_id_check)\n\n# 2. Add _check_event_no_newlines function\n# Insert it before class ServerSentEvent(BaseModel):\n# We'll use a marker to find the position.\nmarker = \"class ServerSentEvent(BaseModel):\"\npos = s.find(marker)\nevent_check_func = \"\"\"\ndef _check_event_no_newlines(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'event' must not contain newline characters\")\n    return v\n\n\"\"\"\ns = s[:pos] + event_check_func + s[pos:]\n\n# 3. Update the event field to use AfterValidator(_check_event_no_newlines)\n# We'll replace:\n#         str | None,\n#         Doc(\n# with:\n#         str | None,\n#         AfterValidator(_check_event_no_newlines),\n#         Doc(\n\n# We need to be careful to only replace it for the 'event' field.\n# The 'event' field is the only one that has 'str | None,' followed by 'Doc('\n# and is not 'id' (which already has AfterValidator).\n\n# Let's use a more specific replacement.\nold_event_part = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\"\"\"\"\nnew_event_part = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_event_no_newlines),\n        Doc(\"\"\"\"\n\n# Wait, the original code uses triple quotes for Doc.\n# Let's use a different approach.\n\n# I'll use a regex or just find the line and insert.\nlines = s.splitlines()\nfor i, line in enumerate(lines):\n    if line.strip() == \"    event: Annotated[\":\n        # Find the line '        str | None,'\n        for j in range(i + 1, len(lines)):\n            if lines[j].strip() == \"        str | None,\":\n                lines.insert(j + 1, \"        AfterValidator(_check_event_no_newlines),\")\n                break\n        break\n\ns = \"\\n\".join(lines)\np.write_text(s)\nPY\n"
}
{"status": "error", "error_type": "CommandError", "error_message": " File \"<stdin>\", line 50\n Doc(\"\"\"\"\n ^\nSyntaxError: unterminated string literal (det
full
{"status": "error", "error_type": "CommandError", "error_message": "  File \"<stdin>\", line 50\n    Doc(\"\"\"\"\n           ^\nSyntaxError: unterminated string literal (detected at line 50)\n", "details": {"stdout": "", "stderr": "  File \"<stdin>\", line 50\n    Doc(\"\"\"\"\n           ^\nSyntaxError: unterminated string literal (detected at line 50)\n", "exit_code": 1}}
16run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# 1. Update _check_id_no_null\nold_id_check = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\"\"\"\n\nnew_id_check = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\"\"\"\n\ns = s.replace(old_id_check, new_id_check)\n\n# 2. Add _check_event_no_newlines function\nmarker = \"class ServerSentEvent(BaseModel):\"\npos = s.find(marker)\nevent_check_func = \"\"\"\ndef _check_event_no_newlines(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'event' must not contain newline characters\")\n    return v\n\"\"\"\ns = s[:pos] + event_check_func + s[pos:]\n\n# 3. Update the event field to use AfterValidator(_check_event_no_newlines)\nlines = s.splitlines()\nfor i, line in enumerate(lines):\n    if line.strip() == \"    event: Annotated[\":\n        for j in range(i + 1, len(lines)):\n            if lines[j].strip() == \"        str | None,\":\n                lines.insert(j + 1, \"        AfterValidator(_check_event_no_newlines),\")\n                break\n        break\n\ns = \"\\n\".join(lines)\np.write_text(s)\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
17run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    print(\"Testing multiline event...\")\n    ServerSentEvent(event=\"line1\\nline2\", data=\"data\")\n    print(\"Failed: multiline event did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\n\ntry:\n    print(\"\\nTesting multiline id...\")\n    ServerSentEvent(id=\"id1\\nid2\", data=\"data\")\n    print(\"Failed: multiline id did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\n\ntry:\n    print(\"\\nTesting null id...\")\n    ServerSentEvent(id=\"id\\0id\", data=\"data\")\n    print(\"Failed: null id did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\nPY\n"
}
{"status": "ok", "stdout": "Testing multiline event...\nFailed: multiline event did not raise ValueError\n\nTesting multiline id...\nPassed: 1 validation error
full
{"status": "ok", "stdout": "Testing multiline event...\nFailed: multiline event did not raise ValueError\n\nTesting multiline id...\nPassed: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain newline characters [type=value_error, input_value='id1\\nid2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\n\nTesting null id...\nPassed: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain null characters [type=value_error, input_value='id\\x00id', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\n", "stderr": "", "exit_code": 0}
18run_command
{
 "command": "sed -n '30,110p' fastapi/sse.py"
}
{"status": "ok", "stdout": " serves mainly as a marker and sets the correct `Content-Type`.\n \"\"\"\n\n media_type = \"text/event-stream\"\n\n\ndef _check_id_n
full
{"status": "ok", "stdout": "    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\n\ndef _check_event_no_newlines(v: str | None) -> str | None:\n    if v is not None and \"\\n\" in v:\n        raise ValueError(\"SSE 'event' must not contain newline characters\")\n    return v\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain strings** are JSON-serialized.\n\n    For example, `data=\"hello\"` produces `data: \"hello\"` on the wire (with\n    quotes).\n    \"\"\"\n\n    data: Annotated[\n        Any,\n        Doc(\n            \"\"\"\n            The event payload.\n\n            Can be any JSON-serializable value: a Pydantic model, dict, list,\n            string, number, etc. It is **always** serialized to JSON: strings\n            are quoted (`\"hello\"` becomes `data: \"hello\"` on the wire).\n\n            Mutually exclusive with `raw_data`.\n            \"\"\"\n        ),\n    ] = None\n    raw_data: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Raw string to send as the `data:` field **without** JSON encoding.\n\n            Use this when you need to send pre-form
19run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# The previous attempt failed to insert AfterValidator for 'event'.\n# Let's look at the file content again.\n# The 'event' field is:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n#             \"\"\"\n#             Optional event type name.\n#\n#             Maps to `addEventListener(event, ...)` on the browser. When omitted,\n#             the browser dispatches on the generic `message` event.\n#             \"\"\"\n#         ),\n#     ] = None\n\n# Let's use a more robust way to insert AfterValidator.\nlines = s.splitlines()\nnew_lines = []\nin_event_annotated = False\nevent_str_none_found = False\n\nfor line in lines:\n    new_lines.append(line)\n    if line.strip() == \"    event: Annotated[\":\n        in_event_annotated = True\n    if in_event_annotated and line.strip() == \"        str | None,\":\n        event_str_none_found = True\n    if in_event_annotated and event_str_none_found and line.strip() == \"        Doc(\":\n        new_lines.insert(-1, \"        AfterValidator(_check_event_no_newlines),\")\n        in_event_annotated = False\n        event_str_none_found = False\n\n# Wait, the above logic is slightly wrong. If I insert, I should not use new_lines.append(line) first.\n# Let's try again.\n\nlines = s.splitlines()\nresult_lines = []\ni = 0\nwhile i < len(lines):\n    line = lines[i]\n    result_lines.append(line)\n    if line.strip() == \"    event: Annotated[\":\n        # Look ahead for '        str | None,' and then '        Doc('\n        j = i + 1\n        while j < len(lines) and lines[j].strip() != \"        Doc(\":\n            result_lines.append(lines[j])\n            j += 1\n        if j < len(lines):\n            # We found '        Doc('. We should insert AfterValidator before it.\n            # But we must ensure '        str | None,' was there.\n            # Actually, the structure is very predictable.\n            result_lines.insert(-1, \"        AfterValidator(_check_event_no_newlines),\")\n            # Now we are at '        Doc('. We should append it and continue.\n            # Wait, the loop above already appended lines up to j-1.\n            # Let's just use a simpler approach.\n    i += 1\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
20run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\nlines = s.splitlines()\nnew_lines = []\ni = 0\nwhile i < len(lines):\n    line = lines[i]\n    new_lines.append(line)\n    if line.strip() == \"    event: Annotated[\":\n        # We are in the event field.\n        # We expect:\n        #     event: Annotated[\n        #         str | None,\n        #         Doc(\n        # We want to insert AfterValidator between 'str | None,' and 'Doc('.\n        \n        # Let's just find the line '        str | None,' and insert after it.\n        # But only if we are within the event field.\n        \n        # Let's just use a simple replacement for the whole block.\n        pass\n    i += 1\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
21run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# Let's use a very simple replacement.\n# We know the exact lines for the event field.\n# I will use a multi-line string that matches exactly.\n\nold_event_block = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\n            \\\"\\\"\\\"\n            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n            the browser dispatches on the generic `message` event.\n            \\\"\\\"\\\"\n        ),\n    ] = None\"\"\"\n\nnew_event_block = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_event_no_newlines),\n        Doc(\n            \\\"\\\"\\\"\n            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n            the browser dispatches on the generic `message` event.\n            \\\"\\\"\\\"\n        ),\n    ] = None\"\"\"\n\n# The problem is the triple quotes. Let's use a different way to define the block.\n# I'll read the file and find the start and end indices.\n\nstart_marker = '    event: Annotated['\nend_marker = '    ] = None'\n\n# Find the start_marker\nstart_idx = s.find(start_marker)\n# Find the end_marker that comes after start_idx\nend_idx = s.find(end_marker, start_idx)\n\n# Now we have the block. We want to insert AfterValidator after 'str | None,'.\nblock = s[start_idx:end_idx + len(end_marker)]\n# The block looks like:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n#             ...\n#         ),\n#     ] = None\n\n# Replace '        str | None,' with '        str | None,\\n        AfterValidator(_check_event_no_newlines),'\nnew_block = block.replace('        str | None,', '        str | None,\\n        AfterValidator(_check_event_no_newlines),')\n\ns = s[:start_idx] + new_block + s[end_idx + len(end_marker):]\np.write_text(s)\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
22run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    print(\"Testing multiline event...\")\n    ServerSentEvent(event=\"line1\\nline2\", data=\"data\")\n    print(\"Failed: multiline event did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\n\ntry:\n    print(\"\\nTesting multiline id...\")\n    ServerSentEvent(id=\"id1\\nid2\", data=\"data\")\n    print(\"Failed: multiline id did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\n\ntry:\n    print(\"\\nTesting null id...\")\n    ServerSentEvent(id=\"id\\0id\", data=\"data\")\n    print(\"Failed: null id did not raise ValueError\")\nexcept (ValueError, ValidationError) as e:\n    print(f\"Passed: {e}\")\nPY\n"
}
{"status": "ok", "stdout": "Testing multiline event...\nPassed: 1 validation error for ServerSentEvent\nevent\n Value error, SSE 'event' must not contain newlin
full
{"status": "ok", "stdout": "Testing multiline event...\nPassed: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE 'event' must not contain newline characters [type=value_error, input_value='line1\\nline2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\n\nTesting multiline id...\nPassed: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain newline characters [type=value_error, input_value='id1\\nid2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\n\nTesting null id...\nPassed: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain null characters [type=value_error, input_value='id\\x00id', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\n", "stderr": "", "exit_code": 0}
23run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q -x"
}
{"status": "ok", "stdout": ".................. [100%]\n=============================== warnings summary ===============================\n../../../../../../../Us
full
{"status": "ok", "stdout": "..................                                                       [100%]\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n18 passed, 1 warning in 1.06s\n", "stderr": "", "exit_code": 0}
24run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..e460b6495 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -34,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..e460b6495 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -34,11 +34,18 @@ class EventSourceResponse(StreamingResponse):\n \n \n def _check_id_no_null(v: str | None) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n     if v is not None and \"\\0\" in v:\n         raise ValueError(\"SSE 'id' must not contain null characters\")\n     return v\n \n \n+\n+def _check_event_no_newlines(v: str | None) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE 'event' must not contain newline characters\")\n+    return v\n class ServerSentEvent(BaseModel):\n     \"\"\"Represents a single Server-Sent Event.\n \n@@ -86,6 +93,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_event_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event type name.\n@@ -219,4 +227,4 @@ KEEPALIVE_COMMENT = b\": ping\\n\\n\"\n \n # Seconds between keep-alive pings when a generator is idle.\n # Private but importable so tests can monkeypatch it.\n-_PING_INTERVAL: float = 15.0\n+_PING_INTERVAL: float = 15.0\n\\ No newline at end of file\n", "stderr": "", "exit_code": 0}
25run_command
{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n# Ensure the file ends with a newline\nif not s.endswith(\"\\n\"):\n    s += \"\\n\"\np.write_text(s)\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
26run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..1c47c3ffd 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -34,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..1c47c3ffd 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -34,11 +34,18 @@ class EventSourceResponse(StreamingResponse):\n \n \n def _check_id_no_null(v: str | None) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n     if v is not None and \"\\0\" in v:\n         raise ValueError(\"SSE 'id' must not contain null characters\")\n     return v\n \n \n+\n+def _check_event_no_newlines(v: str | None) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE 'event' must not contain newline characters\")\n+    return v\n class ServerSentEvent(BaseModel):\n     \"\"\"Represents a single Server-Sent Event.\n \n@@ -86,6 +93,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_event_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event type name.\n", "stderr": "", "exit_code": 0}
27run_command
{
 "command": "submit_patch"
}
{"status": "error", "error_type": "CommandError", "error_message": "/bin/bash: submit_patch: command not found\n", "details": {"stdout": "", "stderr": "/bin/bas
full
{"status": "error", "error_type": "CommandError", "error_message": "/bin/bash: submit_patch: command not found\n", "details": {"stdout": "", "stderr": "/bin/bash: submit_patch: command not found\n", "exit_code": 127}}
28submit_patch
{}
{"status": "ok", "patch_size": 986, "files_changed": 1}
full
{"status": "ok", "patch_size": 986, "files_changed": 1}

Patch

diff --git a/fastapi/sse.py b/fastapi/sse.py
index 901d82496..1c47c3ffd 100644
--- a/fastapi/sse.py
+++ b/fastapi/sse.py
@@ -34,11 +34,18 @@ class EventSourceResponse(StreamingResponse):
 
 
 def _check_id_no_null(v: str | None) -> str | None:
+    if v is not None and "\n" in v:
+        raise ValueError("SSE 'id' must not contain newline characters")
     if v is not None and "\0" in v:
         raise ValueError("SSE 'id' must not contain null characters")
     return v
 
 
+
+def _check_event_no_newlines(v: str | None) -> str | None:
+    if v is not None and "\n" in v:
+        raise ValueError("SSE 'event' must not contain newline characters")
+    return v
 class ServerSentEvent(BaseModel):
     """Represents a single Server-Sent Event.
 
@@ -86,6 +93,7 @@ class ServerSentEvent(BaseModel):
     ] = None
     event: Annotated[
         str | None,
+        AfterValidator(_check_event_no_newlines),
         Doc(
             """
             Optional event type name.

Test output

show
..........F
=================================== FAILURES ===================================
_ test_server_sent_event_single_line_fields_reject_newlines[first\nsecond-event] _

field_name = 'event', value = 'first\nsecond'

    @pytest.mark.parametrize("field_name", ["event", "id"])
    @pytest.mark.parametrize("value", ["first\nsecond", "first\rsecond", "first\r\nsecond"])
    def test_server_sent_event_single_line_fields_reject_newlines(
        field_name: str, value: str
    ):
>       with pytest.raises(ValueError, match=f"SSE '{field_name}' must be a single line"):
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E       AssertionError: Regex pattern did not match.
E         Expected regex: "SSE 'event' must be a single line"
E         Actual message: "1 validation error for ServerSentEvent\nevent\n  Value error, SSE 'event' must not contain newline characters [type=value_error, input_value='first\\nsecond', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error"

tests/test_sse.py:229: AssertionError
=============================== warnings summary ===============================
../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53
  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.
    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 10 passed, 1 warning in 0.45s