From aaa4acfbb1a87804e6a8c5aef78795b7cc7bb769 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 00:41:38 +0000
Subject: [PATCH 1/5] Prepare protocol read-ahead reservation source controls

---
 .../collector-proposal/README.md              |  63 ++++++--
 .../collector-proposal/driver-sources.json    |   9 +-
 .../collector-proposal/proposal.json          |  30 +++-
 tests/test_worker_startup_buffered.py         | 145 ++++++++++++++++++
 tests/worker_startup_buffered.py              |  59 +++++++
 tests/worker_startup_read_budget.py           |  29 +++-
 6 files changed, 310 insertions(+), 25 deletions(-)
 create mode 100644 tests/test_worker_startup_buffered.py
 create mode 100644 tests/worker_startup_buffered.py

diff --git a/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md b/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
index 30e20d5..3314d5a 100644
--- a/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
+++ b/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
@@ -17,8 +17,8 @@ No #184 executor, acquisition source or blocked review is reused here.
 Use a dedicated observer process, with one connection and no retry, for each
 before/after SQL pair. The proposed driver family is PyMySQL 1.2.3, source commit
 `c990e86083c26493f3285b1e047902994c4b1811`. Its package archive, complete runtime
-import closure, authentication/TLS dependencies and installed input are **not
-selected or acquired**. Public release/source identity is only an input to this
+import closure, actual authentication/TLS inputs and installed input are **not
+bound or acquired**. Public release/source identity is only an input to this
 proposal; it does not establish a supported or qualified historical tuple.
 
 The first pure source slice now lives in `tests/worker_startup_raw.py`.
@@ -37,22 +37,44 @@ failure, and checks the same monotonic deadline before and after each callback.
 Synthetic callbacks exercise these rules without a driver or socket. The
 wrapper cannot interrupt a blocked callback. It is not wired to the pinned
 driver's private method and proves no actual handshake or receive coverage.
-No driver, terminator validation, process owner, framing or cleanup is
-implemented by these slices. The collector vectors below remain unexecuted,
-and all operating inputs remain unselected.
+
+`tests/worker_startup_buffered.py` places the same reservation beneath a stock
+`io.BufferedReader`. Its raw input must be supplied before the first protocol
+read. A logical request is bounded before buffered allocation; every raw
+request, including read-ahead, is reserved before reading. Cached bytes are not
+charged twice, and replacing a stream retains the same phase budget/deadline.
+The inert predecessor control demonstrates that a one-byte logical callback
+can otherwise fetch 8192 uncharged bytes through stock buffering. Successor
+controls cover that reservation, cached reads, shared streams, oversize/expiry
+refusals, short EOF and close without draining. They use fixture raw streams,
+not a socket or the proposed driver.
+
+The actual driver binding is still absent. The pinned connection source creates
+its initial buffered reader and recreates it after TLS wrapping. Both sites
+must install the budgeted raw boundary before any read; this primitive refuses
+an already buffered input and does not establish either lifecycle hook. TLS
+handshake deadlines, authentication, result terminators, process ownership,
+framing and cleanup remain unimplemented. The collector vectors below remain
+unexecuted, and all actual operating inputs remain unbound.
 
 The process receives only the fixture identity and dedicated SELECT-only
 fixture connection parameters through the future owner's private input channel.
 Never obtain production endpoints, ambient credentials, config files, plugin
-credentials or a user's home defaults. Network membership, peer identity,
-authentication, grants and TLS trust must be fixed in the later input packet.
+credentials or a user's home defaults. The [source/input decision](https://github.com/durable-workflow/sdk-python/issues/123#issuecomment-6673104109)
+selects a dedicated SELECT-only fixture identity using `caching_sha2_password`
+over required verified TLS, with peer hostname `mysql` and port 3306. This is
+new reconstruction input, not recovered original authentication provenance.
+Actual network/peer, CA, hostname-valid certificate, grants, backend/platform
+and full installed closure remain unbound. No account/certificate/key creation
+is selected by the source decision.
 No opportunistic RSA package installation, plaintext fallback, server-key fetch,
 reconnect, discovery or endpoint substitution is permitted.
 For the proposed TCP path, supply an explicit SSLContext with the selected CA,
 certificate verification and hostname verification. This pinned driver makes
 explicit SSL configuration required, while its default preferred mode can fall
 back to plaintext. Default TLS options are therefore inadmissible. The exact
-context, peer and authentication closure still require selection and controls.
+context, peer and authentication closure still require concrete binding and
+controls. Ambient CA defaults and authentication plugin switches are refused.
 
 The driver source has consequential defaults:
 
@@ -63,7 +85,7 @@ The driver source has consequential defaults:
   to show autocommit before either SELECT; refuse an inherited false state
   without changing it. No transaction or time-zone change is proposed.
 - Socket read/write timeouts bound socket operations, not the complete pair.
-  Buffered row reading and multipart packets have no proposed receive bound.
+  Stock buffered row reading and multipart packets have no such receive bound.
   `max_allowed_packet` is not a receive-size limit.
 - Default converters produce Python values including datetimes. The proposed
   raw path disables Unicode and type-code decoders, retaining only the driver's
@@ -72,7 +94,8 @@ The driver source has consequential defaults:
 
 These facts are bound to [connection source](https://github.com/PyMySQL/PyMySQL/blob/c990e86083c26493f3285b1e047902994c4b1811/pymysql/connections.py),
 [cursor source](https://github.com/PyMySQL/PyMySQL/blob/c990e86083c26493f3285b1e047902994c4b1811/pymysql/cursors.py)
-and [converter source](https://github.com/PyMySQL/PyMySQL/blob/c990e86083c26493f3285b1e047902994c4b1811/pymysql/converters.py).
+[converter source](https://github.com/PyMySQL/PyMySQL/blob/c990e86083c26493f3285b1e047902994c4b1811/pymysql/converters.py)
+and [authentication source](https://github.com/PyMySQL/PyMySQL/blob/c990e86083c26493f3285b1e047902994c4b1811/pymysql/_auth.py).
 The future adapter is still missing; stock driver use alone satisfies none of
 the complete-read, byte-limit, absolute-deadline or cleanup obligations below.
 
@@ -88,14 +111,20 @@ SQL, quote them locally, log `mogrify`, or allow arbitrary SQL/token replacement
 The driver's parameter API is documented [here](https://pymysql.readthedocs.io/en/latest/modules/cursors.html#pymysql.cursors.Cursor.execute).
 It performs driver escaping, not server-side prepared execution.
 
-The proposed raw reader checks each requested byte count **before** calling the
-pinned driver's `_read_bytes`. It refuses a request above 32 KiB or cumulative
-receive reservation above 64 KiB for the entire connection/pair, including
-handshake, setup, descriptions, rows and terminators. This also bounds multipart
-assembly: header reservations and every component body consume the same budget.
+The proposed reader bounds the driver's logical requests and reserves raw
+protocol reads, including buffered read-ahead, **before** their read/allocation
+boundary. It refuses a request above 32 KiB or cumulative raw reservation above
+64 KiB for the entire connection/pair, including greeting, auth/setup,
+descriptions, rows and terminators. Multipart headers and every component body
+consume the same protocol budget; cached bytes are not counted a second time.
+These are MySQL protocol reservations, not a total TLS ciphertext, certificate
+record or OpenSSL allocation cap. TLS negotiation, buffering and blocked I/O
+still share the original absolute phase/case deadline and resource envelope.
+They receive no separate allowance, and no custom TLS engine is proposed.
 A failed read does not refund that reservation or permit a retry. The adapter
-must be source-reviewed against this exact private method before adoption; it
-has not been implemented or exercised here.
+must bind both initial and post-TLS raw readers against the pinned connection
+source before adoption. That lifecycle binding has not been implemented or
+exercised here.
 Construct with `defer_connect=True`, initialize the deadline/receive guard before
 any handshake, and connect only after fixture identity/input validation. The
 complete process lifetime and its output are also bounded by the owner. Send
diff --git a/tests/fixtures/worker_startup_reconstruction/collector-proposal/driver-sources.json b/tests/fixtures/worker_startup_reconstruction/collector-proposal/driver-sources.json
index 0cbaa37..cadcde5 100644
--- a/tests/fixtures/worker_startup_reconstruction/collector-proposal/driver-sources.json
+++ b/tests/fixtures/worker_startup_reconstruction/collector-proposal/driver-sources.json
@@ -38,6 +38,12 @@
       "blob": "e1afd4f8e5171aa139bc6ec3c41c9a38a7252ba8",
       "bytes": 1492,
       "sha256": "0c3fbc0b53bdaad91e4036c511662d282e7a858192a37e952c0a822a8bf8f7ca"
+    },
+    {
+      "path": "pymysql/_auth.py",
+      "blob": "77123d731334783e31a82ff827d0eb469fbfb8c0",
+      "bytes": 7636,
+      "sha256": "0ac2af5973379f156136aefdac9da0c296e3246860999d3446c46325f337a9dc"
     }
   ],
   "status": "public-source-only-unexecuted-driver-proposal",
@@ -93,5 +99,6 @@
       "bytes": 675,
       "sha256": "025db37ab5311650d694f139941e206c2ae085d9b862426efa409a53ddd2ae13"
     }
-  ]
+  ],
+  "source_input_decision": "https://github.com/durable-workflow/sdk-python/issues/123#issuecomment-6673104109"
 }
diff --git a/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json b/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json
index c1e3fe1..d926d6f 100644
--- a/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json
+++ b/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json
@@ -58,7 +58,19 @@
     "defer_connect": true,
     "guard_initialized_before_connect": true,
     "driver_stdout_stderr": "discard; fixed-code framed output only",
-    "tcp_tls": "explicit SSLContext with selected CA, certificate and hostname verification; default preferred mode refused"
+    "tcp_tls": "explicit SSLContext with selected CA, certificate and hostname verification; default preferred mode refused",
+    "observer_authentication_proposal": {
+      "identity": "dedicated SELECT-only fixture observer",
+      "plugin": "caching_sha2_password",
+      "transport": "required verified TLS",
+      "host_identity": "mysql",
+      "port": 3306,
+      "ca_certificate_hostname_and_grants": null,
+      "original_historical_authentication_provenance": false,
+      "ambient_ca_or_credentials": false,
+      "plugin_switch_or_server_key_fetch": false,
+      "account_or_certificate_creation_selected": false
+    }
   },
   "proposed_bounds": {
     "case_seconds": 30,
@@ -73,7 +85,8 @@
     "observer_within_original_runner_cpu": 1,
     "observer_within_original_runner_memory_mib": 512,
     "independent_observer_container": false,
-    "additional_resources_selected": false
+    "additional_resources_selected": false,
+    "receive_budget_domain": "MySQL protocol bytes, including all raw read-ahead; not total TLS ciphertext or OpenSSL allocation"
   },
   "deadlines": {
     "case": "case_started_monotonic + 30",
@@ -121,7 +134,7 @@
   },
   "implementation_steps": [
     "closed SQL-token adapter and raw tuple normalizer with pure byte fixtures",
-    "pinned private-reader receive guard with inert fragmented-packet controls",
+    "bind initial and post-TLS protocol readers before their first read; retain inert read-ahead controls",
     "absolute-budget and framed-output owner integration with inert blocked/trickling child controls",
     "exact-source/input/owner packet and independent cleanup controls before historical execution"
   ],
@@ -145,5 +158,16 @@
     "containers_started": 0,
     "execution_admitted": false,
     "qualification_result": null
+  },
+  "source_input_decision": "https://github.com/durable-workflow/sdk-python/issues/123#issuecomment-6673104109",
+  "protocol_reader_source": {
+    "primitive": "tests/worker_startup_buffered.py",
+    "buffer": "stock io.BufferedReader over a budgeted raw stream",
+    "logical_request_bound_before_buffer_allocation": true,
+    "raw_reservation_before_read": true,
+    "actual_driver_initial_and_post_tls_assignment_binding": null,
+    "tls_handshake_deadline_enforcement": null,
+    "actual_driver_imported": false,
+    "replacement_streams_share_supplied_phase_budget": true
   }
 }
diff --git a/tests/test_worker_startup_buffered.py b/tests/test_worker_startup_buffered.py
new file mode 100644
index 0000000..b54fd66
--- /dev/null
+++ b/tests/test_worker_startup_buffered.py
@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import io
+
+import pytest
+
+from tests.worker_startup_buffered import ProtocolReader
+from tests.worker_startup_read_budget import ReadRefusal, ReceiveBudget
+
+
+class RecordingRaw(io.RawIOBase):
+    def __init__(self, data: bytes) -> None:
+        super().__init__()
+        self.data = data
+        self.offset = 0
+        self.requests: list[int] = []
+        self.close_count = 0
+
+    def readable(self) -> bool:
+        return True
+
+    def readinto(self, buffer) -> int:
+        self.requests.append(len(buffer))
+        data = self.data[self.offset:self.offset + len(buffer)]
+        buffer[:len(data)] = data
+        self.offset += len(data)
+        return len(data)
+
+    def close(self) -> None:
+        if not self.closed:
+            self.close_count += 1
+        super().close()
+
+
+def test_logical_callback_predecessor_does_not_account_for_stock_read_ahead() -> None:
+    raw = RecordingRaw(b"x" * 8192)
+    buffered = io.BufferedReader(raw, buffer_size=8192)
+    budget = ReceiveBudget(buffered.read, 10, clock=lambda: 1)
+    try:
+        assert budget.read(1) == b"x"
+        assert budget.reserved_bytes == 1
+        assert raw.requests == [8192]
+        assert raw.offset == 8192
+    finally:
+        buffered.close()
+
+
+def test_successor_charges_read_ahead_before_raw_read_and_does_not_double_count_cache() -> None:
+    raw = RecordingRaw(b"x" * 8192)
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
+    reader = ProtocolReader(raw, budget)
+    try:
+        assert raw.requests == []
+        assert reader.read(1) == b"x"
+        assert budget.reserved_bytes == 8192
+        assert raw.requests == [8192]
+        assert reader.read(100) == b"x" * 100
+        assert budget.reserved_bytes == 8192
+        assert raw.requests == [8192]
+    finally:
+        reader.close()
+
+
+def test_logical_oversize_is_rejected_before_buffered_allocation_or_raw_read() -> None:
+    raw = RecordingRaw(b"x" * 65536)
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
+    reader = ProtocolReader(raw, budget)
+    try:
+        with pytest.raises(ReadRefusal, match="RECEIVE_LIMIT_EXCEEDED"):
+            reader.read(32769)
+        assert raw.requests == []
+        assert budget.reserved_bytes == 0
+    finally:
+        reader.close()
+
+
+def test_fresh_protocol_streams_share_the_same_phase_reservation() -> None:
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
+    first_raw = RecordingRaw(b"a" * 32768)
+    second_raw = RecordingRaw(b"b" * 32768)
+    extra_raw = RecordingRaw(b"c" * 8192)
+    readers = [ProtocolReader(raw, budget) for raw in (first_raw, second_raw, extra_raw)]
+    try:
+        assert readers[0].read(32768) == b"a" * 32768
+        assert readers[1].read(32768) == b"b" * 32768
+        assert budget.reserved_bytes == 65536
+        with pytest.raises(ReadRefusal, match="RECEIVE_LIMIT_EXCEEDED"):
+            readers[2].read(1)
+        assert extra_raw.requests == []
+    finally:
+        for reader in readers:
+            reader.close()
+
+
+def test_expiry_prevents_returning_cached_protocol_bytes() -> None:
+    now = [1]
+    raw = RecordingRaw(b"x" * 8192)
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: now[0])
+    reader = ProtocolReader(raw, budget)
+    try:
+        assert reader.read(1) == b"x"
+        now[0] = 10
+        with pytest.raises(ReadRefusal, match="TIME_BUDGET_EXPIRED"):
+            reader.read(1)
+        assert raw.requests == [8192]
+        assert budget.reserved_bytes == 8192
+    finally:
+        reader.close()
+
+
+def test_short_eof_refuses_and_cannot_be_retried_or_refunded() -> None:
+    raw = RecordingRaw(b"short")
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
+    reader = ProtocolReader(raw, budget)
+    try:
+        with pytest.raises(ReadRefusal, match="RAW_READ_INCOMPLETE"):
+            reader.read(6)
+        before = list(raw.requests)
+        reserved = budget.reserved_bytes
+        with pytest.raises(ReadRefusal, match="READ_ALREADY_REFUSED"):
+            reader.read(1)
+        assert raw.requests == before
+        assert budget.reserved_bytes == reserved
+    finally:
+        reader.close()
+
+
+def test_close_is_idempotent_and_does_not_drain_protocol_data() -> None:
+    raw = RecordingRaw(b"unconsumed")
+    reader = ProtocolReader(raw, ReceiveBudget(lambda _: b"", 10, clock=lambda: 1))
+    reader.close()
+    reader.close()
+    assert raw.close_count == 1
+    assert raw.requests == []
+
+
+def test_an_existing_buffered_reader_is_not_admitted_as_a_raw_source() -> None:
+    raw = RecordingRaw(b"x")
+    buffered = io.BufferedReader(raw)
+    try:
+        with pytest.raises(TypeError, match="unbuffered raw stream"):
+            ProtocolReader(buffered, ReceiveBudget(lambda _: b"", 10, clock=lambda: 1))
+        assert raw.requests == []
+    finally:
+        buffered.close()
diff --git a/tests/worker_startup_buffered.py b/tests/worker_startup_buffered.py
new file mode 100644
index 0000000..7acc823
--- /dev/null
+++ b/tests/worker_startup_buffered.py
@@ -0,0 +1,59 @@
+"""Proposed protocol-reader primitive using stock buffered I/O and inert inputs.
+
+The raw input must be supplied before any protocol read. Actual pinned-driver
+initial/post-TLS binding, TLS/auth inputs and process cleanup remain unqualified.
+"""
+
+from __future__ import annotations
+
+import io
+from typing import Any
+
+from tests.worker_startup_read_budget import ReceiveBudget
+
+
+class _ReservedRaw(io.RawIOBase):
+    def __init__(self, raw: io.RawIOBase, budget: ReceiveBudget) -> None:
+        super().__init__()
+        self._raw = raw
+        self._budget = budget
+
+    def readable(self) -> bool:
+        return True
+
+    def readinto(self, buffer: Any) -> int:
+        view = memoryview(buffer).cast("B")
+        if view.readonly:
+            raise ValueError("Protocol read buffer must be writable")
+        value = self._budget.read_some(len(view), self._raw.read)
+        view[:len(value)] = value
+        return len(value)
+
+    def close(self) -> None:
+        if self.closed:
+            return
+        try:
+            self._raw.close()
+        finally:
+            super().close()
+
+
+class ProtocolReader:
+    """Charge all raw protocol read-ahead once, with an outer allocation bound."""
+
+    def __init__(self, raw: io.RawIOBase, budget: ReceiveBudget) -> None:
+        if not isinstance(raw, io.RawIOBase):
+            raise TypeError("Protocol source must be an unbuffered raw stream")
+        self._budget = budget
+        self._buffer = io.BufferedReader(_ReservedRaw(raw, budget), buffer_size=8192)
+
+    def read(self, count: int) -> bytes:
+        self._budget.check_request(count)
+        value = self._buffer.read(count)
+        self._budget.check_deadline()
+        if len(value) != count:
+            self._budget._refuse("RAW_READ_INCOMPLETE", "truncated")
+        return value
+
+    def close(self) -> None:
+        self._buffer.close()
diff --git a/tests/worker_startup_read_budget.py b/tests/worker_startup_read_budget.py
index 2912f59..c4fed5b 100644
--- a/tests/worker_startup_read_budget.py
+++ b/tests/worker_startup_read_budget.py
@@ -53,21 +53,42 @@ def _check_time(self) -> None:
         if sample >= self._deadline_ns:
             self._refuse("TIME_BUDGET_EXPIRED", "truncated" if self.started else "missing")
 
-    def read(self, count: int) -> bytes:
+    def _validate_request(self, count: int) -> None:
         if self.refusal is not None:
             raise ReadRefusal("READ_ALREADY_REFUSED", self.refusal[1])
         if type(count) is not int or count < 0:
             self._refuse("READ_COUNT_INVALID", "invalid")
-        if count > self.REQUEST_LIMIT or self.reserved_bytes + count > self.PHASE_LIMIT:
+        if count > self.REQUEST_LIMIT:
+            self._refuse("RECEIVE_LIMIT_EXCEEDED", "truncated")
+
+    def check_request(self, count: int) -> None:
+        """Bound a logical request without charging already buffered bytes again."""
+        self._validate_request(count)
+        self._check_time()
+
+    def check_deadline(self) -> None:
+        self._validate_request(0)
+        self._check_time()
+
+    def read(self, count: int) -> bytes:
+        return self._read(count, self._read_bytes, exact=True)
+
+    def read_some(self, count: int, read_bytes: Callable[[int], bytes]) -> bytes:
+        """Charge a raw-stream reservation; short returns never refund it."""
+        return self._read(count, read_bytes, exact=False)
+
+    def _read(self, count: int, read_bytes: Callable[[int], bytes], *, exact: bool) -> bytes:
+        self._validate_request(count)
+        if self.reserved_bytes + count > self.PHASE_LIMIT:
             self._refuse("RECEIVE_LIMIT_EXCEEDED", "truncated")
         self._check_time()
         self.reserved_bytes += count
         self.started = True
         try:
-            value = self._read_bytes(count)
+            value = read_bytes(count)
         except Exception:
             self._refuse("RAW_READ_FAILED", "truncated")
         self._check_time()
-        if type(value) is not bytes or len(value) != count:
+        if type(value) is not bytes or len(value) > count or (exact and len(value) != count):
             self._refuse("RAW_READ_INCOMPLETE", "truncated")
         return value

From 71863ca594d286c0af6516eba00b1611e708f12c Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 00:54:50 +0000
Subject: [PATCH 2/5] Propose initial and post-TLS reader assignment binding

---
 .../collector-proposal/README.md              | 26 ++++--
 .../collector-proposal/proposal.json          |  9 +-
 tests/test_worker_startup_buffered.py         | 91 ++++++++++++++++++-
 tests/worker_startup_buffered.py              | 58 +++++++++++-
 4 files changed, 174 insertions(+), 10 deletions(-)

diff --git a/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md b/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
index 3314d5a..77e12e3 100644
--- a/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
+++ b/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
@@ -49,12 +49,24 @@ controls cover that reservation, cached reads, shared streams, oversize/expiry
 refusals, short EOF and close without draining. They use fixture raw streams,
 not a socket or the proposed driver.
 
-The actual driver binding is still absent. The pinned connection source creates
-its initial buffered reader and recreates it after TLS wrapping. Both sites
-must install the budgeted raw boundary before any read; this primitive refuses
-an already buffered input and does not establish either lifecycle hook. TLS
+The proposed `ProtocolFileBinding` mixin intercepts `_rfile` assignment. The
+bound connection source assigns a fresh `sock.makefile("rb")` in `connect`
+(line 698) and a fresh `self._sock.makefile("rb")` after TLS wrapping (line 953).
+Place this mixin before the pinned `Connection` in the future fixture class's
+method resolution order, initialize its budget before the deferred driver
+constructor, and retain the driver's packet parser and `_read_bytes` method.
+The setter immediately detaches that fresh buffer and installs the budgeted raw
+boundary before its first read. It shares the budget across replacements,
+refuses to discard unread prefetched bytes, and closes a rejected newly owned
+stream. `None` assignment closes without a read or a fresh deadline allowance.
+Freshness of the incoming buffer is a requirement of these two exact source
+sites; the helper does not certify an arbitrary buffer supplied after use.
+
+Five additional inert controls exercise those assignment/replacement/close
+boundaries with stock in-memory I/O. No PyMySQL class has been imported or
+composed. Actual class composition and driver lifecycle qualification, TLS
 handshake deadlines, authentication, result terminators, process ownership,
-framing and cleanup remain unimplemented. The collector vectors below remain
+framing and cleanup remain unqualified. The collector vectors below remain
 unexecuted, and all actual operating inputs remain unbound.
 
 The process receives only the fixture identity and dedicated SELECT-only
@@ -123,8 +135,8 @@ still share the original absolute phase/case deadline and resource envelope.
 They receive no separate allowance, and no custom TLS engine is proposed.
 A failed read does not refund that reservation or permit a retry. The adapter
 must bind both initial and post-TLS raw readers against the pinned connection
-source before adoption. That lifecycle binding has not been implemented or
-exercised here.
+source before adoption. The mixin above is proposed source; its actual driver
+composition and lifecycle have not been exercised here.
 Construct with `defer_connect=True`, initialize the deadline/receive guard before
 any handshake, and connect only after fixture identity/input validation. The
 complete process lifetime and its output are also bounded by the owner. Send
diff --git a/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json b/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json
index d926d6f..08d62f5 100644
--- a/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json
+++ b/tests/fixtures/worker_startup_reconstruction/collector-proposal/proposal.json
@@ -168,6 +168,13 @@
     "actual_driver_initial_and_post_tls_assignment_binding": null,
     "tls_handshake_deadline_enforcement": null,
     "actual_driver_imported": false,
-    "replacement_streams_share_supplied_phase_budget": true
+    "replacement_streams_share_supplied_phase_budget": true,
+    "proposed_assignment_mixin": "ProtocolFileBinding before pinned Connection; initialize budget before deferred constructor",
+    "pinned_fresh_buffer_assignment_lines": [
+      698,
+      953
+    ],
+    "replacement_with_unread_protocol_data": "refuse",
+    "actual_driver_composition_and_lifecycle_qualification": null
   }
 }
diff --git a/tests/test_worker_startup_buffered.py b/tests/test_worker_startup_buffered.py
index b54fd66..8818c9a 100644
--- a/tests/test_worker_startup_buffered.py
+++ b/tests/test_worker_startup_buffered.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from tests.worker_startup_buffered import ProtocolReader
+from tests.worker_startup_buffered import ProtocolFileBinding, ProtocolReader
 from tests.worker_startup_read_budget import ReadRefusal, ReceiveBudget
 
 
@@ -143,3 +143,92 @@ def test_an_existing_buffered_reader_is_not_admitted_as_a_raw_source() -> None:
         assert raw.requests == []
     finally:
         buffered.close()
+
+
+def test_assignment_binding_detaches_the_fresh_buffer_before_its_first_read() -> None:
+    raw = RecordingRaw(b"x" * 8192)
+    original_buffer = io.BufferedReader(raw)
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
+    binding = ProtocolFileBinding(budget)
+    try:
+        binding._rfile = original_buffer
+        assert raw.requests == []
+        assert binding._rfile is not None
+        assert binding._rfile.read(1) == b"x"
+        assert raw.requests == [8192]
+        assert budget.reserved_bytes == 8192
+        assert binding._rfile.unread_bytes == 8191
+        with pytest.raises(ValueError):
+            original_buffer.read(1)
+    finally:
+        binding._rfile = None
+
+
+def test_fresh_replacement_cannot_reset_the_phase_budget() -> None:
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
+    binding = ProtocolFileBinding(budget)
+    first = RecordingRaw(b"a" * 32768)
+    second = RecordingRaw(b"b" * 32768)
+    extra = RecordingRaw(b"c" * 8192)
+    try:
+        binding._rfile = io.BufferedReader(first)
+        assert binding._rfile is not None
+        assert binding._rfile.read(32768) == b"a" * 32768
+        binding._rfile = io.BufferedReader(second)
+        assert first.closed
+        assert binding._rfile is not None
+        assert binding._rfile.read(32768) == b"b" * 32768
+        binding._rfile = io.BufferedReader(extra)
+        assert second.closed
+        assert binding._rfile is not None
+        with pytest.raises(ReadRefusal, match="RECEIVE_LIMIT_EXCEEDED"):
+            binding._rfile.read(1)
+        assert budget.reserved_bytes == 65536
+        assert extra.requests == []
+    finally:
+        binding._rfile = None
+
+
+def test_replacement_cannot_discard_prefetched_protocol_data() -> None:
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
+    binding = ProtocolFileBinding(budget)
+    first = RecordingRaw(b"a" * 8192)
+    extra = RecordingRaw(b"b" * 8192)
+    try:
+        binding._rfile = io.BufferedReader(first)
+        reader = binding._rfile
+        assert reader is not None
+        assert reader.read(1) == b"a"
+        with pytest.raises(ReadRefusal, match="STREAM_REPLACEMENT_UNREAD"):
+            binding._rfile = io.BufferedReader(extra)
+        assert binding._rfile is reader
+        assert extra.closed
+        assert extra.requests == []
+        assert budget.reserved_bytes == 8192
+        with pytest.raises(ReadRefusal, match="READ_ALREADY_REFUSED"):
+            reader.read(1)
+    finally:
+        binding._rfile = None
+    assert first.closed
+
+
+def test_expired_assignment_closes_its_new_stream_without_reading() -> None:
+    raw = RecordingRaw(b"unused")
+    binding = ProtocolFileBinding(ReceiveBudget(lambda _: b"", 10, clock=lambda: 10))
+    with pytest.raises(ReadRefusal, match="TIME_BUDGET_EXPIRED"):
+        binding._rfile = io.BufferedReader(raw)
+    assert raw.closed
+    assert raw.requests == []
+    assert binding._rfile is None
+
+
+def test_none_assignment_closes_without_a_drain_or_remaining_read_budget() -> None:
+    now = [1]
+    raw = RecordingRaw(b"unused")
+    binding = ProtocolFileBinding(ReceiveBudget(lambda _: b"", 10, clock=lambda: now[0]))
+    binding._rfile = io.BufferedReader(raw)
+    now[0] = 10
+    binding._rfile = None
+    binding._rfile = None
+    assert raw.close_count == 1
+    assert raw.requests == []
diff --git a/tests/worker_startup_buffered.py b/tests/worker_startup_buffered.py
index 7acc823..6266aa1 100644
--- a/tests/worker_startup_buffered.py
+++ b/tests/worker_startup_buffered.py
@@ -17,6 +17,7 @@ def __init__(self, raw: io.RawIOBase, budget: ReceiveBudget) -> None:
         super().__init__()
         self._raw = raw
         self._budget = budget
+        self.received_bytes = 0
 
     def readable(self) -> bool:
         return True
@@ -27,6 +28,7 @@ def readinto(self, buffer: Any) -> int:
             raise ValueError("Protocol read buffer must be writable")
         value = self._budget.read_some(len(view), self._raw.read)
         view[:len(value)] = value
+        self.received_bytes += len(value)
         return len(value)
 
     def close(self) -> None:
@@ -45,7 +47,13 @@ def __init__(self, raw: io.RawIOBase, budget: ReceiveBudget) -> None:
         if not isinstance(raw, io.RawIOBase):
             raise TypeError("Protocol source must be an unbuffered raw stream")
         self._budget = budget
-        self._buffer = io.BufferedReader(_ReservedRaw(raw, budget), buffer_size=8192)
+        self._reserved_raw = _ReservedRaw(raw, budget)
+        self._buffer = io.BufferedReader(self._reserved_raw, buffer_size=8192)
+        self._returned_bytes = 0
+
+    @property
+    def unread_bytes(self) -> int:
+        return self._reserved_raw.received_bytes - self._returned_bytes
 
     def read(self, count: int) -> bytes:
         self._budget.check_request(count)
@@ -53,7 +61,55 @@ def read(self, count: int) -> bytes:
         self._budget.check_deadline()
         if len(value) != count:
             self._budget._refuse("RAW_READ_INCOMPLETE", "truncated")
+        self._returned_bytes += len(value)
         return value
 
     def close(self) -> None:
         self._buffer.close()
+
+
+class ProtocolFileBinding:
+    """Proposed mixin before the pinned Connection in the future owner's MRO.
+
+    Initialize this binding before the deferred driver constructor. Its property
+    intercepts the driver's two fresh makefile assignments before any read.
+    No driver is imported or constructed by this source module.
+    """
+
+    def __init__(self, budget: ReceiveBudget) -> None:
+        self._protocol_budget = budget
+        self._protocol_reader: ProtocolReader | None = None
+
+    @property
+    def _rfile(self) -> ProtocolReader | None:
+        return getattr(self, "_protocol_reader", None)
+
+    @_rfile.setter
+    def _rfile(self, value: io.BufferedReader | None) -> None:
+        previous = self._rfile
+        if value is None:
+            if previous is not None:
+                previous.close()
+            self._protocol_reader = None
+            return
+        if type(value) is not io.BufferedReader:
+            self._protocol_budget._refuse("READ_STREAM_INVALID", "invalid")
+        raw = None
+        reader = None
+        try:
+            self._protocol_budget.check_deadline()
+            if previous is not None and previous.unread_bytes:
+                self._protocol_budget._refuse("STREAM_REPLACEMENT_UNREAD", "invalid")
+            raw = value.detach()
+            reader = ProtocolReader(raw, self._protocol_budget)
+            if previous is not None:
+                previous.close()
+            self._protocol_reader = reader
+        except BaseException:
+            if reader is not None:
+                reader.close()
+            elif raw is not None:
+                raw.close()
+            else:
+                value.close()
+            raise

From 96ac9bee860c35ff1ef6b085848ded2898f68b68 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 02:16:28 +0000
Subject: [PATCH 3/5] ci: preserve raw bytes in corpus baseline reads

---
 scripts/ci/test-validate-regression-corpus.py | 34 +++++++++++++++++++
 scripts/ci/validate-regression-corpus.py      | 16 ++++++---
 2 files changed, 45 insertions(+), 5 deletions(-)

diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py
index 90a1386..822c4de 100644
--- a/scripts/ci/test-validate-regression-corpus.py
+++ b/scripts/ci/test-validate-regression-corpus.py
@@ -782,6 +782,40 @@ def _validate(self) -> dict[str, Any]:
             codec_runner_path=self.codec_runner,
         )
 
+    def test_binary_base_asset_does_not_prevent_corpus_validation(self) -> None:
+        asset = self.root / "assets/font.woff2"
+        asset.parent.mkdir()
+        data = b"wOF2\x00\xbc\xff\x80\r\n"
+        asset.write_bytes(data)
+        self._git("add", "assets/font.woff2")
+        self._git(
+            "-c", "user.name=Regression Corpus Test",
+            "-c", "user.email=regression-corpus@example.invalid",
+            "commit", "--quiet", "--message=binary-baseline",
+        )
+
+        result = self._validate()
+
+        self.assertFalse(result["counts"]["codec"]["related_change"])
+        self.assertEqual(VALIDATOR._ref_files(self.root, "HEAD")["assets/font.woff2"], data)
+
+    def test_unchanged_crlf_fixture_keeps_exact_base_bytes(self) -> None:
+        relative = "tests/fixtures/codec_regressions/base.json"
+        fixture = self.root / relative
+        data = fixture.read_bytes().replace(b"\n", b"\r\n")
+        fixture.write_bytes(data)
+        self._git("-c", "core.autocrlf=false", "add", relative)
+        self._git(
+            "-c", "user.name=Regression Corpus Test",
+            "-c", "user.email=regression-corpus@example.invalid",
+            "commit", "--quiet", "--message=crlf-baseline",
+        )
+
+        result = self._validate()
+
+        self.assertFalse(result["counts"]["codec"]["related_change"])
+        self.assertEqual(VALIDATOR._ref_files(self.root, "HEAD")[relative], data)
+
     def _add_avro_golden_to_base(
         self,
         malformed_wire: str | None = None,
diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py
index a80960f..511bbe5 100644
--- a/scripts/ci/validate-regression-corpus.py
+++ b/scripts/ci/validate-regression-corpus.py
@@ -1080,11 +1080,17 @@ def _tracked_worktree_files(root: Path) -> dict[str, bytes]:
 
 def _ref_files(root: Path, ref: str) -> dict[str, bytes]:
     paths = _run(["git", "ls-tree", "-r", "--name-only", "-z", ref], root).split("\0")
-    return {
-        path: _run(["git", "show", f"{ref}:{path}"], root).encode()
-        for path in paths
-        if path
-    }
+    files = {}
+    for path in filter(None, paths):
+        # Assets may be binary, and immutable fixtures retain their exact line endings.
+        result = subprocess.run(
+            ["git", "show", f"{ref}:{path}"], cwd=root, capture_output=True, check=False,
+        )
+        if result.returncode != 0:
+            detail = result.stderr.decode("utf-8", errors="replace").strip()
+            raise CorpusError(f"git show {ref}:{path} failed: {detail}")
+        files[path] = result.stdout
+    return files
 
 
 def _matches(path: str, pattern: str) -> bool:

From 40cf2db4346b9c48e60569d1856d0fbfbde442e2 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 02:21:31 +0000
Subject: [PATCH 4/5] ci: delimit Git revisions and isolate refusal assertion

---
 scripts/ci/validate-regression-corpus.py | 3 ++-
 tests/test_worker_startup_buffered.py    | 3 ++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py
index 511bbe5..fdcfe78 100644
--- a/scripts/ci/validate-regression-corpus.py
+++ b/scripts/ci/validate-regression-corpus.py
@@ -1084,7 +1084,8 @@ def _ref_files(root: Path, ref: str) -> dict[str, bytes]:
     for path in filter(None, paths):
         # Assets may be binary, and immutable fixtures retain their exact line endings.
         result = subprocess.run(
-            ["git", "show", f"{ref}:{path}"], cwd=root, capture_output=True, check=False,
+            ["git", "show", "--end-of-options", f"{ref}:{path}"],
+            cwd=root, capture_output=True, check=False,
         )
         if result.returncode != 0:
             detail = result.stderr.decode("utf-8", errors="replace").strip()
diff --git a/tests/test_worker_startup_buffered.py b/tests/test_worker_startup_buffered.py
index 8818c9a..ed85c8a 100644
--- a/tests/test_worker_startup_buffered.py
+++ b/tests/test_worker_startup_buffered.py
@@ -137,9 +137,10 @@ def test_close_is_idempotent_and_does_not_drain_protocol_data() -> None:
 def test_an_existing_buffered_reader_is_not_admitted_as_a_raw_source() -> None:
     raw = RecordingRaw(b"x")
     buffered = io.BufferedReader(raw)
+    budget = ReceiveBudget(lambda _: b"", 10, clock=lambda: 1)
     try:
         with pytest.raises(TypeError, match="unbuffered raw stream"):
-            ProtocolReader(buffered, ReceiveBudget(lambda _: b"", 10, clock=lambda: 1))
+            ProtocolReader(buffered, budget)
         assert raw.requests == []
     finally:
         buffered.close()

From 9e30a79ed4840a0a7c4f835252eac8b0f2c47bfe Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 02:24:22 +0000
Subject: [PATCH 5/5] ci: bind corpus baseline to its resolved commit

---
 scripts/ci/test-validate-regression-corpus.py | 7 +++++++
 scripts/ci/validate-regression-corpus.py      | 6 +++++-
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py
index 822c4de..ced15a8 100644
--- a/scripts/ci/test-validate-regression-corpus.py
+++ b/scripts/ci/test-validate-regression-corpus.py
@@ -816,6 +816,13 @@ def test_unchanged_crlf_fixture_keeps_exact_base_bytes(self) -> None:
         self.assertFalse(result["counts"]["codec"]["related_change"])
         self.assertEqual(VALIDATOR._ref_files(self.root, "HEAD")[relative], data)
 
+    def test_symbolic_base_ref_is_bound_to_its_complete_commit_identity(self) -> None:
+        expected = VALIDATOR._run(["git", "rev-parse", "HEAD"], self.root).strip()
+
+        result = self._validate()
+
+        self.assertEqual(result["base_ref"], expected)
+
     def _add_avro_golden_to_base(
         self,
         malformed_wire: str | None = None,
diff --git a/scripts/ci/validate-regression-corpus.py b/scripts/ci/validate-regression-corpus.py
index fdcfe78..270d863 100644
--- a/scripts/ci/validate-regression-corpus.py
+++ b/scripts/ci/validate-regression-corpus.py
@@ -1422,7 +1422,11 @@ def validate(
     base_files: dict[str, bytes] = {}
     base_evidence: list[Evidence] = []
     if base_ref and not ZERO_COMMIT.fullmatch(base_ref):
-        _run(["git", "rev-parse", "--verify", f"{base_ref}^{{commit}}"], root)
+        base_ref = _run(
+            ["git", "rev-parse", "--verify", "--end-of-options", f"{base_ref}^{{commit}}"], root,
+        ).strip()
+        if not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", base_ref):
+            raise CorpusError("base ref must resolve to a complete Git commit ID")
         changed, added_paths = _changed_paths(root, base_ref)
         base_files = _ref_files(root, base_ref)
         try: