@@ -67,8 +67,10 @@ own: none is a value a placement resolved.
 | reading one request, head and body together (`REQUEST_READ_SECONDS`) | 10 s wall clock, from its first byte | `408` once the body has begun; a head that has not arrived is dropped unanswered |
 | the request line and headers (`MAX_HEAD_BYTES`) | 32 KiB | `431` |
 | requests being read at once (`READ_MEMORY_BYTES` ÷ (32 KiB + `max-input-bytes`), at most 1,024) | 256 MiB of heads and bodies: 248 readers at the default 1 MiB input | `503`, before any of the request is read |
+| one request's body, whatever the declaration (`READ_MEMORY_BYTES` − 32 KiB) | just under 256 MiB | `413`: a declared `max-input-bytes` larger than the whole read budget is held to the budget |
+| documents being decoded | one at a time | waits its turn, still holding its reader |
 | calls whose handler is running (the declared `concurrency`) | `HARNESS_CONCURRENCY`, 32, when the declaration states 0 | `503` |
-| open connections (readers + concurrency + `IDLE_CONNECTIONS`, 64) | — | the connection idle the longest is closed to make room; a new one is never refused |
+| open connections (readers + concurrency + `IDLE_CONNECTIONS`, 64) | — | the connection idle the longest is closed to make room; a new one is never refused, and a request that begins on a connection in the instant it is closed is refused unread rather than run with nowhere to answer |
 | an idle kept-alive connection (`IDLE_SECONDS`) | 120 s, advertised as `Keep-Alive: timeout=120` | closed |
 | one write of an answer (`WRITE_STALL_SECONDS`) | 10 s | the connection is dropped |
 
@@ -81,6 +83,17 @@ call. A slot is held until the handler really returns, not until the call is
 answered: a call answered `504` at its deadline keeps its slot while its handler
 runs on, so the declared concurrency bounds author code running.
 
+A request keeps its reader until its body has been decoded and checked against
+the input schema: a decoded document can be many times the bytes it came from —
+an array of empty objects is about twenty — so the readers' bound is a bound on
+decoding only while decoding sits inside it. The order is: read the body, take a
+concurrency slot (or answer `503` without decoding anything), decode and check
+the payload one call at a time, release the reader, run the handler. A payload
+the schema refuses is dropped before the next is decoded, so beyond the read
+budget the process holds one document being decoded and, under each running
+call's slot, a payload its contract admits — the envelope the declaration's
+`max-input-bytes` and `concurrency` state.
+
 **Concurrency is not yet one number.** Core's declaration says `0` means the
 facility default and defines none, so the harness admits 32 when the declaration
 is silent. The durable-work runtime decides how many calls it dispatches from
@@ -143,7 +156,9 @@ The code becomes a response header, so it must match
 `^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,63}$` — the grammar the durable-work runtime reads
 it with, which is narrower than core's 1 to 128 characters. A code outside it
 raises `ValueError` where it is constructed, and the handler's call then answers
-as any untyped error does: unproven.
+as any untyped error does: unproven. The code is read-only, and checked again
+where the header is set: a code changed after construction answers `500` with
+no failure code.
 
 Everything else leaves the effect **unproven**, because the harness cannot know
 which side of its effect a handler stopped on. A caller that read a bare status
@@ -156,7 +171,7 @@ as proof would stop looking for a receipt that exists.
 | raised anything else | `500`, no failure code | unproven |
 | was still running at the deadline | `504`, no failure code | unproven |
 | returned an answer the contract refuses | `500`, no failure code | unproven |
-| was never reached (payload, identity, effect id, framing) | `400`/`408`/`411`/`431`, no failure code | unproven |
+| was never reached (payload, identity, effect id, framing) | `400`/`408`/`411`/`413`/`431`, no failure code | unproven |
 | was never reached (over a bound, or stopping) | `503`, no failure code | unproven |
 
 The statuses are for humans and proxies: core classifies on the failure-code
@@ -185,7 +200,9 @@ already open, which answer `503` with `Connection: close` — and gives the call
 in flight `SHUTDOWN_GRACE_SECONDS` (30 s) to finish — including a handler whose
 call was already answered at its deadline. Cutting them off would turn every one
 of them into an unproven effect at once. The harness exits `0` as soon as the
-last handler has returned; an idle connection holds nothing. If the grace ends
+last handler has returned; an idle connection holds nothing. The process ends
+with that code directly rather than through interpreter shutdown, which can
+abort while connection threads are still parked or writing a diagnostic. If the grace ends
 with handlers still running, they are cut off and the harness exits `1`, saying
 how many effects it left unproven.
 
 
@@ -35,6 +35,7 @@
 from .protocol import (
     ADDRESS_VARIABLE,
     FAILURE_CODE_HEADER,
+    FAILURE_CODE_PATTERN,
     HARNESS_CONCURRENCY,
     IDLE_CONNECTIONS,
     IDLE_SECONDS,
@@ -194,8 +195,22 @@ def stop(*_: object) -> None:
             "their effects are unproven",
             file=sys.stderr,
         )
-        return 1
-    return 0
+    _exit(1 if remaining else 0)
+    return 1 if remaining else 0
+
+
+def _exit(code: int) -> None:
+    """End the process with this code, whatever its connection threads are doing.
+
+    Interpreter finalization with daemon threads still parked on connections —
+    or holding the lock of a stream they are writing a diagnostic to — can
+    abort instead of exiting, and a placement would read a crash where the code
+    says what happened to the calls. Streams get a bounded moment to flush.
+    """
+    flushing = threading.Thread(target=lambda: (sys.stdout.flush(), sys.stderr.flush()), daemon=True)
+    flushing.start()
+    flushing.join(2.0)
+    os._exit(code)
 
 
 class _Input(socket.SocketIO):
@@ -269,6 +284,11 @@ def __init__(self, host: str, port: int, handler, *, readers: int, connections:
         super().__init__((host, port), handler)
 
     def process_request(self, request, client_address) -> None:
+        self.register(request)
+        super().process_request(request, client_address)
+
+    def register(self, request) -> None:
+        """Count a new connection in, idle, closing the one idle the longest if it is at the bound."""
         with self._lock:
             if len(self._open) >= self._connections:
                 oldest = next((connection for connection, idle in self._open.items() if idle), None)
@@ -279,14 +299,26 @@ def process_request(self, request, client_address) -> None:
                     except OSError:
                         pass
             self._open[request] = True
-        super().process_request(request, client_address)
 
-    def idle(self, request, idle: bool) -> None:
-        """Mark a connection idle (moved to the back of the eviction order) or busy."""
+    def idle(self, request) -> None:
+        """Mark a connection idle again, at the back of the eviction order."""
         with self._lock:
             if request in self._open:
                 del self._open[request]
-                self._open[request] = idle
+                self._open[request] = True
+
+    def claim(self, request) -> bool:
+        """Mark a connection busy, unless it was already closed to make room.
+
+        A connection is closed to make room only while it is idle, so a
+        request that began on it in that same instant is refused before it
+        is read, rather than run with no connection to answer on.
+        """
+        with self._lock:
+            if request not in self._open:
+                return False
+            self._open[request] = False
+            return True
 
     def shutdown_request(self, request) -> None:
         with self._lock:
@@ -296,6 +328,11 @@ def shutdown_request(self, request) -> None:
 
 def _server(address: str, contract: Contract, calls: _Calls, handle, receipt_of) -> ThreadingHTTPServer:
     host, port = _listen(address)
+    decoding = threading.Lock()
+    readers = max(1, min(MAX_READERS, READ_MEMORY_BYTES // (MAX_HEAD_BYTES + contract.max_input_bytes)))
+    # The largest body one reader may hold. A declaration whose input bound is
+    # larger than the whole budget is held to the budget instead.
+    readable = READ_MEMORY_BYTES - MAX_HEAD_BYTES
 
     class Handler(BaseHTTPRequestHandler):
         protocol_version = "HTTP/1.1"
@@ -313,7 +350,6 @@ def setup(self) -> None:
 
         def handle_one_request(self) -> None:
             """Read one request under the harness's bounds, then serve it."""
-            self.server.idle(self.connection, True)
             self._input.deadline = None
             try:
                 begun = self._body.peek(1)
@@ -324,7 +360,9 @@ def handle_one_request(self) -> None:
                 # connection was closed to make room for a new one.
                 self.close_connection = True
                 return
-            self.server.idle(self.connection, False)
+            if not self.server.claim(self.connection):
+                self.close_connection = True
+                return
             self._prepare()
             if not self.server.readers.acquire(blocking=False):
                 # Refused before a byte of it is read, so no peer can make the
@@ -342,6 +380,9 @@ def handle_one_request(self) -> None:
             finally:
                 self.rfile = self._body
                 self._done_reading()
+                # Back of the eviction order: the connection idle the longest
+                # is the one whose last request ended first.
+                self.server.idle(self.connection)
 
         def _prepare(self) -> None:
             """Make an answer possible before a request line was parsed."""
@@ -375,19 +416,29 @@ def do_POST(self) -> None:  # noqa: N802 — BaseHTTPRequestHandler's spelling
             if serve is None:
                 self._answer(_Answer(404, code="unimplemented", message=f"{self.path} is not served"))
                 return
-            call = self._call()
-            if isinstance(call, _Answer):
-                self._answer(call)
+            body = self._read()
+            if isinstance(body, _Answer):
+                self._answer(body)
                 return
             # A slot is taken only once the whole request has arrived, so a
-            # request still arriving never holds one.
+            # request still arriving never holds one — and before the body is
+            # decoded, so a call refused for want of one is never decoded.
             slot = calls.admit()
             if isinstance(slot, str):
+                self._done_reading()
                 self._answer(_Answer(503, code="unavailable", message=slot))
                 return
             try:
                 try:
-                    answer = serve(call, slot)
+                    try:
+                        call = self._decode(body)
+                    finally:
+                        # The request stays under the readers' memory bound
+                        # until it is decoded; from here the decoded call is
+                        # held under its concurrency slot instead.
+                        body = None
+                        self._done_reading()
+                    answer = call if isinstance(call, _Answer) else serve(call, slot)
                 except Exception as err:  # noqa: BLE001 — a defect in the framing still answers
                     answer = _Answer(500, code="internal", message=_describe(err))
                 self._answer(answer)
@@ -424,31 +475,55 @@ def _framing(self) -> int | _Answer:
             if declared > contract.max_input_bytes:
                 return _Answer(400, code="invalid_argument", message=(
                     f"input is {declared} bytes, over the declared {contract.max_input_bytes} byte bound"))
+            if declared > readable:
+                return _Answer(413, code="resource_exhausted", message=(
+                    f"input is {declared} bytes, over the {readable} bytes one request may hold in memory"))
             return declared
 
-        def _call(self) -> Call | _Answer:
-            """Read exactly one call off the connection, or refuse it."""
+        def _read(self) -> bytes | _Answer:
+            """Read exactly one call's body off the connection, or refuse it.
+
+            The request keeps its reader until it has been decoded: the bytes
+            read here count against the readers' memory bound.
+            """
+            declared = self._framing()
+            if isinstance(declared, _Answer):
+                return declared
             try:
-                declared = self._framing()
-                if isinstance(declared, _Answer):
-                    return declared
-                try:
-                    body = self._body.read(declared)
-                except TimeoutError:
-                    return _Answer(408, code="deadline_exceeded",
-                                   message=f"the request was not read within {REQUEST_READ_SECONDS:g}s")
-            finally:
-                self._done_reading()
+                body = self._body.read(declared)
+            except TimeoutError:
+                return _Answer(408, code="deadline_exceeded",
+                               message=f"the request was not read within {REQUEST_READ_SECONDS:g}s")
             if len(body) != declared:
                 return _Answer(400, code="invalid_argument", message="the body ended before its Content-Length")
-            try:
-                return read_call(self.headers, body, contract)
-            except ProtocolError as err:
-                return _Answer(400, code="invalid_argument", message=str(err))
+            return body
+
+        def _decode(self, body: bytes) -> Call | _Answer:
+            """Decode one call and check it against the input schema, one call at a time.
+
+            A decoded document can be many times the bytes it came from — an
+            array of empty objects is about twenty — so concurrent decoding
+            would put memory beyond any bound on what is read. A document the
+            schema refuses is dropped before the next is decoded, so only a
+            call the contract admits outlives this, under its concurrency
+            slot. Decoding and checking are CPU-bound under the interpreter
+            lock, so doing them one call at a time costs no throughput.
+            """
+            with decoding:
+                try:
+                    call = read_call(self.headers, body, contract)
+                    contract.input.validate(call.payload, "input")
+                except (ProtocolError, SchemaError) as err:
+                    return _Answer(400, code="invalid_argument", message=str(err))
+                return call
 
         def _answer(self, answer: _Answer) -> None:
             # Anything but a success may have left request bytes unread, and a
             # stopping harness keeps no connection for a call it will refuse.
+            if answer.failure is not None and not _asserts(answer.failure):
+                # Checked again where it is sent: a code changed after it was
+                # constructed asserts nothing, and never reaches a header.
+                answer = _Answer(500, code="internal", message="the handler's failure code changed after it was checked")
             close = answer.status != 200 or calls.stopping
             self.connection.settimeout(WRITE_STALL_SECONDS)
             if answer.failure is not None:
@@ -481,10 +556,14 @@ def log_request(self, *_: object) -> None:
         def log_message(self, *_: object) -> None:
             """Diagnostics are what the harness prints deliberately, not a request log."""
 
-    readers = max(1, min(MAX_READERS, READ_MEMORY_BYTES // (MAX_HEAD_BYTES + contract.max_input_bytes)))
     return _Server(host, port, Handler, readers=readers, connections=readers + calls.limit + IDLE_CONNECTIONS)
 
 
+def _asserts(failure: HandlerFailure) -> bool:
+    """Whether a failure's code is still one the grammar admits."""
+    return isinstance(failure.code, str) and FAILURE_CODE_PATTERN.fullmatch(failure.code) is not None
+
+
 def _listen(address: str) -> tuple[str, int]:
     """Split the allocated host:port.
 
@@ -499,12 +578,7 @@ def _listen(address: str) -> tuple[str, int]:
 
 
 def _invoke(contract: Contract, handle, call: Call, slot: _Slot) -> _Answer:
-    """Answer one call: validate the payload, run the handler, validate what it returned."""
-    try:
-        contract.input.validate(call.payload, "input")
-    except SchemaError as err:
-        return _Answer(400, code="invalid_argument", message=str(err))
-
+    """Answer one call whose payload the contract admits: run the handler, validate what it returned."""
     outcome = _run(handle, _context(contract, call), call, slot)
     if outcome.failure is not None or outcome.status != 200:
         return outcome
@@ -519,16 +593,11 @@ def _invoke(contract: Contract, handle, call: Call, slot: _Slot) -> _Answer:
 
 
 def _lookup(contract: Contract, receipt_of, call: Call, slot: _Slot) -> _Answer:
-    """Answer a recovery: what did this effect commit?
+    """Answer a recovery whose payload the contract admits: what did this effect commit?
 
     Absent answers 404 and is never an error — absent is "not yet known", and a
     caller reads it as inconclusive rather than as permission to invoke again.
     """
-    try:
-        contract.input.validate(call.payload, "input")
-    except SchemaError as err:
-        return _Answer(400, code="invalid_argument", message=str(err))
-
     outcome = _run(receipt_of, _context(contract, call), call, slot)
     if outcome.failure is not None or outcome.status != 200:
         return outcome
@@ -565,7 +634,12 @@ def invoke() -> None:
 
     worker = threading.Thread(target=invoke, daemon=True)
     slot.hold()
-    worker.start()
+    try:
+        worker.start()
+    except BaseException:
+        # The thread that would have released this hold never ran.
+        slot.release()
+        raise
     worker.join(call.budget)
     if worker.is_alive():
         return _Answer(504, code="deadline_exceeded", message="the deadline passed")