When you need this. A developer cancels a code completion and immediately asks a new question. If the first request keeps decoding, it still occupies model time and cache memory. Implement an HTTP stream whose lifetime owns the generation iterator.
What you run. Actual local HTTP implementation; controlled socket tests plus a separate real-model request record. No reverse proxy was deployed.
This lesson builds on trace a real inference request from prompt to first token.
Own the stream lifetime
Server-sent events (SSE) frame text events with a blank line. JSON encoding protects embedded newlines from becoming accidental event boundaries. A stream is not one JSON document: the client reads complete events as they arrive.
The server acquires a bounded execution slot before starting inference. It writes and flushes generated text, catches a failed socket write, closes the generation iterator, and releases the slot in finally. Closing the iterator reaches the model adapter’s cleanup path. Cancellation is cooperative: work already submitted to the device may finish.
server.py is an actual loopback HTTP server backed by the same local decoder as the first lesson. test_server.py exercises real sockets with a controlled text fixture, including a client disconnect. The fixture tests transport and cleanup; the separate HTTP run sends requests to the real model.
Run the HTTP stream
Download the lab from this page, extract it, and run these commands from its directory. The default command reads saved run records or runs the explicitly described local algorithm; it never downloads a model or calls a paid API.
python3 -m unittest -v test_server.pyRead server.py
"""Bounded loopback SSE server with explicit cancellation and a real local-model adapter."""
from __future__ import annotations
import argparse
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
def handler_for(backend, capacity=1):
slots = threading.BoundedSemaphore(capacity)
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, *args):
pass
def do_POST(self):
if self.path != "/generate":
self.send_error(404)
return
try:
size = int(self.headers.get("Content-Length", "0"))
if not 0 < size <= 16384:
raise ValueError("Expected a bounded JSON body")
body = json.loads(self.rfile.read(size))
prompt = body["prompt"]
limit = body.get("max_new_tokens", 24)
if not isinstance(prompt, str) or not prompt.strip():
raise ValueError("Prompt must not be empty")
if not isinstance(limit, int) or not 1 <= limit <= 256:
raise ValueError("Output limit must be between 1 and 256")
except (ValueError, KeyError, TypeError):
self.send_error(400)
return
if not slots.acquire(blocking=False):
self.send_error(429)
return
stream = None
try:
self.send_response(200)
self.send_header("Content-Type", "text/event-stream; charset=utf-8")
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "close")
self.end_headers()
stream = backend.stream(prompt, limit)
for token in stream:
frame = (
"data: "
+ json.dumps({"token": token}, ensure_ascii=False)
+ "\n\n"
)
self.wfile.write(frame.encode())
self.wfile.flush()
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
# Detects failed writes. A silent peer may require socket timeout/heartbeat in a deployed adapter.
pass
finally:
if stream is not None:
stream.close()
slots.release()
self.close_connection = True
return Handler
class ModelBackend:
def __init__(self, model_path, device):
from live_inference import LocalDecoder
self.decoder = LocalDecoder(model_path, device)
def stream(self, prompt, limit):
d = self.decoder
cache = None
result = None
try:
ids = d.encode(prompt)
result = d._forward(ids)
cache = result.past_key_values
eos = d.model.generation_config.eos_token_id
eos = set(eos if isinstance(eos, list) else [eos])
emitted = []
previous = ""
for step in range(limit):
token = result.logits[:, -1].argmax(-1, keepdim=True)
d.sync()
token_id = int(token.item())
emitted.append(token_id)
text = d.tokenizer.decode(emitted, skip_special_tokens=True)
if not text.endswith("\ufffd") and text.startswith(previous):
piece = text[len(previous) :]
previous = text
if piece:
yield piece
if token_id in eos or step + 1 == limit:
break
result = d._forward(token, cache)
cache = result.past_key_values
finally:
del cache, result
d.sync()
def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--model-path", required=True)
p.add_argument("--device", choices=["cpu", "mps"], default="cpu")
p.add_argument("--port", type=int, default=8766)
a = p.parse_args()
server = ThreadingHTTPServer(
("127.0.0.1", a.port), handler_for(ModelBackend(a.model_path, a.device))
)
server.serve_forever()
if __name__ == "__main__":
main()This result checks real HTTP framing and iterator cleanup using controlled text. The newline remains inside the JSON string, and the normal stream ends with its application-level completion marker. The socket tests separately close a client before the output cap.
| Quantity | Recorded value |
|---|---|
| HTTP status | 200 |
| Content events | 2 |
| Backend steps | 2 |
| Backend released | true |
Inspect the complete local result
{
"execution_mode": "actual loopback HTTP; controlled text fixture, no model inference",
"http_status": 200,
"content_type": "text/event-stream; charset=utf-8",
"content_events": 2,
"done_marker": "data: [DONE]",
"first_payload": {
"token": "line one\nline two"
},
"backend_steps": 2,
"backend_released": true
}
This image is an application-generated report of the saved program result, not a new inference run. Open the full-resolution report or read the raw result.
Bound latency and cancellation work
A synchronous model call per text event is simple to inspect but limits concurrency. This example admits one active model request and returns HTTP 429 for excess work. A production scheduler would multiplex admitted requests and propagate cancellation into its engine. A write failure detects many disconnects, but a silent stalled peer also needs deadlines and heartbeats.
Disconnect cleanup. A client closes its socket during generation; the handler closes the backend iterator and releases the bounded slot before the output cap.
Empty prompt admission. An empty prompt returns HTTP 400 before the backend starts. A second active request returns 429 while the single model slot is occupied.
Read the executable checks
"""Real loopback HTTP tests. Backend pieces are a controlled transport fixture, not inference."""
import contextlib
import json
import socket
import threading
import time
import unittest
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from server import handler_for
class FixtureBackend:
def __init__(self):
self.released = threading.Event()
self.count = 0
def stream(self, prompt, limit):
try:
for _ in range(limit):
self.count += 1
yield "line one\nline two"
time.sleep(0.005)
finally:
self.released.set()
@contextlib.contextmanager
def serving(backend):
server = ThreadingHTTPServer(("127.0.0.1", 0), handler_for(backend))
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
yield server.server_address
finally:
server.shutdown()
server.server_close()
thread.join()
class StreamingTests(unittest.TestCase):
def test_sse_frames_escape_newlines(self):
backend = FixtureBackend()
with serving(backend) as (host, port):
req = urllib.request.Request(
f"http://{host}:{port}/generate",
data=json.dumps(
{"prompt": "Explain caching", "max_new_tokens": 2}
).encode(),
)
with urllib.request.urlopen(req) as response:
body = response.read().decode()
frames = body.strip().split("\n\n")
self.assertEqual(len(frames), 3)
self.assertEqual(json.loads(frames[0][6:])["token"], "line one\nline two")
self.assertEqual(frames[-1], "data: [DONE]")
self.assertTrue(backend.released.is_set())
def test_disconnect_closes_backend_before_output_cap(self):
backend = FixtureBackend()
with serving(backend) as (host, port):
connection = socket.create_connection((host, port))
body = json.dumps(
{"prompt": "Explain caches", "max_new_tokens": 128}
).encode()
connection.sendall(
f"POST /generate HTTP/1.1\r\nHost: localhost\r\nContent-Length: {len(body)}\r\n\r\n".encode()
+ body
)
received = b""
while b"data:" not in received:
received += connection.recv(4096)
connection.shutdown(socket.SHUT_RDWR)
connection.close()
self.assertTrue(backend.released.wait(2))
self.assertLess(backend.count, 128)
def test_empty_prompt_is_rejected_before_backend_starts(self):
backend = FixtureBackend()
with serving(backend) as (host, port):
req = urllib.request.Request(
f"http://{host}:{port}/generate", data=b'{"prompt":""}'
)
with self.assertRaises(urllib.error.HTTPError) as caught:
urllib.request.urlopen(req)
self.assertEqual(caught.exception.code, 400)
self.assertEqual(backend.count, 0)
if __name__ == "__main__":
unittest.main()
class AdmissionTests(unittest.TestCase):
def test_second_active_request_is_rejected(self):
backend = FixtureBackend()
with serving(backend) as (host, port):
url = f"http://{host}:{port}/generate"
request = lambda: urllib.request.Request(
url, data=b'{"prompt":"Explain caching","max_new_tokens":128}'
)
first = urllib.request.urlopen(request())
try:
self.assertTrue(first.readline().startswith(b"data:"))
with self.assertRaises(urllib.error.HTTPError) as caught:
urllib.request.urlopen(request())
self.assertEqual(caught.exception.code, 429)
finally:
first.close()
self.assertTrue(backend.released.wait(2))Run this optional command only for the actual local endpoint or backend shown. Model paths refer to the first lesson’s explicit download; the server binds to loopback.
python3 -m venv .venv
.venv/bin/python -m pip install torch==2.14.1 transformers==5.20.2
.venv/bin/python server.py --model-path /path/to/downloaded/model --device mps --port 8766Try a change and explain the result
Run the socket tests, then start the server and close a client mid-stream. Observe that cleanup happens before the output cap. Explain why closing a browser connection alone does not prove that an inference engine released its KV cache.
Read the explanation
The stream lifetime owns the generation iterator and its cache. A socket failure must reach cleanup.
Handle real transport limits
Use a bounded request body, output limit, request timeout, and cancellation path. Test fragmented events and multiline text. Keep authentication, TLS, proxy buffering, and engine-specific abort calls as explicit deployment work; this loopback server does not implement them.
References
Sources checked October 24, 2026. Runtime-specific commands and results apply to the recorded versions; linked rolling documentation can change. The application scenarios are teaching examples, not claims about an author’s production incident.