diff --git a/ext/iodine/fio.h b/ext/iodine/fio.h
index 42daf96c..3ca1746e 100644
--- a/ext/iodine/fio.h
+++ b/ext/iodine/fio.h
@@ -457,7 +457,7 @@ Logging and testing helpers
 #define FIO_LOG____LENGTH_BORDER FIO_LOG_LENGTH_LIMIT
 #endif
 /** The logging level */
-int __attribute__((weak)) FIO_LOG_LEVEL;
+extern int FIO_LOG_LEVEL;
 
 #pragma weak FIO_LOG2STDERR
 void __attribute__((format(printf, 1, 0), weak))
diff --git a/ext/iodine/http.c b/ext/iodine/http.c
index 64257c02..dfc6c5f1 100644
--- a/ext/iodine/http.c
+++ b/ext/iodine/http.c
@@ -91,6 +91,13 @@ static inline void remove_content_length(http_s *r) {
   fiobj_hash_delete2(r->private_data.out_headers, cl_hash);
 }
 
+static inline void remove_transfer_encoding(http_s *r) {
+  static uint64_t te_hash = 0;
+  if (!te_hash)
+    te_hash = fiobj_hash_string("transfer-encoding", 17);
+  fiobj_hash_delete2(r->private_data.out_headers, te_hash);
+}
+
 static inline void add_content_type(http_s *r) {
   static uint64_t ct_hash = 0;
   if (!ct_hash)
@@ -378,6 +385,64 @@ intptr_t http_uuid(http_s *h) {
   return ((http_fio_protocol_s *)h->private_data.flag)->uuid;
 }
 
+/**
+ * Marks the response as a streaming response: the connection's protocol will
+ * not auto-finalize it when the request callback returns, and the `http_s`
+ * handle remains valid until `http_streaming_end` completes the response.
+ */
+void http_streaming_start(http_s *h) {
+  if (HTTP_INVALID_HANDLE(h))
+    return;
+  remove_content_length(h);
+  remove_transfer_encoding(h);
+  ((http_vtable_s *)h->private_data.vtbl)->http_streaming_start(h);
+}
+
+/**
+ * Completes a streaming response: sends the terminating chunk via
+ * `http_finish` and resumes normal request handling on the connection.
+ *
+ * AFTER THIS FUNCTION IS CALLED, THE `http_s` OBJECT IS NO LONGER VALID.
+ */
+void http_streaming_end(http_s *h) {
+  if (HTTP_INVALID_HANDLE(h))
+    return;
+  ((http_vtable_s *)h->private_data.vtbl)->http_streaming_end(h);
+}
+
+/**
+ * Arms a one-shot wake when the outgoing queue drains or the connection closes.
+ * Re-arm after each blocked write.
+ */
+void http_streaming_arm_wake(http_s *h) {
+  if (HTTP_INVALID_HANDLE(h))
+    return;
+  ((http_vtable_s *)h->private_data.vtbl)->http_streaming_arm_wake(h);
+}
+
+/**
+ * Copies the current streaming response's NUL-terminated wake channel name to
+ * `dest`, which must hold at least `HTTP_WAKE_CHANNEL_MAX` bytes. Returns its
+ * length, or 0 if there's no active streaming response.
+ */
+size_t http_streaming_wake_channel(http_s *h, char dest[HTTP_WAKE_CHANNEL_MAX]) {
+  static const char prefix[] = "iodine:stream:";
+  if (HTTP_INVALID_HANDLE(h) || !dest)
+    return 0;
+
+  http_fio_protocol_s *p = (http_fio_protocol_s *)h->private_data.flag;
+  if (!p->stream_generation)
+    return 0;
+
+  memcpy(dest, prefix, sizeof(prefix) - 1);
+  size_t len = sizeof(prefix) - 1;
+  len += fio_ltoa(dest + len, (int64_t)p->uuid, 16);
+  dest[len++] = ':';
+  len += fio_ltoa(dest + len, (int64_t)p->stream_generation, 16);
+  dest[len] = 0;
+  return len;
+}
+
 /**
  * Sends the response headers and the specified file (the response's body).
  *
diff --git a/ext/iodine/http.h b/ext/iodine/http.h
index 90bc680d..5d6b7642 100644
--- a/ext/iodine/http.h
+++ b/ext/iodine/http.h
@@ -230,6 +230,53 @@ int http_stream(http_s *h, void *data, uintptr_t length);
  */
 intptr_t http_uuid(http_s *h);
 
+/**
+ * Marks the response as a streaming response.
+ *
+ * The connection's protocol will not auto-finalize the response when the
+ * request callback returns, and the `http_s` handle remains valid for
+ * repeated `http_stream` calls until `http_streaming_end` completes the
+ * response.
+ *
+ * Any application-supplied `Content-Length` or `Transfer-Encoding` header is
+ * removed: the streaming transport owns the response framing (it adds
+ * `Transfer-Encoding: chunked` on the first write, or `Content-Length: 0`
+ * when the stream closes without writing).
+ */
+void http_streaming_start(http_s *h);
+
+/**
+ * Completes a streaming response: sends the terminating chunk via
+ * `http_finish` and resumes normal request handling on the connection.
+ *
+ * AFTER THIS FUNCTION IS CALLED, THE `http_s` OBJECT IS NO LONGER VALID.
+ */
+void http_streaming_end(http_s *h);
+
+/**
+ * Arms a one-shot wake for the streaming response.
+ *
+ * The protocol publishes "drain" or "close" to the process-local wake channel
+ * when the socket queue drains or the connection closes. Re-arm after each
+ * blocked write.
+ */
+void http_streaming_arm_wake(http_s *h);
+
+/** Upper bound (including the NUL) for a wake channel name: the
+ * "iodine:stream:" prefix plus two hex numbers of up to 20 characters each. */
+#define HTTP_WAKE_CHANNEL_MAX 64
+
+/**
+ * Copies the current streaming response's NUL-terminated wake channel name to
+ * `dest`, which must hold at least `HTTP_WAKE_CHANNEL_MAX` bytes.
+ *
+ * The process-local name stays the same for this response and changes for later
+ * responses on the same keep-alive connection.
+ *
+ * Returns its length, or 0 if there's no active streaming response.
+ */
+size_t http_streaming_wake_channel(http_s *h, char dest[HTTP_WAKE_CHANNEL_MAX]);
+
 /**
  * Sends the response headers and the specified file (the response's body).
  *
diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c
index b8f5a70f..0e7d1791 100644
--- a/ext/iodine/http1.c
+++ b/ext/iodine/http1.c
@@ -28,6 +28,8 @@ typedef struct http1pr_s {
   uint8_t close;
   uint8_t is_client;
   uint8_t stop;
+  uint8_t streaming;
+  uint8_t stream_wake;
   http_stream_state_e stream_state;
   uint8_t buf[];
 } http1pr_s;
@@ -46,10 +48,18 @@ inline static void h1_reset(http1pr_s *p) { p->header_size = 0; }
 #define http1_pr2handle(pr) (((http1pr_s *)(pr))->request)
 #define handle2pr(h) ((http1pr_s *)h->private_data.flag)
 
+#define HTTP1_STREAM_WAKE_DRAIN "drain"
+#define HTTP1_STREAM_WAKE_CLOSE "close"
+
+static void http1_stream_wake_publish(http1pr_s *p, const char *msg,
+                                      size_t len);
+
 /* cleanup an HTTP/1.1 handler object */
 static inline void http1_after_finish(http_s *h) {
   http1pr_s *p = handle2pr(h);
   p->stop = p->stop & (~1UL);
+  p->streaming = 0;
+  p->stream_wake = 0;
   p->stream_state = HTTP_STREAM_IDLE;
   if (h != &p->request) {
     http_s_destroy(h, 0);
@@ -289,6 +299,40 @@ static int http1_stream(http_s *h, void *data, uintptr_t length) {
   return 0;
 }
 
+/** Marks the in-progress response as streaming: the parser must not
+ * auto-finish it and must not parse further pipelined requests until the
+ * stream completes. */
+static void http1_streaming_start(http_s *h) {
+  http1pr_s *p = handle2pr(h);
+  if (!p->streaming) {
+    if (++p->p.stream_generation == 0)
+      p->p.stream_generation = 1; /* zero means no active stream */
+  }
+  p->streaming = 1;
+  p->stream_wake = 0;
+}
+
+/* Arm one wake notification for the next drain or disconnect. */
+static void http1_streaming_arm_wake(http_s *h) {
+  handle2pr(h)->stream_wake = 1;
+}
+
+/** Completes a streaming response: sends the terminating chunk via
+ * `http_finish`, then re-arms the parser for any buffered pipelined data
+ * that was suspended while the stream was active. */
+static void http1_streaming_end(http_s *h) {
+  http1pr_s *p = handle2pr(h);
+  const intptr_t uuid = p->p.uuid;
+  /* Explicit close wakes a blocked producer before http_finish resets it. */
+  if (p->streaming && p->stream_wake) {
+    p->stream_wake = 0;
+    http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_CLOSE, sizeof(HTTP1_STREAM_WAKE_CLOSE) - 1);
+  }
+  p->streaming = 0;
+  http_finish(h);
+  fio_force_event(uuid, FIO_EVENT_ON_DATA);
+}
+
 /** Push for data - unsupported. */
 static int http1_push_data(http_s *h, void *data, uintptr_t length,
                            FIOBJ mime_type) {
@@ -319,7 +363,8 @@ static void http1_on_pause(http_s *h, http_fio_protocol_s *pr) {
  * called after the resume task had completed.
  */
 static void http1_on_resume(http_s *h, http_fio_protocol_s *pr) {
-  if (!((http1pr_s *)pr)->stop) {
+  http1pr_s *p = (http1pr_s *)pr;
+  if (!p->stop || p->streaming) {
     fio_resume(pr->uuid);
   }
   (void)h;
@@ -595,6 +640,9 @@ struct http_vtable_s HTTP1_VTABLE = {
     .http_send_body = http1_send_body,
     .http_sendfile = http1_sendfile,
     .http_stream = http1_stream,
+    .http_streaming_start = http1_streaming_start,
+    .http_streaming_end = http1_streaming_end,
+    .http_streaming_arm_wake = http1_streaming_arm_wake,
     .http_finish = htt1p_finish,
     .http_push_data = http1_push_data,
     .http_push_file = http1_push_file,
@@ -617,7 +665,7 @@ Parser Callbacks
 static int http1_on_request(http1_parser_s *parser) {
   http1pr_s *p = parser2http(parser);
   http_on_request_handler______internal(&http1_pr2handle(p), p->p.settings);
-  if (p->request.method && !p->stop)
+  if (p->request.method && !p->stop && !p->streaming)
     http_finish(&p->request);
   h1_reset(p);
   return fio_is_closed(p->p.uuid);
@@ -626,7 +674,7 @@ static int http1_on_request(http1_parser_s *parser) {
 static int http1_on_response(http1_parser_s *parser) {
   http1pr_s *p = parser2http(parser);
   http_on_response_handler______internal(&http1_pr2handle(p), p->p.settings);
-  if (p->request.status_str && !p->stop)
+  if (p->request.status_str && !p->stop && !p->streaming)
     http_finish(&p->request);
   h1_reset(p);
   return fio_is_closed(p->p.uuid);
@@ -755,7 +803,7 @@ static inline void http1_consume_data(intptr_t uuid, http1pr_s *p) {
     i = http1_parse(&p->parser, p->buf + (org_len - p->buf_len), p->buf_len);
     p->buf_len -= i;
     --pipeline_limit;
-  } while (i && p->buf_len && pipeline_limit && !p->stop);
+  } while (i && p->buf_len && pipeline_limit && !p->stop && !p->streaming);
 
   if (p->buf_len && org_len != p->buf_len) {
     memmove(p->buf, p->buf + (org_len - p->buf_len), p->buf_len);
@@ -787,7 +835,7 @@ static inline void http1_consume_data(intptr_t uuid, http1pr_s *p) {
 /** called when a data is available, but will not run concurrently */
 static void http1_on_data(intptr_t uuid, fio_protocol_s *protocol) {
   http1pr_s *p = (http1pr_s *)protocol;
-  if (p->stop) {
+  if (p->stop || p->streaming) {
     fio_suspend(uuid);
     return;
   }
@@ -801,13 +849,31 @@ static void http1_on_data(intptr_t uuid, fio_protocol_s *protocol) {
   http1_consume_data(uuid, p);
 }
 
+/* Notify a blocked producer without calling Ruby here. */
+static void http1_stream_wake_publish(http1pr_s *p, const char *msg,
+                                      size_t len) {
+  char channel[HTTP_WAKE_CHANNEL_MAX];
+  size_t channel_len = http_streaming_wake_channel(&p->request, channel);
+  if (!channel_len)
+    return;
+  fio_publish(.engine = FIO_PUBSUB_PROCESS,
+              .channel = {.len = channel_len, .data = channel},
+              .message = {.len = len, .data = (char *)msg});
+}
+
 /** called when the connection was closed, but will not run concurrently */
 static void http1_on_close(intptr_t uuid, fio_protocol_s *protocol) {
+  http1pr_s *p = (http1pr_s *)protocol;
+  /* Wake blocked producers on disconnect. The protocol keeps the original
+   * UUID; the callback UUID may be newer. The generation separates streams. */
+  if (p->streaming && p->stream_wake) {
+    p->stream_wake = 0;
+    http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_CLOSE, sizeof(HTTP1_STREAM_WAKE_CLOSE) - 1);
+  }
   http1_destroy(protocol);
-  (void)uuid;
 }
 
-/** called when the connection was closed, but will not run concurrently */
+/** called when all pending socket data was sent (the queue drained) */
 static void http1_on_ready(intptr_t uuid, fio_protocol_s *protocol) {
   /* resume slow clients from suspension */
   http1pr_s *p = (http1pr_s *)protocol;
@@ -815,7 +881,11 @@ static void http1_on_ready(intptr_t uuid, fio_protocol_s *protocol) {
     p->stop ^= 4; /* flip back the bit, so it's zero */
     fio_force_event(uuid, FIO_EVENT_ON_DATA);
   }
-  (void)protocol;
+  /* Wake a blocked producer after the socket queue drains. */
+  if (p->streaming && p->stream_wake) {
+    p->stream_wake = 0;
+    http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_DRAIN, sizeof(HTTP1_STREAM_WAKE_DRAIN) - 1);
+  }
 }
 
 /** called when a data is available for the first time */
diff --git a/ext/iodine/http_internal.h b/ext/iodine/http_internal.h
index 53182977..5d4a2fca 100644
--- a/ext/iodine/http_internal.h
+++ b/ext/iodine/http_internal.h
@@ -46,6 +46,12 @@ struct http_vtable_s {
                              uintptr_t offset);
   /** Should send existing headers and data and prepare for streaming */
   int (*const http_stream)(http_s *h, void *data, uintptr_t length);
+  /** Should mark the response as streaming, preventing auto-finalization */
+  void (*const http_streaming_start)(http_s *h);
+  /** Should complete a streaming response and resume request handling */
+  void (*const http_streaming_end)(http_s *h);
+  /** Arms a one-shot drain or disconnect wake for a blocked stream */
+  void (*const http_streaming_arm_wake)(http_s *h);
   /** Should send existing headers or complete streaming */
   void (*const http_finish)(http_s *h);
   /** Push for data. */
@@ -75,6 +81,7 @@ struct http_fio_protocol_s {
   fio_protocol_s protocol;   /* facil.io protocol */
   intptr_t uuid;             /* socket uuid */
   http_settings_s *settings; /* pointer to HTTP settings */
+  uint64_t stream_generation; /* streaming response generation */
 };
 
 #define http2protocol(h) ((http_fio_protocol_s *)h->private_data.flag)
diff --git a/ext/iodine/iodine_http.c b/ext/iodine/iodine_http.c
index 4686b9c4..e674569c 100644
--- a/ext/iodine/iodine_http.c
+++ b/ext/iodine/iodine_http.c
@@ -589,6 +589,14 @@ static inline int ruby2c_response_send(iodine_http_request_handle_s *handle,
     if (rb_respond_to(body, close_method_id))
       IodineCaller.call(body, close_method_id);
     return 0;
+  } else if (rb_respond_to(body, iodine_call_proc_id)) {
+    // Rage owns producer scheduling. Iodine invokes the callable
+    VALUE stream = IodineRackStream.create(handle->h);
+    if (stream == Qnil)
+      return -1;
+    IodineCaller.call2(body, iodine_call_proc_id, 1, &stream);
+    handle->type = IODINE_HTTP_NONE;
+    return 0;
   }
   return -1;
 }
diff --git a/ext/iodine/iodine_rack_stream.c b/ext/iodine/iodine_rack_stream.c
index 18540a7f..d1b27559 100644
--- a/ext/iodine/iodine_rack_stream.c
+++ b/ext/iodine/iodine_rack_stream.c
@@ -6,21 +6,18 @@ typedef enum {
   IODINE_STREAM_IDLE = 0,     /* created, nothing written yet */
   IODINE_STREAM_HEADERS_SENT, /* first write flushed the response headers */
   IODINE_STREAM_STREAMING,    /* chunks flowing */
-  IODINE_STREAM_BLOCKED,      /* paused on backpressure, fiber yielded */
+  IODINE_STREAM_BLOCKED,      /* caller must wait for readiness and retry */
   IODINE_STREAM_CLOSING,      /* close requested, finishing safely */
   IODINE_STREAM_CLOSED,       /* terminal: completed */
   IODINE_STREAM_ERROR,        /* terminal: write failure / disconnect */
 } iodine_stream_state_e;
 
 typedef struct {
-  /* TODO(phase-3): don't persist across http_pause/http_resume (invalidates h). */
-  http_s *h;
+  http_s *h;                   /* stays valid while the response is in streaming mode */
   intptr_t uuid;               /* socket uuid, for fio_pending / fio_is_valid */
+  char wake_channel[HTTP_WAKE_CHANNEL_MAX]; /* fixed wake channel for this response */
+  size_t wake_channel_len;
   iodine_stream_state_e state;
-  VALUE fiber;                 /* producer fiber; also held as an ivar so the GC marks it */
-  size_t high_watermark;       /* pause threshold */
-  size_t low_watermark;        /* resume threshold */
-  int blocked;                 /* backpressure flag */
   int freed;                   /* terminal guard: teardown runs exactly once */
 } stream_ctx_t;
 
@@ -38,7 +35,6 @@ Core data / helpers
 static VALUE rRackStream;
 
 static ID ctx_var_id;   /* ivar holding the stream_ctx_t pointer */
-static ID fiber_var_id; /* ivar holding the producer fiber */
 static ID iodine_new_func_id;
 
 /* write() return values (cached symbols) */
@@ -64,7 +60,6 @@ static void stream_teardown(VALUE stream) {
   ctx->freed = 1;
   ctx->state = IODINE_STREAM_CLOSED;
   set_ctx(stream, NULL);
-  rb_ivar_set(stream, fiber_var_id, Qnil);
   free(ctx);
 }
 
@@ -103,15 +98,15 @@ static VALUE rack_stream_write(VALUE self, VALUE data) {
     return SYM_error;
   }
 
-  /* 6. backpressure -> yield/retry (write would overflow HARD, or past HIGH)
-   * TODO: register http_pause here and resume from http1_on_ready at LOW. */
+  /* 6. The caller waits and retries this chunk. A blocked write arms one wake
+   * for the next drain or disconnect; another blocked retry re-arms it. */
   if (pending + packets_needed >= IODINE_STREAM_HARD_MAX ||
-      pending >= ctx->high_watermark) {
-    ctx->blocked = 1;
+      pending >= IODINE_STREAM_HIGH_WATERMARK) {
     ctx->state = IODINE_STREAM_BLOCKED;
+    http_streaming_arm_wake(ctx->h);
     return SYM_would_block;
   }
- 
+
   /* 7. send in <= CHUNK_SIZE slices; empty chunk runs once to flush headers. */
   const char *p = RSTRING_PTR(data);
   size_t remaining = RSTRING_LEN(data);
@@ -126,22 +121,34 @@ static VALUE rack_stream_write(VALUE self, VALUE data) {
     remaining -= n;
   } while (remaining);
 
-  if (ctx->state < IODINE_STREAM_STREAMING)
-    ctx->state = IODINE_STREAM_STREAMING; /* first write flushed the headers */
+  /* The first write flushes headers; any successful write clears BLOCKED. */
+  ctx->state = IODINE_STREAM_STREAMING;
   return SYM_ok;
 }
 
+/* Process-local channel for "drain" and "close" after :would_block. Ruby owns
+ * Fiber resumption; a retry re-arms the wake if it still blocks. */
+static VALUE rack_stream_wake_channel(VALUE self) {
+  stream_ctx_t *ctx = get_ctx(self);
+  if (!ctx || ctx->state >= IODINE_STREAM_CLOSED)
+    return Qnil;
+  if (!ctx->wake_channel_len)
+    return Qnil;
+  return rb_str_new(ctx->wake_channel, (long)ctx->wake_channel_len);
+}
+
 /* Closes the stream. Idempotent in every state. Sends the terminating
- * zero-length chunk via http_finish exactly once when the connection is still
- * alive, then frees the context. */
+ * zero-length chunk via http_streaming_end exactly once when the connection is
+ * still alive, then frees the context. */
 static VALUE rack_stream_close(VALUE self) {
   stream_ctx_t *ctx = get_ctx(self);
   if (!ctx || ctx->freed)
     return Qnil; /* already closed -> no-op */
 
-  if (ctx->state < IODINE_STREAM_CLOSING && fio_is_valid(ctx->uuid)) {
+  /* End the stream whenever the connection is still alive */
+  if (fio_is_valid(ctx->uuid)) {
     ctx->state = IODINE_STREAM_CLOSING;
-    http_finish(ctx->h); /* invalidates the http_s handle */
+    http_streaming_end(ctx->h); /* invalidates the http_s handle */
   }
   stream_teardown(self);
   return Qnil;
@@ -159,7 +166,7 @@ static VALUE rack_stream_is_closed(VALUE self) {
 C land API
 ***************************************************************************** */
 
-static VALUE new_rack_stream(http_s *h, VALUE fiber) {
+static VALUE new_rack_stream(http_s *h) {
   stream_ctx_t *ctx = malloc(sizeof(*ctx));
   if (!ctx)
     return Qnil;
@@ -167,22 +174,19 @@ static VALUE new_rack_stream(http_s *h, VALUE fiber) {
       .h = h,
       .uuid = http_uuid(h), /* stable connection id; cached for the write path */
       .state = IODINE_STREAM_IDLE,
-      .fiber = fiber,
-      .high_watermark = IODINE_STREAM_HIGH_WATERMARK,
-      .low_watermark = IODINE_STREAM_LOW_WATERMARK,
-      .blocked = 0,
       .freed = 0,
   };
 
+  /* the response now outlives the request callback; only an explicit close
+   * (http_streaming_end) finishes it. */
+  http_streaming_start(h);
+  ctx->wake_channel_len = http_streaming_wake_channel(h, ctx->wake_channel);
+
   VALUE stream = rb_funcall2(rRackStream, iodine_new_func_id, 0, NULL);
   set_ctx(stream, ctx);
-  /* hold the fiber as an ivar so Ruby's GC keeps it alive while blocked. */
-  rb_ivar_set(stream, fiber_var_id, fiber);
   return stream;
 }
 
-static void close_rack_stream(VALUE stream) { stream_teardown(stream); }
-
 /* *****************************************************************************
 Initialization
 ***************************************************************************** */
@@ -191,7 +195,6 @@ static void init_rack_stream(void) {
   rRackStream = rb_define_class_under(IodineBaseModule, "RackStream", rb_cObject);
 
   ctx_var_id = rb_intern("stream_ctx");
-  fiber_var_id = rb_intern("stream_fiber");
   iodine_new_func_id = rb_intern("new");
 
   SYM_ok = ID2SYM(rb_intern("ok"));
@@ -203,10 +206,10 @@ static void init_rack_stream(void) {
   rb_define_method(rRackStream, "write", rack_stream_write, 1);
   rb_define_method(rRackStream, "close", rack_stream_close, 0);
   rb_define_method(rRackStream, "closed?", rack_stream_is_closed, 0);
+  rb_define_method(rRackStream, "wake_channel", rack_stream_wake_channel, 0);
 }
 
 struct IodineRackStream IodineRackStream = {
     .create = new_rack_stream,
-    .close = close_rack_stream,
     .init = init_rack_stream,
 };
diff --git a/ext/iodine/iodine_rack_stream.h b/ext/iodine/iodine_rack_stream.h
index a66bd005..d652f6ad 100644
--- a/ext/iodine/iodine_rack_stream.h
+++ b/ext/iodine/iodine_rack_stream.h
@@ -6,8 +6,7 @@
 #include "http.h"
 
 extern struct IodineRackStream {
-  VALUE (*create)(http_s *h, VALUE fiber);
-  void (*close)(VALUE stream);
+  VALUE (*create)(http_s *h);
   void (*init)(void);
 
 } IodineRackStream;
diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb
new file mode 100644
index 00000000..43cdc892
--- /dev/null
+++ b/spec/integration/response_streaming_spec.rb
@@ -0,0 +1,263 @@
+require 'spec_helper'
+
+# Functional tests for HTTP response streaming: a real Iodine server runs the
+# `response_streaming` app and the HTTP gem consumes the response incrementally.
+RSpec.describe 'HTTP response streaming', with_app: :response_streaming do
+  let(:expected) { "chunk-0\nchunk-1\nchunk-2\nchunk-3\nchunk-4\n" }
+
+  def consume_body(response)
+    body = +""
+    response.body.each { |fragment| body << fragment }
+    body
+  end
+
+  def backpressure_result(path = '/backpressure-result')
+    consume_body(http_get(path))
+  end
+
+  def wait_for_backpressure(deadline: 10, result_path: '/backpressure-result')
+    start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    loop do
+      result = backpressure_result(result_path)
+      return result if yield(result)
+      if Process.clock_gettime(Process::CLOCK_MONOTONIC) - start > deadline
+        raise "timed out waiting for backpressure state, last: #{result}"
+      end
+      sleep 0.05
+    end
+  end
+
+  it 'responds 200 and reassembles the full streamed body' do
+    response = http_get("/")
+    expect(response.status).to eq(200)
+    expect(consume_body(response)).to eq(expected)
+  end
+
+  it 'invokes the callable body in the current Fiber' do
+    response = http_get("/")
+    expect(consume_body(response)).to eq(expected)
+
+    state = http_get('/stream-state')
+    expect(consume_body(state)).to eq('same_fiber=true')
+  end
+
+  it 'uses chunked transfer encoding, not Content-Length' do
+    response = http_get("/")
+    expect(response.chunked?).to be(true)
+    expect(response.headers).not_to include('Content-Length')
+    expect(consume_body(response)).to eq(expected)
+  end
+
+  it 'delivers chunks incrementally rather than buffering the whole response' do
+    first_seen = {}
+    buf = +""
+
+    response = http_get("/")
+    response.body.each do |fragment|
+      observed_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      buf << fragment
+      5.times do |i|
+        first_seen[i] ||= observed_at if buf.include?("chunk-#{i}\n")
+      end
+    end
+
+    expect(buf).to eq(expected)
+    expect(first_seen.keys.sort).to eq([0, 1, 2, 3, 4])
+    expect(first_seen[4] - first_seen[0]).to be > 0.1
+  end
+
+  it 'strips a conflicting application-supplied Content-Length' do
+    response = http_get('/conflicting-length')
+    expect(response.status).to eq(200)
+    expect(response.chunked?).to be(true)
+    expect(response.headers).not_to include('Content-Length')
+    expect(consume_body(response)).to eq('hello world')
+  end
+
+  it 'strips an application-supplied Transfer-Encoding when the stream closes without writing' do
+    response = http_get('/te-no-write')
+    expect(response.status).to eq(200)
+    expect(response.chunked?).to be(false)
+    expect(response.headers.get('Transfer-Encoding')).to be_empty
+    expect(response.headers['Content-Length']).to eq('0')
+    expect(consume_body(response)).to eq("")
+  end
+
+  it 'strips application-supplied framing headers when an oversized write fails the stream' do
+    response = http_get('/framing-oversized')
+    expect(response.status).to eq(200)
+    expect(response.chunked?).to be(false)
+    expect(response.headers.get('Transfer-Encoding')).to be_empty
+    expect(response.headers['Content-Length']).to eq('0')
+    expect(consume_body(response)).to eq("")
+  end
+
+  it 'sends exactly one Transfer-Encoding: chunked on a successful write' do
+    response = http_get('/te-write')
+    expect(response.status).to eq(200)
+    expect(response.headers.get('Transfer-Encoding')).to eq(['chunked'])
+    expect(response.headers.get('Content-Length')).to be_empty
+    expect(consume_body(response)).to eq('hello')
+  end
+
+  it 'completes a stream closed without writing as a normal empty response' do
+    response = http_get('/no-write')
+    expect(response.status).to eq(200)
+    expect(response.chunked?).to be(false)
+    expect(response.headers['Content-Length']).to eq('0')
+    expect(consume_body(response)).to eq("")
+  end
+
+  it 'starts chunked framing on an explicit empty first write' do
+    response = http_get('/empty-write')
+    expect(response.status).to eq(200)
+    expect(response.chunked?).to be(true)
+    expect(response.headers).not_to include('Content-Length')
+    expect(consume_body(response)).to eq("")
+  end
+
+  it 'completes the response cleanly after an oversized write fails the stream' do
+    response = http_get('/oversized')
+    expect(response.status).to eq(200)
+    expect(consume_body(response)).to eq("")
+
+    result = http_get('/oversized-result')
+    expect(consume_body(result)).to eq('result=error')
+  end
+
+  it 'serves a multi-part each body through the buffered non-streaming path' do
+    response = http_get('/each-body')
+    expect(response.status).to eq(200)
+    expect(response.chunked?).to be(false)
+    expect(response.headers['Content-Length']).to eq('12')
+    expect(consume_body(response)).to eq('each-body-ok')
+  end
+
+  it 'serves a status-only response through the empty non-streaming path' do
+    response = http_get('/status-only')
+    expect(response.status).to eq(204)
+    expect(response.headers.get('Transfer-Encoding')).to be_empty
+    # Iodine's http_finish adds Content-Length: 0 to every header-only
+    # response, including 204. RFC 9110 forbids it on 204; worth raising
+    # upstream separately.
+    expect(response.headers['Content-Length']).to eq('0')
+    expect(response.headers.get('Content-Type')).to be_empty
+    expect(consume_body(response)).to eq("")
+  end
+
+  it 'serves a plain request after a streamed response on the same connection' do
+    http_client.persistent("http://localhost:#{server_port}") do |client|
+      streamed = client.get('/')
+      expect(streamed.status).to eq(200)
+      expect(streamed.chunked?).to be(true)
+      expect(streamed.headers['Connection']).to eq('keep-alive')
+      expect(consume_body(streamed)).to eq(expected)
+
+      plain = client.get('/each-body')
+      expect(plain.status).to eq(200)
+      expect(plain.chunked?).to be(false)
+      expect(plain.headers['Content-Length']).to eq('12')
+      expect(consume_body(plain)).to eq('each-body-ok')
+    end
+  end
+
+  it 'treats close as idempotent and rejects writes after the terminal state' do
+    http_client.persistent("http://localhost:#{server_port}") do |client|
+      response = client.get('/double-close')
+      expect(response.status).to eq(200)
+      expect(response.chunked?).to be(true)
+      expect(consume_body(response)).to eq('payload')
+
+      followup = client.get('/each-body')
+      expect(followup.status).to eq(200)
+      expect(consume_body(followup)).to eq('each-body-ok')
+    end
+
+    result = consume_body(http_get('/double-close-result'))
+    expect(result).to eq(
+      'first_close=nil second_close=nil closed=true ' \
+      'write_after_close=closed wake_channel=nil'
+    )
+  end
+
+  it 'wakes a blocked producer when the outgoing queue drains' do
+    response = http_get('/backpressure')
+
+    wait_for_backpressure { |r| r =~ /would_blocks=[1-9]/ }
+
+    expect(consume_body(response)).to eq('x' * (256 * 16_384))
+
+    result = wait_for_backpressure do |r|
+      r.include?('result=completed') && r.include?('unsubscribed=true')
+    end
+    expect(result).to match(
+      /result=completed sent=256 would_blocks=[1-9]\d* wakes=[1-9]\d* finished=true unsubscribed=true/
+    )
+  end
+
+  it 'wakes a blocked producer when another callback closes the stream' do
+    # The body stays unconsumed until the server reports the close, so the
+    # outgoing queue backs up and the producer blocks on its own.
+    response = http_get('/backpressure-close')
+
+    result = wait_for_backpressure(result_path: '/backpressure-close-result') do |r|
+      r.include?('result=closed') && r.include?('unsubscribed=true')
+    end
+    expect(result).to match(
+      %r{\Aresult=closed\s+sent=\d+\s+would_blocks=1\s+
+         wakes=1\s+last_wake=close\s+close_scheduled=true\s+
+         closed_externally=true\s+finished=true\s+unsubscribed=true\z}x
+    )
+
+    sent = result[/sent=(\d+)/, 1].to_i
+    expect(response.status).to eq(200)
+    expect(consume_body(response)).to eq('x' * (sent * 16_384))
+  end
+
+  it 'wakes a parked producer when the client disconnects mid-stream' do
+    client = http_client
+    client.get("http://localhost:#{server_port}/backpressure")
+
+    wait_for_backpressure { |r| r =~ /would_blocks=[1-9]/ }
+    client.close
+
+    result = wait_for_backpressure do |r|
+      r.include?('result=disconnected') && r.include?('unsubscribed=true')
+    end
+    expect(result).to match(/result=disconnected .*finished=true unsubscribed=true/)
+
+    response = http_get('/te-write')
+    expect(consume_body(response)).to eq('hello')
+  end
+
+  it 'isolates wake subscriptions for sequential streams on one connection' do
+    expected_stream = 'x' * (256 * 16_384)
+
+    http_client.persistent("http://localhost:#{server_port}") do |client|
+      first = client.get('/backpressure')
+      expect(first.status).to eq(200)
+      expect(consume_body(first)).to eq(expected_stream)
+
+      second = client.get('/backpressure')
+      expect(second.status).to eq(200)
+      expect(consume_body(second)).to eq(expected_stream)
+    end
+  end
+
+  it 'keeps streaming after the callable returns' do
+    body = +""
+    released = false
+
+    response = http_get('/async')
+    response.body.each do |fragment|
+      body << fragment
+      next if released || !body.include?("marker-a\n")
+
+      released = true
+      expect(http_get('/release').status).to eq(204)
+    end
+
+    expect(released).to be(true)
+    expect(body).to eq("marker-a\nmarker-b\n")
+  end
+end
diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru
new file mode 100644
index 00000000..812ace8d
--- /dev/null
+++ b/spec/support/apps/response_streaming.ru
@@ -0,0 +1,245 @@
+# Test app for HTTP response streaming.
+# The body responds to `call(stream)` (Rack streaming body), so Iodine should
+# hand it a RackStream writer and stream each chunk incrementally.
+same_fiber = nil
+oversized_result = nil
+release_channel = "response-streaming-release"
+backpressure = nil
+backpressure_close = nil
+double_close = nil
+
+run ->(env) do
+  if env['PATH_INFO'] == '/stream-state'
+    next [200, {}, ["same_fiber=#{same_fiber}"]]
+  end
+
+  if env['PATH_INFO'] == '/release'
+    Iodine.publish(release_channel, '', Iodine::PubSub::PROCESS)
+    next [204, {}, []]
+  end
+
+  if env['PATH_INFO'] == '/conflicting-length'
+    body = proc do |stream|
+      stream.write("hello ")
+      stream.write("world")
+      stream.close
+    end
+
+    next [200, { 'Content-Length' => '999' }, body]
+  end
+
+  if env['PATH_INFO'] == '/no-write'
+    body = proc do |stream|
+      stream.close
+    end
+
+    next [200, { 'Content-Length' => '5' }, body]
+  end
+
+  if env['PATH_INFO'] == '/te-no-write'
+    body = proc do |stream|
+      stream.close
+    end
+
+    next [200, { 'Transfer-Encoding' => 'chunked' }, body]
+  end
+
+  if env['PATH_INFO'] == '/te-write'
+    body = proc do |stream|
+      stream.write("hello")
+      stream.close
+    end
+
+    next [200, { 'Transfer-Encoding' => 'chunked' }, body]
+  end
+
+  if env['PATH_INFO'] == '/framing-oversized'
+    body = proc do |stream|
+      stream.write("x" * (2 * 1024 * 1024))
+      stream.close
+    end
+
+    next [200, { 'Transfer-Encoding' => 'chunked', 'Content-Length' => '999' }, body]
+  end
+
+  if env['PATH_INFO'] == '/empty-write'
+    body = proc do |stream|
+      stream.write("")
+      stream.close
+    end
+
+    next [200, {}, body]
+  end
+
+  if env['PATH_INFO'] == '/oversized'
+    body = proc do |stream|
+      oversized_result = stream.write("x" * (2 * 1024 * 1024))
+      stream.close
+    end
+
+    next [200, {}, body]
+  end
+
+  if env['PATH_INFO'] == '/oversized-result'
+    next [200, {}, ["result=#{oversized_result}"]]
+  end
+
+  # Non-streaming response regressions.
+  if env['PATH_INFO'] == '/each-body'
+    next [200, {}, ['each-', 'body-', 'ok']]
+  end
+
+  if env['PATH_INFO'] == '/status-only'
+    next [204, {}, []]
+  end
+
+  if env['PATH_INFO'] == '/double-close'
+    state = {}
+    double_close = state
+
+    body = proc do |stream|
+      stream.write("payload")
+      state[:first_close] = stream.close.inspect
+      state[:closed_after_first] = stream.closed?
+      state[:second_close] = stream.close.inspect
+      state[:write_after_close] = stream.write("late")
+      state[:wake_channel_after_close] = stream.wake_channel.inspect
+    end
+
+    next [200, {}, body]
+  end
+
+  if env['PATH_INFO'] == '/double-close-result'
+    s = double_close || {}
+    next [200, {}, [
+      "first_close=#{s[:first_close]} second_close=#{s[:second_close]} " \
+      "closed=#{s[:closed_after_first]} write_after_close=#{s[:write_after_close]} " \
+      "wake_channel=#{s[:wake_channel_after_close]}"
+    ]]
+  end
+
+  if ['/backpressure', '/backpressure-close'].include?(env['PATH_INFO'])
+    close_while_blocked = env['PATH_INFO'] == '/backpressure-close'
+    state = {
+      result: nil,
+      sent: 0,
+      would_blocks: 0,
+      wakes: 0,
+      last_wake: nil,
+      finished: false,
+      unsubscribed: false
+    }
+    if close_while_blocked
+      backpressure_close = state
+    else
+      backpressure = state
+    end
+
+    body = proc do |stream|
+      payload = "x" * 16_384
+      total = 256
+
+      producer = Fiber.new do
+        sent = 0
+        while sent < total
+          case (status = stream.write(payload))
+          when :ok
+            sent += 1
+            state[:sent] = sent
+          when :would_block
+            state[:would_blocks] += 1
+            Fiber.yield
+            state[:wakes] += 1
+          else
+            state[:result] = status
+            break
+          end
+        end
+        state[:result] ||= :completed
+        stream.close
+        state[:finished] = true
+      end
+
+      channel = stream.wake_channel
+      Iodine.subscribe(channel) do |_, message|
+        state[:last_wake] = message
+        producer.resume if producer.alive?
+        unless producer.alive?
+          Iodine.defer do
+            removed = Iodine.unsubscribe(channel)
+            state[:unsubscribed] = removed && !Iodine.subscribed?(channel)
+          end
+        end
+      end
+
+      if close_while_blocked
+        state[:close_scheduled] = true
+        Iodine.defer do
+          producer.resume
+          if producer.alive? && state[:would_blocks] > 0
+            state[:closed_externally] = true
+            stream.close
+          elsif !producer.alive?
+            Iodine.defer do
+              removed = Iodine.unsubscribe(channel)
+              state[:unsubscribed] = removed && !Iodine.subscribed?(channel)
+            end
+          end
+        end
+      else
+        producer.resume
+        unless producer.alive?
+          Iodine.defer do
+            removed = Iodine.unsubscribe(channel)
+            state[:unsubscribed] = removed && !Iodine.subscribed?(channel)
+          end
+        end
+      end
+    end
+
+    next [200, {}, body]
+  end
+
+  if env['PATH_INFO'] == '/backpressure-result'
+    s = backpressure || {}
+    next [200, {}, [
+      "result=#{s[:result]} sent=#{s[:sent]} would_blocks=#{s[:would_blocks]} " \
+      "wakes=#{s[:wakes]} finished=#{s[:finished]} unsubscribed=#{s[:unsubscribed]}"
+    ]]
+  end
+
+  if env['PATH_INFO'] == '/backpressure-close-result'
+    s = backpressure_close || {}
+    next [200, {}, [
+      "result=#{s[:result]} sent=#{s[:sent]} would_blocks=#{s[:would_blocks]} " \
+      "wakes=#{s[:wakes]} last_wake=#{s[:last_wake]} " \
+      "close_scheduled=#{s[:close_scheduled]} closed_externally=#{s[:closed_externally]} " \
+      "finished=#{s[:finished]} unsubscribed=#{s[:unsubscribed]}"
+    ]]
+  end
+
+  if env['PATH_INFO'] == '/async'
+    body = proc do |stream|
+      Iodine.subscribe(release_channel) do
+        stream.write("marker-b\n")
+        stream.close
+        Iodine.defer { Iodine.unsubscribe(release_channel) }
+      end
+      stream.write("marker-a\n")
+    end
+
+    next [200, {}, body]
+  end
+
+  request_fiber = Fiber.current
+  body = proc do |stream|
+    same_fiber = Fiber.current.equal?(request_fiber)
+    5.times do |i|
+      stream.write("chunk-#{i}\n")
+      sleep 0.05 # a gap so incremental delivery is observable
+    end
+    stream.close
+  end
+
+  [200, {}, body]
+end
diff --git a/spec/unit/rack_stream_spec.rb b/spec/unit/rack_stream_spec.rb
index 67a8f1a8..50624626 100644
--- a/spec/unit/rack_stream_spec.rb
+++ b/spec/unit/rack_stream_spec.rb
@@ -5,7 +5,9 @@
     expect(defined?(Iodine::Base::RackStream)).to eq('constant')
   end
 
-  it 'exposes the writer API: write, close, closed?' do
-    expect(Iodine::Base::RackStream.instance_methods(false)).to include(:write, :close, :closed?)
+  it 'exposes the writer and wake-channel API' do
+    expect(Iodine::Base::RackStream.instance_methods(false)).to include(
+      :write, :close, :closed?, :wake_channel
+    )
   end
 end