diff --git a/src/core/types.ts b/src/core/types.ts
index 90d1817ea..d120fbfb5 100644
--- a/src/core/types.ts
+++ b/src/core/types.ts
@@ -2,7 +2,7 @@ export type Action = "advance" | "replace" | "restore"
 
 export type Position = { x: number; y: number }
 
-export type StreamSource = {
+export type StreamSource = (EventSource | WebSocket) & {
   addEventListener(
     type: "message",
     listener: (event: MessageEvent) => void,
diff --git a/src/elements/stream_source_element.ts b/src/elements/stream_source_element.ts
index 48f97e5fe..c12047b83 100644
--- a/src/elements/stream_source_element.ts
+++ b/src/elements/stream_source_element.ts
@@ -12,6 +12,8 @@ export class StreamSourceElement extends HTMLElement {
 
   disconnectedCallback() {
     if (this.streamSource) {
+      this.streamSource.close()
+
       disconnectStreamSource(this.streamSource)
     }
   }
diff --git a/src/tests/functional/stream_tests.ts b/src/tests/functional/stream_tests.ts
index 3d159b2e3..0fb604726 100644
--- a/src/tests/functional/stream_tests.ts
+++ b/src/tests/functional/stream_tests.ts
@@ -1,6 +1,7 @@
-import { test } from "@playwright/test"
+import { Page, test } from "@playwright/test"
 import { assert } from "chai"
 import { nextBeat, nextEventNamed, readEventLogs, waitUntilNoSelector, waitUntilText } from "../helpers/page"
+import { StreamSourceElement } from "../../elements/stream_source_element"
 
 test.beforeEach(async ({ page }) => {
   await page.goto("/src/tests/fixtures/stream.html")
@@ -104,6 +105,9 @@ test("test receiving a stream message over SSE", async ({ page }) => {
       `<turbo-stream-source id="stream-source" src="/__turbo/messages"></turbo-stream-source>`
     )
   })
+  await nextBeat()
+  assert.equal(await getReadyState(page, "stream-source"), await page.evaluate(() => EventSource.OPEN))
+
   const messages = await page.locator("#messages .message")
 
   assert.deepEqual(await messages.allTextContents(), ["First"])
@@ -113,11 +117,33 @@ test("test receiving a stream message over SSE", async ({ page }) => {
   await waitUntilText(page, "Hello world!")
   assert.deepEqual(await messages.allTextContents(), ["First", "Hello world!"])
 
-  await page.evaluate(() => document.getElementById("stream-source")?.remove())
-  await nextBeat()
+  const readyState = await page.evaluate((id) => {
+    const element = document.getElementById(id) as StreamSourceElement
+
+    if (element && element.streamSource) {
+      element.remove()
+
+      return element.streamSource.readyState
+    } else {
+      return -1
+    }
+  }, "stream-source")
+  assert.equal(readyState, await page.evaluate(() => EventSource.CLOSED))
 
   await page.click("#async button")
   await nextBeat()
 
   assert.deepEqual(await messages.allTextContents(), ["First", "Hello world!"])
 })
+
+async function getReadyState(page: Page, id: string): Promise<number> {
+  return page.evaluate((id) => {
+    const element = document.getElementById(id) as StreamSourceElement
+
+    if (element?.streamSource) {
+      return element.streamSource.readyState
+    } else {
+      return -1
+    }
+  }, id)
+}