← pod_r2_v1f

fastapi_15588

failed WRONG_FIX wrong_fix · 24 tool calls · 170 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 (25)

#ToolArgumentsResult
1run_command
{
 "command": "grep -rn \"EventSourceResponse\" ."
}
{"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": "ls -R fastapi/sse"
}
{"status": "error", "error_type": "CommandError", "error_message": "ls: fastapi/sse: No such file or directory\n", "details": {"stdout": "", "stderr": "ls: fast
full
{"status": "error", "error_type": "CommandError", "error_message": "ls: fastapi/sse: No such file or directory\n", "details": {"stdout": "", "stderr": "ls: fastapi/sse: No such file or directory\n", "exit_code": 1}}
3run_command
{
 "command": "find fastapi -name \"*sse*\""
}
{"status": "ok", "stdout": "fastapi/sse.py\n", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "fastapi/sse.py\n", "stderr": "", "exit_code": 0}
4run_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
5run_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
6run_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}
7run_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}
8run_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}
9run_command
{
 "command": "python3 -m pytest tests/test_sse.py"
}
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, p
full
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /private/workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\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\n======================== 18 passed, 1 warning in 1.05s =========================\n", "stderr": "", "exit_code": 0}
10run_command
{
 "command": "cat > /tmp/test_sse_validation.py <<'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=\"hello\\nworld\", data=\"some data\")\n\n@app.get(\"/sse-id\")\nasync def sse_id_endpoint():\n    yield ServerSentEvent(event=\"hello\", id=\"id\\nwith\\nnewline\", data=\"some data\")\n\nclient = TestClient(app)\n\ndef test_sse_event_newline():\n    with pytest.raises(ValueError, match=\"SSE 'event' must not contain newline characters\"):\n        ServerSentEvent(event=\"hello\\nworld\", data=\"some data\")\n\ndef test_sse_id_newline():\n    with pytest.raises(ValueError, match=\"SSE 'id' must not contain newline characters\"):\n        ServerSentEvent(event=\"hello\", id=\"id\\nwith\\nnewline\", data=\"some data\")\n\ndef test_sse_valid():\n    event = ServerSentEvent(event=\"hello\", id=\"123\", data=\"some data\")\n    assert event.event == \"hello\"\n    assert event.id == \"123\"\nPY\npython3 -m pytest /tmp/test_sse_validation.py\n"
}
{"status": "error", "error_type": "CommandError", "error_message": "============================= test session starts ==============================\nplatform d
full
{"status": "error", "error_type": "CommandError", "error_message": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /tmp\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 3 items\n\n../../../../../../../tmp/test_sse_validation.py F [ 33%]\nF.                                                                       [100%]\n\n=================================== FAILURES ===================================\n____________________________ test_sse_event_newline ____________________________\n\n    def test_sse_event_newline():\n>       with pytest.raises(ValueError, match=\"SSE 'event' must not contain newline characters\"):\n             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nE       Failed: DID NOT RAISE ValueError\n\n/tmp/test_sse_validation.py:19: Failed\n_____________________________ test_sse_id_newline ______________________________\n\n    def test_sse_id_newline():\n>       with pytest.raises(ValueError, match=\"SSE 'id' must not contain newline characters\"):\n             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nE       Failed: DID NOT RAISE ValueError\n\n/tmp/test_sse_validation.py:23: Failed\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[[], AbstractC
11run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\ndef check_newline(v: str | None) -> str | None:\n    if v is not None and \"\\n\" in v:\n        raise ValueError(\"SSE field must not contain newline characters\")\n    return v\n\n# We need to adapt this to be more specific for 'id' and 'event'\n# But the current _check_id_no_null is only for id.\n\n# Let's define a more general one or specific ones.\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
12run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# Define the validators\nnew_validators = \"\"\"\ndef _check_sse_field_no_newline(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE field must not contain newline characters\")\n    return v\n\"\"\"\n\n# We want to use it for 'event' and 'id'.\n# For 'id', it already has AfterValidator(_check_id_no_null).\n# We can add another AfterValidator or combine them.\n\n# Let's see where they are.\n# event: Annotated[str | None, Doc(...)] = None\n# id: Annotated[str | None, AfterValidator(_check_id_no_null), Doc(...)] = None\n\n# Let's just add the validator function and update the Annotated types.\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
13run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# Add the validator function\n# It should be before ServerSentEvent class.\n# _check_id_no_null is at line 30.\n\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\"\"\"\n\nnew_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    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\ndef _check_event_no_newline(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.replace(old_id_check, new_id_check)\n\n# Update event field\n# event: Annotated[\n#     str | None,\n#     Doc(\n#         \"\"\"\n#         Optional event type name.\n#         ...\n#         \"\"\"\n#     ),\n# ] = None\n\n# We need to insert AfterValidator(_check_event_no_newline) before Doc.\n\n# Let's find the event field definition.\n# It's around line 78.\n\n# Using a more robust replacement for event field.\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# I'll use a regex or a simple replace if I can be precise.\n\n# Let's try to find the exact block for event.\n# I'll use the current content of s to be sure.\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
14run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# 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 \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\ndef _check_event_no_newline(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.replace(old_id_check, new_id_check)\n\n# Update event field\n# We want to add AfterValidator(_check_event_no_newline),\n# before Doc(...)\n\n# The event field starts with:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n\n# Let's replace:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n# with:\n#     event: Annotated[\n#         str | None,\n#         AfterValidator(_check_event_no_newline),\n#         Doc(\n\n# I'll use the exact indentation.\nold_event_start = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\"\"\"\"\nnew_event_start = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_event_no_newline),\n        Doc(\"\"\"\"\n\n# Wait, the Doc is a triple-quoted string.\n# Let's check the file content again.\nPY\n"
}
{"status": "error", "error_type": "CommandError", "error_message": " File \"<stdin>\", line 47\n Doc(\"\"\"\"\n ^\nSyntaxError: unterminated string literal (det
full
{"status": "error", "error_type": "CommandError", "error_message": "  File \"<stdin>\", line 47\n    Doc(\"\"\"\"\n           ^\nSyntaxError: unterminated string literal (detected at line 47)\n", "details": {"stdout": "", "stderr": "  File \"<stdin>\", line 47\n    Doc(\"\"\"\"\n           ^\nSyntaxError: unterminated string literal (detected at line 47)\n", "exit_code": 1}}
15run_command
{
 "command": "sed -n '70,85p' fastapi/sse.py"
}
{"status": "ok", "stdout": " \"\"\"\n ),\n ] = None\n raw_data: Annotated[\n str | None,\n Doc(\n \"\"\"\n Raw string to send as the `data:` field **without** J
full
{"status": "ok", "stdout": "            \"\"\"\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-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", "stderr": "", "exit_code": 0}
16run_command
{
 "command": "sed -n '75,95p' fastapi/sse.py"
}
{"status": "ok", "stdout": " Doc(\n \"\"\"\n Raw string to send as the `data:` field **without** JSON encoding.\n\n Use this when you need to send pre-formatted
full
{"status": "ok", "stdout": "        Doc(\n            \"\"\"\n            Raw string to send as the `data:` field **without** JSON encoding.\n\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            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", "stderr": "", "exit_code": 0}
17run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# 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 \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\ndef _check_event_no_newline(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.replace(old_id_check, new_id_check)\n\n# Update event field\n# We want to replace:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n# with:\n#     event: Annotated[\n#         str | None,\n#         AfterValidator(_check_event_no_newline),\n#         Doc(\n\n# I'll use a more precise replacement.\nold_event_start = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\"\"\"\"\n# This is tricky because of the triple quotes.\n# Let's use the fact that the line is exactly:\n#         Doc(\n\n# I'll just use a simple string replacement for the line.\n# But I need to make sure it's the right one.\n\n# Let's try replacing the line \"        Doc(\" that follows \"        str | None,\" and \"    event: Annotated[\".\n\n# Actually, I can just use s.replace() with a larger block.\n# I'll use the exact lines from sed output.\n\n# Let's try this:\nold_event_block = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\n            \\\"\\\"\\\"\n            Optional event type name.\"\"\"\nnew_event_block = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_event_no_newline),\n        Doc(\n            \\\"\\\"\\\"\n            Optional event type name.\"\"\"\n\n# Wait, the sed output shows:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n#             \"\"\"\n#             Optional event type name.\n\n# Let's try to be very careful.\nPY\n"
}
{"status": "error", "error_type": "CommandError", "error_message": " File \"<stdin>\", line 39\n Doc(\"\"\"\"\n ^\nSyntaxError: unterminated string literal (det
full
{"status": "error", "error_type": "CommandError", "error_message": "  File \"<stdin>\", line 39\n    Doc(\"\"\"\"\n           ^\nSyntaxError: unterminated string literal (detected at line 39)\n", "details": {"stdout": "", "stderr": "  File \"<stdin>\", line 39\n    Doc(\"\"\"\"\n           ^\nSyntaxError: unterminated string literal (detected at line 39)\n", "exit_code": 1}}
18run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# 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 \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\ndef _check_event_no_newline(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.replace(old_id_check, new_id_check)\n\n# Update event field\n# I will use a simple line-by-line replacement.\nlines = s.splitlines()\nfor i, line in enumerate(lines):\n    if line.strip() == \"Doc(\" and i > 0 and \"str | None,\" in lines[i-1] and \"event: Annotated[\" in lines[i-2]:\n        lines.insert(i, \"        AfterValidator(_check_event_no_newline),\")\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}
19run_command
{
 "command": "python3 -m pytest /tmp/test_sse_validation.py"
}
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, p
full
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /tmp\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 3 items\n\n../../../../../../../tmp/test_sse_validation.py . [ 33%]\n..                                                                       [100%]\n\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\n========================= 3 passed, 1 warning in 0.31s =========================\n", "stderr": "", "exit_code": 0}
20run_command
{
 "command": "python3 -m pytest tests/test_sse.py"
}
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, p
full
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /private/workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\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\n======================== 18 passed, 1 warning in 1.04s =========================\n", "stderr": "", "exit_code": 0}
21run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..8af3e424b 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -36,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..8af3e424b 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -36,6 +36,13 @@ class EventSourceResponse(StreamingResponse):\n 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+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n+    return v\n+\n+def _check_event_no_newline(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 \n@@ -86,6 +93,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_event_no_newline),\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}
22run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n# Remove the trailing whitespace/newline issue if any, but the diff shows it's just a newline change.\n# Actually, the diff shows:\n# -_PING_INTERVAL: float = 15.0\n# +_PING_INTERVAL: float = 15.0\n# \\ No newline at end of file\n# This means I might have removed the trailing newline.\n\n# Let's just make sure 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}
23run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..418551dc0 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -36,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..418551dc0 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -36,6 +36,13 @@ class EventSourceResponse(StreamingResponse):\n 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+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n+    return v\n+\n+def _check_event_no_newline(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 \n@@ -86,6 +93,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_event_no_newline),\n         Doc(\n             \"\"\"\n             Optional event type name.\n", "stderr": "", "exit_code": 0}
24run_command
{
 "command": "python3 -m pytest /tmp/test_sse_validation.py && python3 -m pytest tests/test_sse.py"
}
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, p
full
{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /tmp\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 3 items\n\n../../../../../../../tmp/test_sse_validation.py . [ 33%]\n..                                                                       [100%]\n\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\n========================= 3 passed, 1 warning in 0.27s =========================\n============================= test session starts ==============================\nplatform darwin -- Python 3.13.15, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: not supported, timer_resolution: 41.7ns)\nrootdir: /private/workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: cov-7.1.0, xdist-3.8.0, timeout-2.4.0, codspeed-5.0.3, sugar-1.1.1, anyio-4.15.1, inline-snapshot-0.35.4\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-po
25submit_patch
{}
{"status": "ok", "patch_size": 895, "files_changed": 1}
full
{"status": "ok", "patch_size": 895, "files_changed": 1}

Patch

diff --git a/fastapi/sse.py b/fastapi/sse.py
index 901d82496..418551dc0 100644
--- a/fastapi/sse.py
+++ b/fastapi/sse.py
@@ -36,6 +36,13 @@ class EventSourceResponse(StreamingResponse):
 def _check_id_no_null(v: str | None) -> str | None:
     if v is not None and "\0" in v:
         raise ValueError("SSE 'id' must not contain null characters")
+    if v is not None and "\n" in v:
+        raise ValueError("SSE 'id' must not contain newline characters")
+    return v
+
+def _check_event_no_newline(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
 
 
@@ -86,6 +93,7 @@ class ServerSentEvent(BaseModel):
     ] = None
     event: Annotated[
         str | None,
+        AfterValidator(_check_event_no_newline),
         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.44s