diff --git a/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md b/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
index e4b1e63..30e20d5 100644
--- a/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
+++ b/tests/fixtures/worker_startup_reconstruction/collector-proposal/README.md
@@ -21,6 +21,26 @@ import closure, authentication/TLS dependencies and installed input are **not
 selected 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`.
+`bind_original_query` verifies the two immutable SQL texts, translates their
+closed placeholder inventory and returns values separately without invoking a
+driver. `decode_raw_snapshot` validates supplied ordered aliases and positional
+bytes/null values, enforces field-size limits, pads valid whole-second timestamp
+text and calls the existing projection decoder. Its controls consume inert
+tuples only. They do not prove that a driver produced those types or that a read
+completed; the future collector must establish its bounded capture before using
+this transform. A complete supplied projection still has an unknown phase.
+`tests/worker_startup_read_budget.py` adds the next pure callback wrapper. It
+reserves each requested count before the callback, enforces the proposed
+32-KiB/64-KiB limits, never refunds a failed read, refuses further calls after
+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.
+
 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
diff --git a/tests/test_worker_startup_raw.py b/tests/test_worker_startup_raw.py
new file mode 100644
index 0000000..9413ca0
--- /dev/null
+++ b/tests/test_worker_startup_raw.py
@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+import datetime as dt
+from dataclasses import replace
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tests import worker_startup_raw as raw
+from tests.test_worker_startup_snapshot import IDENTITY, NOW, row
+from tests.worker_startup_snapshot import RUN_FIELDS, TASK_FIELDS, ProjectionRefusal
+
+
+def supplied(kind: Any) -> tuple[tuple[str, ...], tuple[Any, ...]]:
+    columns = raw.TASK_COLUMNS if kind == "task" else raw.RUN_COLUMNS
+    values = row(kind)
+    return columns, tuple(None if values[field] is None else str(values[field]).encode("ascii") for field in columns)
+
+
+@pytest.mark.parametrize("kind,fields", [("task", TASK_FIELDS), ("run", RUN_FIELDS)])
+def test_positional_original_rows_decode_without_a_capture_or_phase_verdict(kind: Any, fields: set[str]) -> None:
+    aliases, values = supplied(kind)
+    result = raw.decode_raw_snapshot(kind, IDENTITY, aliases, [values])
+    assert set(aliases) == fields
+    assert result.projection_state == "complete"
+    assert result.phase_verdict == "unknown"
+    assert result.row is not None
+    assert result.row["database_time_utc"] == NOW
+    assert result.row["session_time_zone"] == "SYSTEM"
+
+
+@pytest.mark.parametrize("kind", ["task", "run"])
+def test_empty_and_duplicate_results_do_not_select_or_borrow_a_row(kind: Any) -> None:
+    aliases, values = supplied(kind)
+    missing = raw.decode_raw_snapshot(kind, IDENTITY, aliases, [])
+    assert missing.projection_state == "missing"
+    assert missing.row is None
+    duplicate = raw.decode_raw_snapshot(kind, IDENTITY, aliases, [values, values])
+    assert duplicate.projection_state == "invalid"
+    assert duplicate.row is None
+
+
+@pytest.mark.parametrize("change", ["duplicate", "order", "case", "missing", "bytes", "list"])
+def test_column_collision_or_order_change_is_refused_before_mapping(change: str) -> None:
+    columns, values = supplied("task")
+    aliases: Any = columns
+    if change == "duplicate":
+        aliases = (columns[1], *columns[1:])
+    elif change == "order":
+        aliases = (columns[1], columns[0], *columns[2:])
+    elif change == "case":
+        aliases = (columns[0].upper(), *columns[1:])
+    elif change == "missing":
+        aliases = columns[:-1]
+    elif change == "bytes":
+        aliases = tuple(field.encode() for field in columns)
+    else:
+        aliases = list(columns)
+    result = raw.decode_raw_snapshot("task", IDENTITY, aliases, [values])
+    assert result.refusal_code == "RAW_COLUMN_SET_INVALID"
+    assert result.row is None
+
+
+@pytest.mark.parametrize("value", [dt.datetime(2026, 11, 5), 1, True, 1.0, [], b"x" * 256, b"\xff"])
+def test_native_or_oversized_values_are_never_coerced(value: Any) -> None:
+    aliases, supplied_values = supplied("task")
+    values = list(supplied_values)
+    values[aliases.index("compatibility")] = value
+    result = raw.decode_raw_snapshot("task", IDENTITY, aliases, [tuple(values)])
+    assert result.projection_state == "invalid"
+    assert result.field == "compatibility"
+    assert result.row is None
+
+
+def test_rejected_value_is_not_formatted_or_retained() -> None:
+    class DoNotFormat:
+        def __str__(self) -> str:
+            raise AssertionError("implicit string conversion")
+
+        def __repr__(self) -> str:
+            raise AssertionError("implicit representation")
+
+    aliases, supplied_values = supplied("task")
+    values = list(supplied_values)
+    values[aliases.index("queue")] = DoNotFormat()
+    result = raw.decode_raw_snapshot("task", IDENTITY, aliases, [tuple(values)])
+    assert result.refusal_code == "RAW_SCALAR_TYPE_OR_SIZE"
+    assert result.row is None
+
+
+@pytest.mark.parametrize("value,expected", [(None, None), (b"", ""), (b"NULL", "NULL")])
+def test_null_empty_and_literal_null_keep_their_meaning(value: Any, expected: str | None) -> None:
+    aliases, supplied_values = supplied("task")
+    values = list(supplied_values)
+    values[aliases.index("compatibility")] = value
+    result = raw.decode_raw_snapshot("task", IDENTITY, aliases, [tuple(values)])
+    assert result.projection_state == "complete"
+    assert result.row is not None
+    assert result.row["compatibility"] == expected
+
+
+@pytest.mark.parametrize("value", [b"2026-11-05 20:00:00", NOW.encode()])
+def test_whole_seconds_are_padded_without_clock_conversion(value: bytes) -> None:
+    aliases, supplied_values = supplied("run")
+    values = list(supplied_values)
+    values[0] = value
+    result = raw.decode_raw_snapshot("run", IDENTITY, aliases, [tuple(values)])
+    assert result.projection_state == "complete"
+    assert result.row is not None
+    assert result.row["database_time_utc"] == (NOW if len(value) == 26 else "2026-11-05 20:00:00.000000")
+    assert result.phase_verdict == "unknown"
+
+
+@pytest.mark.parametrize("value", [
+    b"2026-02-30 20:00:00", b"0000-00-00 00:00:00", b"2026-11-05", b"2026-11-05 20:00:00.1",
+    b"2026-11-05 20:00:00.123456Z", b"2026-11-05T20:00:00",
+])
+def test_invalid_or_partial_timestamp_is_not_normalized_into_evidence(value: bytes) -> None:
+    aliases, supplied_values = supplied("run")
+    values = (value, *supplied_values[1:])
+    result = raw.decode_raw_snapshot("run", IDENTITY, aliases, [values])
+    assert result.projection_state == "invalid"
+    assert result.row is None
+
+
+@pytest.mark.parametrize("kind", ["task", "run"])
+def test_query_parameters_remain_separate_from_the_closed_sql_text(kind: Any) -> None:
+    identity = replace(IDENTITY, namespace="quote' :namespace %(original_run_id)s")
+    query, parameters = raw.bind_original_query(kind, identity)
+    assert identity.namespace not in query
+    assert parameters["namespace"] == identity.namespace
+    assert query.count("%(namespace)s") == 1
+    assert query.count("%(original_run_id)s") == 1
+    assert set(parameters) == {
+        "namespace", "original_run_id", "original_task_id" if kind == "task" else "workflow_instance_id",
+    }
+
+
+def test_changed_sql_source_cannot_be_used_as_an_arbitrary_query(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    source = raw.SQL_DIRECTORY / "original-task.sql"
+    (tmp_path / source.name).write_bytes(source.read_bytes().replace(b"SELECT", b"DELETE", 1))
+    monkeypatch.setattr(raw, "SQL_DIRECTORY", tmp_path)
+    with pytest.raises(ProjectionRefusal, match="SQL_SOURCE_CHANGED"):
+        raw.bind_original_query("task", IDENTITY)
+
+
+@pytest.mark.parametrize("kind", ["task", "run"])
+def test_ordered_aliases_match_the_immutable_projection_text(kind: Any) -> None:
+    query = (raw.SQL_DIRECTORY / f"original-{kind}.sql").read_text()
+    selected = query.split("SELECT\n", 1)[1].split("\nFROM ", 1)[0]
+    aliases = tuple(line.strip().rstrip(",").rsplit(" AS ", 1)[-1].rsplit(".", 1)[-1] for line in selected.splitlines())
+    assert aliases == (raw.TASK_COLUMNS if kind == "task" else raw.RUN_COLUMNS)
+
+
+def test_missing_projection_text_returns_a_fixed_refusal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(raw, "SQL_DIRECTORY", tmp_path)
+    with pytest.raises(ProjectionRefusal, match="SQL_SOURCE_UNAVAILABLE"):
+        raw.bind_original_query("run", IDENTITY)
+
+
+def test_raw_identity_comparison_is_case_sensitive() -> None:
+    aliases, supplied_values = supplied("task")
+    values = list(supplied_values)
+    values[aliases.index("original_run_id")] = IDENTITY.original_run_id.lower().encode()
+    result = raw.decode_raw_snapshot("task", IDENTITY, aliases, [tuple(values)])
+    assert result.refusal_code == "IDENTITY_MISMATCH"
+    assert result.row is None
+
+
+@pytest.mark.parametrize("value", [b"0", b"4294967295"])
+def test_canonical_raw_counter_boundaries_pass_the_existing_decoder(value: bytes) -> None:
+    aliases, supplied_values = supplied("task")
+    values = list(supplied_values)
+    values[aliases.index("attempt_count")] = value
+    values[aliases.index("lease_owner_present")] = b"1"
+    result = raw.decode_raw_snapshot("task", IDENTITY, aliases, [tuple(values)])
+    assert result.projection_state == "complete"
+    assert result.row is not None
+    assert result.row["attempt_count"] == int(value)
+    assert result.row["lease_owner_present"] is True
+
+
+@pytest.mark.parametrize("change", ["short", "list", "noncanonical-bit", "noncanonical-counter"])
+def test_incomplete_row_or_noncanonical_number_cannot_be_coerced(change: str) -> None:
+    aliases, supplied_values = supplied("task")
+    values: Any = supplied_values
+    if change == "short":
+        values = values[:-1]
+    elif change == "list":
+        values = list(values)
+    else:
+        changed = list(values)
+        changed[aliases.index("lease_owner_present" if change == "noncanonical-bit" else "attempt_count")] = b"01"
+        values = tuple(changed)
+    result = raw.decode_raw_snapshot("task", IDENTITY, aliases, [values])
+    assert result.projection_state == "invalid"
+    assert result.row is None
diff --git a/tests/test_worker_startup_read_budget.py b/tests/test_worker_startup_read_budget.py
new file mode 100644
index 0000000..47baa16
--- /dev/null
+++ b/tests/test_worker_startup_read_budget.py
@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from tests.worker_startup_read_budget import ReadRefusal, ReceiveBudget
+
+
+def test_sequential_requests_share_one_reservation_budget() -> None:
+    calls = []
+
+    def read(count: int) -> bytes:
+        calls.append(count)
+        return b"x" * count
+
+    budget = ReceiveBudget(read, 10, clock=lambda: 1)
+    assert len(budget.read(32768)) == 32768
+    assert len(budget.read(32768)) == 32768
+    with pytest.raises(ReadRefusal, match="RECEIVE_LIMIT_EXCEEDED") as error:
+        budget.read(1)
+    assert error.value.state == "truncated"
+    assert budget.reserved_bytes == 65536
+    assert calls == [32768, 32768]
+
+
+def test_oversized_request_is_refused_before_the_callback_or_allocation() -> None:
+    calls = []
+    budget = ReceiveBudget(lambda count: calls.append(count) or b"", 10, clock=lambda: 1)
+    with pytest.raises(ReadRefusal, match="RECEIVE_LIMIT_EXCEEDED"):
+        budget.read(32769)
+    assert calls == []
+    assert budget.reserved_bytes == 0
+
+
+def test_failed_read_keeps_its_reservation_and_cannot_be_retried() -> None:
+    calls = []
+
+    def fail(count: int) -> bytes:
+        calls.append(count)
+        raise RuntimeError("unretained transport diagnostic")
+
+    budget = ReceiveBudget(fail, 10, clock=lambda: 1)
+    with pytest.raises(ReadRefusal, match="RAW_READ_FAILED") as error:
+        budget.read(16)
+    assert str(error.value) == "RAW_READ_FAILED"
+    assert error.value.__cause__ is None
+    with pytest.raises(ReadRefusal, match="READ_ALREADY_REFUSED"):
+        budget.read(16)
+    assert budget.refusal == ("RAW_READ_FAILED", "truncated")
+    assert budget.reserved_bytes == 16
+    assert calls == [16]
+
+
+def test_expired_before_first_read_is_missing_and_issues_no_callback() -> None:
+    calls = []
+    budget = ReceiveBudget(lambda count: calls.append(count) or b"", 10, clock=lambda: 10)
+    with pytest.raises(ReadRefusal, match="TIME_BUDGET_EXPIRED") as error:
+        budget.read(4)
+    assert error.value.state == "missing"
+    assert budget.started is False
+    assert calls == []
+
+
+def test_fast_reads_cannot_reset_the_absolute_deadline() -> None:
+    now = [0]
+
+    def read(count: int) -> bytes:
+        now[0] += 6
+        return b"x" * count
+
+    budget = ReceiveBudget(read, 10, clock=lambda: now[0])
+    assert budget.read(4) == b"xxxx"
+    with pytest.raises(ReadRefusal, match="TIME_BUDGET_EXPIRED") as error:
+        budget.read(4)
+    assert error.value.state == "truncated"
+    assert budget.reserved_bytes == 8
+    assert now == [12]
+
+
+@pytest.mark.parametrize("count", [True, False, -1, 1.0, "1", None])
+def test_invalid_read_count_does_not_reach_the_reader(count: Any) -> None:
+    calls = []
+    budget = ReceiveBudget(lambda value: calls.append(value) or b"", 10, clock=lambda: 1)
+    with pytest.raises(ReadRefusal, match="READ_COUNT_INVALID"):
+        budget.read(count)
+    assert calls == []
+
+
+@pytest.mark.parametrize("deadline", [True, -1, 1.0, None, 2**63])
+def test_deadline_requires_a_bounded_monotonic_integer(deadline: Any) -> None:
+    with pytest.raises(ReadRefusal, match="DEADLINE_INVALID"):
+        ReceiveBudget(lambda count: b"x" * count, deadline)
+
+
+@pytest.mark.parametrize("value", [b"", b"xx", bytearray(b"x"), memoryview(b"x"), "x", None])
+def test_partial_or_changed_native_result_cannot_be_promoted(value: Any) -> None:
+    budget = ReceiveBudget(lambda _: value, 10, clock=lambda: 1)
+    with pytest.raises(ReadRefusal, match="RAW_READ_INCOMPLETE") as error:
+        budget.read(1)
+    assert error.value.state == "truncated"
+    assert budget.reserved_bytes == 1
+
+
+@pytest.mark.parametrize("sample", [True, -1, 1.0, None, 2**63])
+def test_invalid_clock_sample_prevents_a_read(sample: Any) -> None:
+    calls = []
+    budget = ReceiveBudget(lambda count: calls.append(count) or b"", 10, clock=lambda: sample)
+    with pytest.raises(ReadRefusal, match="MONOTONIC_SAMPLE_INVALID"):
+        budget.read(1)
+    assert calls == []
+
+
+def test_clock_reversal_prevents_the_next_read() -> None:
+    now = [5]
+    calls = []
+    budget = ReceiveBudget(lambda count: calls.append(count) or b"x" * count, 10, clock=lambda: now[0])
+    assert budget.read(1) == b"x"
+    now[0] = 4
+    with pytest.raises(ReadRefusal, match="MONOTONIC_SAMPLE_INVALID"):
+        budget.read(1)
+    assert calls == [1]
diff --git a/tests/worker_startup_raw.py b/tests/worker_startup_raw.py
new file mode 100644
index 0000000..755be81
--- /dev/null
+++ b/tests/worker_startup_raw.py
@@ -0,0 +1,106 @@
+"""Pure binding and normalization of supplied historical fixture projections.
+
+No driver, query, capture-completeness proof, transport or process is provided.
+The future collector must establish complete bounded capture before using this
+transform; a complete projection retains the decoder's unknown phase verdict.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+from pathlib import Path
+from typing import Any, Literal, cast
+
+from tests.worker_startup_snapshot import (
+    RUN_TIMES,
+    TASK_TIMES,
+    OriginalIdentity,
+    ProjectionRefusal,
+    SnapshotProjection,
+    decode_snapshot,
+    identity_fields,
+    require,
+    timestamp,
+)
+
+Kind = Literal["task", "run"]
+SQL_DIRECTORY = Path(__file__).parent / "fixtures/worker_startup_reconstruction/sql-proposal"
+SQL_IDENTITIES = {
+    "task": (1140, "4385b4d0f08e22be9f053f282283111282f7215d12e9247d03ca0642e3e2b825"),
+    "run": (675, "025db37ab5311650d694f139941e206c2ae085d9b862426efa409a53ddd2ae13"),
+}
+TASK_COLUMNS = (
+    "database_time_utc", "session_time_zone", "system_time_zone", "original_task_id", "original_run_id",
+    "namespace", "task_type", "status", "queue", "compatibility", "available_at", "leased_at", "lease_expires_at",
+    "lease_owner_present", "attempt_count", "repair_count", "last_dispatch_attempt_at", "last_dispatched_at",
+    "last_claim_failed_at", "repair_available_at", "dispatch_error_present", "claim_error_present",
+    "task_error_present",
+)
+RUN_COLUMNS = (
+    "database_time_utc", "session_time_zone", "system_time_zone", "original_run_id", "workflow_instance_id",
+    "namespace", "status", "execution_deadline_at", "run_deadline_at",
+)
+TIMESTAMP_FIELDS = TASK_TIMES | RUN_TIMES | {"database_time_utc"}
+SCALAR_LIMITS = {
+    **dict.fromkeys(TIMESTAMP_FIELDS, 26),
+    **dict.fromkeys(("original_task_id", "original_run_id"), 26),
+    **dict.fromkeys(("attempt_count", "repair_count"), 10),
+    **dict.fromkeys(("lease_owner_present", "dispatch_error_present", "claim_error_present", "task_error_present"), 1),
+    "namespace": 128, "workflow_instance_id": 191, "status": 32, "task_type": 32,
+    "session_time_zone": 64, "system_time_zone": 64, "queue": 255, "compatibility": 255,
+}
+
+
+def bind_original_query(kind: Kind, identity: OriginalIdentity) -> tuple[str, dict[str, str]]:
+    """Return immutable SQL and separate values; no driver escaping/execution."""
+    parameters = identity_fields(kind, identity)
+    length, digest = SQL_IDENTITIES[kind]
+    try:
+        with (SQL_DIRECTORY / f"original-{kind}.sql").open("rb") as source:
+            raw = source.read(length + 1)
+    except OSError as error:
+        raise ProjectionRefusal("SQL_SOURCE_UNAVAILABLE") from error
+    require(len(raw) == length and hashlib.sha256(raw).hexdigest() == digest, "SQL_SOURCE_CHANGED")
+    query = raw.decode("ascii")
+    tokens = re.findall(r":([a-z_]+)", query)
+    require(len(tokens) == len(parameters) and set(tokens) == set(parameters), "SQL_TOKEN_SET_INVALID")
+    translated = re.sub(r":([a-z_]+)", lambda match: f"%({match[1]})s", query)
+    return translated, parameters
+
+
+def normalize_scalar(field: str, value: Any) -> str | None:
+    if value is None:
+        return None
+    require(type(value) is bytes and len(value) <= SCALAR_LIMITS[field], "RAW_SCALAR_TYPE_OR_SIZE", field)
+    try:
+        normalized = cast(bytes, value).decode("ascii")
+    except UnicodeDecodeError as error:
+        raise ProjectionRefusal("RAW_SCALAR_ENCODING", field) from error
+    if field in TIMESTAMP_FIELDS:
+        if len(normalized) == 19:
+            normalized += ".000000"
+        timestamp(normalized, field)
+    return normalized
+
+
+def decode_raw_snapshot(kind: Kind, identity: OriginalIdentity, aliases: Any, rows: Any) -> SnapshotProjection:
+    """Validate supplied positional bytes only; metadata/termination remain unproved."""
+    try:
+        identity_fields(kind, identity)
+        expected = TASK_COLUMNS if kind == "task" else RUN_COLUMNS
+        require(
+            type(aliases) is tuple and len(aliases) == len(expected)
+            and all(type(alias) is str for alias in aliases) and aliases == expected,
+            "RAW_COLUMN_SET_INVALID",
+        )
+        require(type(rows) in (tuple, list) and len(rows) <= 1, "ROW_COUNT_OR_TYPE_INVALID")
+        normalized = []
+        for row in rows:
+            require(type(row) is tuple and len(row) == len(expected), "RAW_ROW_SHAPE_INVALID")
+            normalized.append({
+                field: normalize_scalar(field, value) for field, value in zip(expected, row, strict=True)
+            })
+        return decode_snapshot(kind, identity, normalized)
+    except ProjectionRefusal as error:
+        return SnapshotProjection("invalid", error.code, error.field)
diff --git a/tests/worker_startup_read_budget.py b/tests/worker_startup_read_budget.py
new file mode 100644
index 0000000..2912f59
--- /dev/null
+++ b/tests/worker_startup_read_budget.py
@@ -0,0 +1,73 @@
+"""Pure receive-budget wrapper; it cannot interrupt a blocked driver callback.
+
+The proposed driver adapter must route every header/body/setup read through one
+instance before connecting. A process owner and confirmed cleanup remain absent.
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from typing import Literal, NoReturn
+
+State = Literal["missing", "truncated", "invalid"]
+
+
+class ReadRefusal(ValueError):
+    def __init__(self, code: str, state: State) -> None:
+        super().__init__(code)
+        self.code = code
+        self.state = state
+
+
+class ReceiveBudget:
+    REQUEST_LIMIT = 32_768
+    PHASE_LIMIT = 65_536
+
+    def __init__(
+        self, read_bytes: Callable[[int], bytes], deadline_ns: int,
+        *, clock: Callable[[], int] = time.monotonic_ns,
+    ) -> None:
+        if type(deadline_ns) is not int or not 0 <= deadline_ns <= 2**63 - 1:
+            raise ReadRefusal("DEADLINE_INVALID", "invalid")
+        self._read_bytes = read_bytes
+        self._clock = clock
+        self._deadline_ns = deadline_ns
+        self.reserved_bytes = 0
+        self.started = False
+        self.refusal: tuple[str, State] | None = None
+        self._last_sample = 0
+
+    def _refuse(self, code: str, state: State) -> NoReturn:
+        self.refusal = (code, state)
+        raise ReadRefusal(code, state) from None
+
+    def _check_time(self) -> None:
+        try:
+            sample = self._clock()
+        except Exception:
+            self._refuse("MONOTONIC_SAMPLE_INVALID", "invalid")
+        if type(sample) is not int or not self._last_sample <= sample <= 2**63 - 1:
+            self._refuse("MONOTONIC_SAMPLE_INVALID", "invalid")
+        self._last_sample = sample
+        if sample >= self._deadline_ns:
+            self._refuse("TIME_BUDGET_EXPIRED", "truncated" if self.started else "missing")
+
+    def read(self, count: int) -> bytes:
+        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:
+            self._refuse("RECEIVE_LIMIT_EXCEEDED", "truncated")
+        self._check_time()
+        self.reserved_bytes += count
+        self.started = True
+        try:
+            value = self._read_bytes(count)
+        except Exception:
+            self._refuse("RAW_READ_FAILED", "truncated")
+        self._check_time()
+        if type(value) is not bytes or len(value) != count:
+            self._refuse("RAW_READ_INCOMPLETE", "truncated")
+        return value