diff --git a/sentry_sdk/integrations/_wsgi_common.py b/sentry_sdk/integrations/_wsgi_common.py
index cf1a365209..ef66ea479a 100644
--- a/sentry_sdk/integrations/_wsgi_common.py
+++ b/sentry_sdk/integrations/_wsgi_common.py
@@ -209,6 +209,15 @@ def _is_json_content_type(ct: "Optional[str]") -> bool:
     )
 
 
+def _serialize_request_body_data(data: "Any") -> str:
+    def _default(value: "Any") -> "Any":
+        if isinstance(value, AnnotatedValue):
+            return value.value
+        return str(value)
+
+    return json.dumps(data, default=_default)
+
+
 def _filter_headers(
     headers: "Mapping[str, str]",
     use_annotated_value: bool = True,
diff --git a/sentry_sdk/integrations/flask.py b/sentry_sdk/integrations/flask.py
index 8636dff067..d838bb36f9 100644
--- a/sentry_sdk/integrations/flask.py
+++ b/sentry_sdk/integrations/flask.py
@@ -5,11 +5,16 @@
 from sentry_sdk.integrations._wsgi_common import (
     DEFAULT_HTTP_METHODS_TO_CAPTURE,
     RequestExtractor,
+    _serialize_request_body_data,
+    request_body_within_bounds,
 )
 from sentry_sdk.integrations.wsgi import SentryWsgiMiddleware
 from sentry_sdk.scope import should_send_default_pii
+from sentry_sdk.traces import StreamedSpan, _get_current_streamed_span
 from sentry_sdk.tracing import SOURCE_FOR_STYLE
+from sentry_sdk.tracing_utils import has_span_streaming_enabled
 from sentry_sdk.utils import (
+    AnnotatedValue,
     capture_internal_exceptions,
     ensure_integration_enabled,
     event_from_exception,
@@ -36,9 +41,11 @@
     from flask.signals import (
         before_render_template,
         got_request_exception,
+        request_finished,
         request_started,
     )
     from markupsafe import Markup
+    from werkzeug.exceptions import ClientDisconnected
 except ImportError:
     raise DidNotEnable("Flask is not installed")
 
@@ -88,6 +95,7 @@ def setup_once() -> None:
 
         before_render_template.connect(_add_sentry_trace)
         request_started.connect(_request_started)
+        request_finished.connect(_request_finished)
         got_request_exception.connect(_capture_exception)
 
         old_app = Flask.__call__
@@ -95,10 +103,9 @@ def setup_once() -> None:
         def sentry_patched_wsgi_app(
             self: "Any", environ: "Dict[str, str]", start_response: "Callable[..., Any]"
         ) -> "_ScopedResponse":
-            if sentry_sdk.get_client().get_integration(FlaskIntegration) is None:
-                return old_app(self, environ, start_response)
-
             integration = sentry_sdk.get_client().get_integration(FlaskIntegration)
+            if integration is None:
+                return old_app(self, environ, start_response)
 
             middleware = SentryWsgiMiddleware(
                 lambda *a, **kw: old_app(self, *a, **kw),
@@ -160,6 +167,72 @@ def _request_started(app: "Flask", **kwargs: "Any") -> None:
     scope.add_event_processor(evt_processor)
 
 
+def _request_finished(sender: "Flask", response: "Any", **kwargs: "Any") -> None:
+    integration = sentry_sdk.get_client().get_integration(FlaskIntegration)
+    if integration is None:
+        return
+
+    client = sentry_sdk.get_client()
+    if has_span_streaming_enabled(client.options):
+        request = flask_request._get_current_object()
+        _set_request_body_data_on_streaming_segment(request, client)
+
+
+def _set_request_body_data_on_streaming_segment(
+    request: "Request", client: "sentry_sdk.client.BaseClient"
+) -> None:
+    current_span = _get_current_streamed_span()
+    if type(current_span) is not StreamedSpan:
+        return
+
+    with capture_internal_exceptions():
+        content_length = int(request.content_length or 0)
+
+        # Proceeding without a content length means that we may be consuming the request
+        # without respecting the bounds specified by the user via `max_request_body_size`
+        # option in the SDK.
+        if not content_length:
+            return
+
+        if not request_body_within_bounds(client, content_length):
+            data = AnnotatedValue.substituted_because_over_size_limit()
+        else:
+            raw_data = getattr(request, "_cached_data", None)
+            parsed_body = None
+            if "form" in request.__dict__:
+                extractor = FlaskRequestExtractor(request)
+                parsed_body = extractor.parsed_body()
+            elif raw_data is not None:
+                extractor = FlaskRequestExtractor(request)
+                if extractor.is_json():
+                    parsed_body = extractor.json()
+            else:
+                # The route never read the body via Werkzeug, but it
+                # may have consumed wsgi.input directly. get_data()
+                # raises ClientDisconnected if the stream is exhausted.
+                try:
+                    raw_data = request.get_data()
+                except ClientDisconnected:
+                    raw_data = None
+
+                if raw_data:
+                    extractor = FlaskRequestExtractor(request)
+                    if extractor.is_json():
+                        parsed_body = extractor.json()
+
+            if parsed_body is not None:
+                data = parsed_body
+            elif raw_data:
+                data = AnnotatedValue.substituted_because_raw_data()
+            else:
+                return
+
+        current_span._segment.set_attribute(
+            "http.request.body.data",
+            _serialize_request_body_data(data),
+        )
+
+
 class FlaskRequestExtractor(RequestExtractor):
     def env(self) -> "Dict[str, str]":
         return self.request.environ
diff --git a/sentry_sdk/integrations/starlette.py b/sentry_sdk/integrations/starlette.py
index ddc58d7183..a4882210ad 100644
--- a/sentry_sdk/integrations/starlette.py
+++ b/sentry_sdk/integrations/starlette.py
@@ -1,5 +1,4 @@
 import functools
-import json
 import sys
 import warnings
 from collections.abc import Set
@@ -18,6 +17,7 @@
     DEFAULT_HTTP_METHODS_TO_CAPTURE,
     HttpCodeRangeContainer,
     _is_json_content_type,
+    _serialize_request_body_data,
     request_body_within_bounds,
 )
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
@@ -241,16 +241,6 @@ async def _sentry_send(*args: "Any", **kwargs: "Any") -> "Any":
     return middleware_class
 
 
-def _serialize_request_body_data(data: "Any") -> str:
-    # data may be a JSON-serializable value, an AnnotatedValue, or a dict with AnnotatedValue values
-    def _default(value: "Any") -> "Any":
-        if isinstance(value, AnnotatedValue):
-            return value.value
-        return str(value)
-
-    return json.dumps(data, default=_default)
-
-
 def _set_request_body_data_on_streaming_segment(
     info: "Optional[Dict[str, Any]]",
 ) -> None:
diff --git a/tests/integrations/flask/test_flask.py b/tests/integrations/flask/test_flask.py
index 8d2d1b3c95..a2e71555a6 100644
--- a/tests/integrations/flask/test_flask.py
+++ b/tests/integrations/flask/test_flask.py
@@ -15,6 +15,8 @@
 from flask.views import View
 from flask_login import LoginManager, login_user
 
+from sentry_sdk.traces import SpanStatus
+
 try:
     from werkzeug.wrappers.request import UnsupportedMediaType
 except ImportError:
@@ -27,6 +29,11 @@
     capture_message,
     set_tag,
 )
+from sentry_sdk._types import (
+    OVER_SIZE_LIMIT_SUBSTITUTE,
+    SENSITIVE_DATA_SUBSTITUTE,
+    UNPARSABLE_RAW_DATA_SUBSTITUTE,
+)
 from sentry_sdk.consts import DEFAULT_MAX_VALUE_LENGTH
 from sentry_sdk.integrations.logging import LoggingIntegration
 from sentry_sdk.serializer import MAX_DATABAG_BREADTH
@@ -92,29 +99,49 @@ def test_has_context(sentry_init, app, capture_events):
         ("/message/123456", "url", "/message/<int:message_id>", "route"),
     ],
 )
+@pytest.mark.parametrize("span_streaming", [True, False])
 def test_transaction_style(
     sentry_init,
     app,
     capture_events,
+    capture_items,
     url,
     transaction_style,
     expected_transaction,
     expected_source,
+    span_streaming,
 ):
     sentry_init(
         integrations=[
             flask_sentry.FlaskIntegration(transaction_style=transaction_style)
-        ]
+        ],
+        traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
     )
-    events = capture_events()
+
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        events = capture_events()
 
     client = app.test_client()
     response = client.get(url)
     assert response.status_code == 200
 
-    (event,) = events
-    assert event["transaction"] == expected_transaction
-    assert event["transaction_info"] == {"source": expected_source}
+    if span_streaming:
+        sentry_sdk.flush()
+
+        assert len(items) == 1
+
+        span = items[0].payload
+
+        assert span["is_segment"] is True
+        assert span["name"] == expected_transaction
+        assert span["attributes"]["sentry.span.source"] == expected_source
+    else:
+        message_event, transaction_event = events
+        assert transaction_event["transaction"] == expected_transaction
+        assert transaction_event["transaction_info"] == {"source": expected_source}
 
 
 @pytest.mark.parametrize("debug", (True, False))
@@ -379,13 +406,18 @@ def index():
     assert len(event["request"]["data"]["foo"]) == DEFAULT_MAX_VALUE_LENGTH
 
 
-def test_flask_formdata_request_appear_transaction_body(
-    sentry_init, capture_events, app
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_flask_formdata_request_appear_in_segment_or_transaction_body(
+    sentry_init, capture_events, capture_items, app, span_streaming
 ):
     """
-    Test that ensures that transaction request data contains body, even if no exception was raised
+    Test that ensures that transaction/segment request data contains body, even if no exception was raised
     """
-    sentry_init(integrations=[flask_sentry.FlaskIntegration()], traces_sample_rate=1.0)
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
+    )
 
     data = {"username": "sentry-user", "age": "26"}
 
@@ -403,17 +435,29 @@ def index():
         capture_message("hi")
         return "ok"
 
-    events = capture_events()
+    if span_streaming:
+        items = capture_items("event", "span")
+    else:
+        events = capture_events()
 
     client = app.test_client()
     response = client.post("/", data=data)
     assert response.status_code == 200
 
-    event, transaction_event = events
+    if span_streaming:
+        sentry_sdk.flush()
+
+        spans = [i for i in items if i.type == "span"]
+        assert len(spans) == 1
+        span = spans[0].payload
+
+        assert span["attributes"]["http.request.body.data"] == json.dumps(data)
+    else:
+        event, transaction_event = events
 
-    assert "request" in transaction_event
-    assert "data" in transaction_event["request"]
-    assert transaction_event["request"]["data"] == data
+        assert "request" in transaction_event
+        assert "data" in transaction_event["request"]
+        assert transaction_event["request"]["data"] == data
 
 
 @pytest.mark.parametrize("input_char", ["a", b"a"])
@@ -750,8 +794,15 @@ def zerodivision(e):
     assert not events
 
 
-def test_tracing_success(sentry_init, capture_events, app):
-    sentry_init(traces_sample_rate=1.0, integrations=[flask_sentry.FlaskIntegration()])
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_tracing_success(
+    sentry_init, capture_events, capture_items, app, span_streaming
+):
+    sentry_init(
+        traces_sample_rate=1.0,
+        integrations=[flask_sentry.FlaskIntegration()],
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
+    )
 
     @app.before_request
     def _():
@@ -763,30 +814,61 @@ def hi_tx():
         capture_message("hi")
         return "ok"
 
-    events = capture_events()
+    if span_streaming:
+        items = capture_items("event", "span")
+    else:
+        events = capture_events()
 
     with app.test_client() as client:
         response = client.get("/message_tx")
         assert response.status_code == 200
 
-    message_event, transaction_event = events
+    if span_streaming:
+        sentry_sdk.flush()
+
+        spans = [i for i in items if i.type == "span"]
+        message_events = [i for i in items if i.type == "event"]
+
+        assert len(spans) == 1
+        span = spans[0].payload
+        assert span["is_segment"] is True
+        assert span["name"] == "hi_tx"
+        assert span["status"] == SpanStatus.OK
+        assert span["attributes"]["sentry.op"] == "http.server"
+        assert span["attributes"]["sentry.origin"] == "auto.http.flask"
+
+        assert len(message_events) == 1
+        assert message_events[0].payload["message"] == "hi"
+        assert message_events[0].payload["transaction"] == "hi_tx"
+        assert message_events[0].payload["tags"]["view"] == "yes"
+        assert message_events[0].payload["tags"]["before_request"] == "yes"
+    else:
+        message_event, transaction_event = events
 
-    assert transaction_event["type"] == "transaction"
-    assert transaction_event["transaction"] == "hi_tx"
-    assert transaction_event["contexts"]["trace"]["status"] == "ok"
-    assert transaction_event["tags"]["view"] == "yes"
-    assert transaction_event["tags"]["before_request"] == "yes"
+        assert transaction_event["type"] == "transaction"
+        assert transaction_event["transaction"] == "hi_tx"
+        assert transaction_event["contexts"]["trace"]["status"] == "ok"
+        assert transaction_event["tags"]["view"] == "yes"
+        assert transaction_event["tags"]["before_request"] == "yes"
 
-    assert message_event["message"] == "hi"
-    assert message_event["transaction"] == "hi_tx"
-    assert message_event["tags"]["view"] == "yes"
-    assert message_event["tags"]["before_request"] == "yes"
+        assert message_event["message"] == "hi"
+        assert message_event["transaction"] == "hi_tx"
+        assert message_event["tags"]["view"] == "yes"
+        assert message_event["tags"]["before_request"] == "yes"
 
 
-def test_tracing_error(sentry_init, capture_events, app):
-    sentry_init(traces_sample_rate=1.0, integrations=[flask_sentry.FlaskIntegration()])
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_tracing_error(sentry_init, capture_events, capture_items, app, span_streaming):
+    sentry_init(
+        traces_sample_rate=1.0,
+        integrations=[flask_sentry.FlaskIntegration()],
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
+    )
 
-    events = capture_events()
+    if span_streaming:
+        items = capture_items("event", "span")
+    else:
+        events = capture_events()
 
     @app.route("/error")
     def error():
@@ -797,15 +879,33 @@ def error():
             response = client.get("/error")
             assert response.status_code == 500
 
-    error_event, transaction_event = events
+    if span_streaming:
+        sentry_sdk.flush()
+
+        spans = [i for i in items if i.type == "span"]
+        error_events = [i for i in items if i.type == "event"]
+
+        assert len(spans) == 1
+        span = spans[0].payload
+        assert span["is_segment"] is True
+        assert span["name"] == "error"
+        assert span["status"] == SpanStatus.ERROR
+        assert span["attributes"]["sentry.op"] == "http.server"
+        assert span["attributes"]["sentry.origin"] == "auto.http.flask"
+        assert len(error_events) == 1
+        assert error_events[0].payload["transaction"] == "error"
+        (exception,) = error_events[0].payload["exception"]["values"]
+        assert exception["type"] == "ZeroDivisionError"
+    else:
+        error_event, transaction_event = events
 
-    assert transaction_event["type"] == "transaction"
-    assert transaction_event["transaction"] == "error"
-    assert transaction_event["contexts"]["trace"]["status"] == "internal_error"
+        assert transaction_event["type"] == "transaction"
+        assert transaction_event["transaction"] == "error"
+        assert transaction_event["contexts"]["trace"]["status"] == "internal_error"
 
-    assert error_event["transaction"] == "error"
-    (exception,) = error_event["exception"]["values"]
-    assert exception["type"] == "ZeroDivisionError"
+        assert error_event["transaction"] == "error"
+        (exception,) = error_event["exception"]["values"]
+        assert exception["type"] == "ZeroDivisionError"
 
 
 def test_error_has_trace_context_if_tracing_disabled(sentry_init, capture_events, app):
@@ -924,92 +1024,134 @@ def index():
     assert event["request"]["headers"]["Proxy-Authorization"] == "[Filtered]"
 
 
-def test_response_status_code_ok_in_transaction_context(
-    sentry_init, capture_envelopes, app
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_response_status_code_ok(
+    sentry_init, capture_envelopes, capture_items, app, span_streaming
 ):
     """
-    Tests that the response status code is added to the transaction context.
+    Tests that the response status code is added to the transaction/span.
     This also works for when there is an Exception during the request, but somehow the test flask app doesn't seem to trigger that.
     """
     sentry_init(
         integrations=[flask_sentry.FlaskIntegration()],
         traces_sample_rate=1.0,
         release="demo-release",
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
     )
 
-    envelopes = capture_envelopes()
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        envelopes = capture_envelopes()
 
     client = app.test_client()
     client.get("/message")
 
     sentry_sdk.get_client().flush()
 
-    (_, transaction_envelope, _) = envelopes
-    transaction = transaction_envelope.get_transaction_event()
+    if span_streaming:
+        assert len(items) == 1
+        span = items[0].payload
+        assert span["attributes"]["http.response.status_code"] == 200
+        assert span["status"] == SpanStatus.OK
+    else:
+        (_, transaction_envelope, _) = envelopes
+        transaction = transaction_envelope.get_transaction_event()
 
-    assert transaction["type"] == "transaction"
-    assert len(transaction["contexts"]) > 0
-    assert "response" in transaction["contexts"].keys(), (
-        "Response context not found in transaction"
-    )
-    assert transaction["contexts"]["response"]["status_code"] == 200
+        assert transaction["type"] == "transaction"
+        assert len(transaction["contexts"]) > 0
+        assert "response" in transaction["contexts"].keys(), (
+            "Response context not found in transaction"
+        )
+        assert transaction["contexts"]["response"]["status_code"] == 200
 
 
-def test_response_status_code_not_found_in_transaction_context(
-    sentry_init, capture_envelopes, app
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_response_status_code_not_found(
+    sentry_init, capture_envelopes, capture_items, app, span_streaming
 ):
     sentry_init(
         integrations=[flask_sentry.FlaskIntegration()],
         traces_sample_rate=1.0,
         release="demo-release",
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
     )
 
-    envelopes = capture_envelopes()
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        envelopes = capture_envelopes()
 
     client = app.test_client()
     client.get("/not-existing-route")
 
     sentry_sdk.get_client().flush()
 
-    (transaction_envelope, _) = envelopes
-    transaction = transaction_envelope.get_transaction_event()
+    if span_streaming:
+        assert len(items) == 1
+        span = items[0].payload
+        assert span["attributes"]["http.response.status_code"] == 404
+        assert span["status"] == SpanStatus.ERROR
+    else:
+        (transaction_envelope, _) = envelopes
+        transaction = transaction_envelope.get_transaction_event()
 
-    assert transaction["type"] == "transaction"
-    assert len(transaction["contexts"]) > 0
-    assert "response" in transaction["contexts"].keys(), (
-        "Response context not found in transaction"
-    )
-    assert transaction["contexts"]["response"]["status_code"] == 404
+        assert transaction["type"] == "transaction"
+        assert len(transaction["contexts"]) > 0
+        assert "response" in transaction["contexts"].keys(), (
+            "Response context not found in transaction"
+        )
+        assert transaction["contexts"]["response"]["status_code"] == 404
 
 
-def test_span_origin(sentry_init, app, capture_events):
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_span_origin(sentry_init, app, capture_events, capture_items, span_streaming):
     sentry_init(
         integrations=[flask_sentry.FlaskIntegration()],
         traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
     )
-    events = capture_events()
+
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        events = capture_events()
 
     client = app.test_client()
     client.get("/message")
 
-    (_, event) = events
+    if span_streaming:
+        sentry_sdk.flush()
 
-    assert event["contexts"]["trace"]["origin"] == "auto.http.flask"
+        assert len(items) == 1
+        span = items[0].payload
+        assert span["attributes"]["sentry.origin"] == "auto.http.flask"
+    else:
+        (_, event) = events
+        assert event["contexts"]["trace"]["origin"] == "auto.http.flask"
 
 
-def test_transaction_http_method_default(
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_http_method_default(
     sentry_init,
     app,
     capture_events,
+    capture_items,
+    span_streaming,
 ):
     """
-    By default OPTIONS and HEAD requests do not create a transaction.
+    By default OPTIONS and HEAD requests do not create a transaction/segment.
     """
     sentry_init(
         traces_sample_rate=1.0,
         integrations=[flask_sentry.FlaskIntegration()],
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
     )
-    events = capture_events()
+
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        events = capture_events()
 
     client = app.test_client()
     response = client.get("/nomessage")
@@ -1021,16 +1163,25 @@ def test_transaction_http_method_default(
     response = client.head("/nomessage")
     assert response.status_code == 200
 
-    (event,) = events
+    if span_streaming:
+        sentry_sdk.flush()
 
-    assert len(events) == 1
-    assert event["request"]["method"] == "GET"
+        assert len(items) == 1
+        span = items[0].payload
+        assert span["attributes"]["http.request.method"] == "GET"
+    else:
+        assert len(events) == 1
+        (event,) = events
+        assert event["request"]["method"] == "GET"
 
 
-def test_transaction_http_method_custom(
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_http_method_custom(
     sentry_init,
     app,
     capture_events,
+    capture_items,
+    span_streaming,
 ):
     """
     Configure FlaskIntegration to ONLY capture OPTIONS and HEAD requests.
@@ -1045,8 +1196,13 @@ def test_transaction_http_method_custom(
                 )  # capitalization does not matter
             )  # case does not matter
         ],
+        _experiments={"trace_lifecycle": "stream"} if span_streaming else {},
     )
-    events = capture_events()
+
+    if span_streaming:
+        items = capture_items("span")
+    else:
+        events = capture_events()
 
     client = app.test_client()
     response = client.get("/nomessage")
@@ -1058,8 +1214,281 @@ def test_transaction_http_method_custom(
     response = client.head("/nomessage")
     assert response.status_code == 200
 
-    assert len(events) == 2
+    if span_streaming:
+        sentry_sdk.flush()
+
+        assert len(items) == 2
+        assert items[0].payload["attributes"]["http.request.method"] == "OPTIONS"
+        assert items[1].payload["attributes"]["http.request.method"] == "HEAD"
+    else:
+        assert len(events) == 2
+        (event1, event2) = events
+        assert event1["request"]["method"] == "OPTIONS"
+        assert event2["request"]["method"] == "HEAD"
+
+
+def test_request_body_captured_on_segment_span_streaming(
+    sentry_init, capture_items, app
+):
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream"},
+    )
+
+    body = {"key": "value"}
+
+    @app.route("/body", methods=["POST"])
+    def body_endpoint():
+        request.get_json()
+        return "ok"
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    response = client.post("/body", json=body)
+    assert response.status_code == 200
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
+    assert span["attributes"]["http.request.body.data"] == json.dumps(body)
+
+
+def test_request_body_not_read_span_streaming(sentry_init, capture_items, app):
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        max_request_body_size="never",
+        _experiments={"trace_lifecycle": "stream"},
+    )
+
+    @app.route("/body", methods=["POST"])
+    def body_endpoint():
+        request.get_json()
+        return "ok"
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    response = client.post("/body", json={"key": "value"})
+    assert response.status_code == 200
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
+    assert span["attributes"]["http.request.body.data"] == json.dumps(
+        OVER_SIZE_LIMIT_SUBSTITUTE
+    )
+
+
+def test_request_body_over_size_limit_span_streaming(sentry_init, capture_items, app):
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        max_request_body_size="small",
+        _experiments={"trace_lifecycle": "stream"},
+    )
+
+    @app.route("/body", methods=["POST"])
+    def body_endpoint():
+        request.get_data()
+        return "ok"
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    response = client.post("/body", data=b"x" * 2000)
+    assert response.status_code == 200
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
+    assert span["attributes"]["http.request.body.data"] == json.dumps(
+        OVER_SIZE_LIMIT_SUBSTITUTE
+    )
+
+
+def test_request_body_raw_data_substituted_span_streaming(
+    sentry_init, capture_items, app
+):
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        max_request_body_size="always",
+        _experiments={"trace_lifecycle": "stream"},
+    )
+
+    @app.route("/body", methods=["POST"])
+    def body_endpoint():
+        request.get_data()
+        return "ok"
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    response = client.post(
+        "/body", data=b"some raw bytes", content_type="application/octet-stream"
+    )
+    assert response.status_code == 200
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
+    assert span["attributes"]["http.request.body.data"] == json.dumps(
+        UNPARSABLE_RAW_DATA_SUBSTITUTE
+    )
+
+
+def test_sensitive_header_scrubbing_span_streaming(sentry_init, capture_items, app):
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream"},
+    )
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    response = client.get(
+        "/message",
+        headers={
+            "Authorization": "Bearer secret-token",
+            "X-Custom-Header": "passthrough",
+        },
+    )
+    assert response.status_code == 200
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
+    assert (
+        span["attributes"]["http.request.header.authorization"]
+        == SENSITIVE_DATA_SUBSTITUTE
+    )
+    assert span["attributes"]["http.request.header.x-custom-header"] == "passthrough"
+
+
+def test_request_body_captured_when_route_ignores_body_span_streaming(
+    sentry_init, capture_items, app
+):
+    """
+    When the route handler never reads the request body, the SDK should
+    still capture it in _request_finished via request.get_data().
+    """
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        max_request_body_size="always",
+        _experiments={"trace_lifecycle": "stream"},
+    )
+
+    body = {"key": "value"}
+
+    @app.route("/ignore-body", methods=["POST"])
+    def ignore_body_endpoint():
+        return "ok"
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    response = client.post("/ignore-body", json=body)
+    assert response.status_code == 200
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
+    assert span["attributes"]["http.request.body.data"] == json.dumps(body)
+
+
+def test_client_disconnected_handled_gracefully_span_streaming(
+    sentry_init, capture_items, app
+):
+    from unittest.mock import patch
+
+    from werkzeug.exceptions import ClientDisconnected
+
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        max_request_body_size="always",
+        _experiments={"trace_lifecycle": "stream"},
+    )
+
+    @app.route("/disconnect", methods=["POST"])
+    def disconnect_endpoint():
+        # Simulate a client that disconnected: patch get_data so that
+        # when _request_finished tries to read the body it raises.
+        request._cached_data = None
+        request.__dict__.pop("form", None)
+        patch.object(
+            type(request._get_current_object()),
+            "get_data",
+            side_effect=ClientDisconnected(),
+        ).start()
+        return "ok"
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    try:
+        response = client.post(
+            "/disconnect",
+            data=b'{"key": "value"}',
+            content_type="application/json",
+        )
+        assert response.status_code == 200
+    finally:
+        patch.stopall()
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
+    assert "http.request.body.data" not in span.get("attributes", {})
+
+
+def test_wsgi_input_direct_read_does_not_hang_span_streaming(
+    sentry_init, capture_items, app
+):
+    """
+    Regression test: reading wsgi.input directly must not hang when span streaming is enabled.
+    The SDK must not consume wsgi.input before user code runs.
+    """
+    sentry_init(
+        integrations=[flask_sentry.FlaskIntegration()],
+        traces_sample_rate=1.0,
+        _experiments={"trace_lifecycle": "stream"},
+        max_request_body_size="always",
+    )
+
+    @app.route("/raw-wsgi", methods=["POST"])
+    def raw_wsgi_endpoint():
+        content_length = int(request.environ.get("CONTENT_LENGTH", 0))
+        body = request.environ["wsgi.input"].read(content_length)
+        return {"size": len(body), "body": body.decode("utf-8", errors="replace")}
+
+    items = capture_items("span")
+
+    client = app.test_client()
+    response = client.post(
+        "/raw-wsgi", data=b"hello from test", content_type="text/plain"
+    )
+    assert response.status_code == 200
+    assert response.get_json()["body"] == "hello from test"
+
+    sentry_sdk.flush()
+
+    assert len(items) == 1
+    span = items[0].payload
 
-    (event1, event2) = events
-    assert event1["request"]["method"] == "OPTIONS"
-    assert event2["request"]["method"] == "HEAD"
+    # The SDK should not have captured the body since the user read wsgi.input
+    # directly (bypassing Werkzeug's cache). This is an acceptable trade-off
+    # vs. consuming the stream and causing user applications to hang.
+    assert "http.request.body.data" not in span.get("attributes", {})