diff --git a/.gitignore b/.gitignore
index 4a00110..d6d51b3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -190,7 +190,7 @@ cython_debug/
 #   be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #   and can be added to the global gitignore or merged into this file.  For a more nuclear
 #   option (not recommended) you can uncomment the following to ignore the entire idea folder.
-# .idea/
+ .idea/
 
 # Abstra
 #   Abstra is an AI-powered process automation framework.
diff --git a/docs/reference/generated/seriousdb.exceptions.md b/docs/reference/generated/seriousdb.exceptions.md
index 4f9a65d..4bb532b 100644
--- a/docs/reference/generated/seriousdb.exceptions.md
+++ b/docs/reference/generated/seriousdb.exceptions.md
@@ -16,6 +16,12 @@ Base class for expected application errors.
   * **default_detail** (*str*) – Detail used when none is given.
   * **detail** (*str*) – Human readable description of this error.
 
+### *exception* DiskManagerError(detail=None)
+
+Bases: [`ApplicationError`](#seriousdb.exceptions.ApplicationError)
+
+A raw page read, write or allocation on the database file failed.
+
 ### *exception* ResourceNotFoundError(detail=None)
 
 Bases: [`ApplicationError`](#seriousdb.exceptions.ApplicationError)
diff --git a/flake.lock b/flake.lock
index fc28473..389caae 100644
--- a/flake.lock
+++ b/flake.lock
@@ -24,4 +24,4 @@
   },
   "root": "root",
   "version": 7
-}
+}
\ No newline at end of file
diff --git a/src/seriousdb/disk_manager.py b/src/seriousdb/disk_manager.py
new file mode 100644
index 0000000..96e5391
--- /dev/null
+++ b/src/seriousdb/disk_manager.py
@@ -0,0 +1,250 @@
+"""Raw fixed-size page I/O on the ``.sdb`` database file."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from threading import RLock
+from typing import Self
+
+from .exceptions import DiskManagerError
+from .storage_format import PAGE_SIZE
+from .types import UInt32
+
+__all__ = ["DiskManager"]
+
+# os.pread/os.pwrite do not exist on Windows. Looked up with getattr so that
+# type checkers do not depend on the platform they run on.
+_pread = getattr(os, "pread", None)
+_pwrite = getattr(os, "pwrite", None)
+# Windows only: stops newline translation.
+_O_BINARY = getattr(os, "O_BINARY", 0)
+
+
+@contextmanager
+def _os_errors(path: Path) -> Iterator[None]:
+    """Re-raise any OSError as DiskManagerError, keeping it as the cause."""
+    try:
+        yield
+    except OSError as e:
+        raise DiskManagerError(f"I/O error on {path}: {e}") from e
+
+
+class DiskManager:
+    """Read and write fixed-size pages of the database file by page ID.
+
+    Page n is stored at byte offset n * PAGE_SIZE. This class does raw
+    page I/O only: it knows nothing about page contents, and locking is left
+    to the concurrency manager.
+
+    Parameters
+    ----------
+    path : str or Path
+        Database file, created if missing.
+    repair_torn_page : bool, optional
+        Truncate a partial last page instead of raising. Default False.
+
+    Raises
+    ------
+    DiskManagerError
+        If the file size is not a multiple of PAGE_SIZE and
+        repair_torn_page is false.
+    OSError
+        If the file cannot be opened.
+
+    Notes
+    -----
+    A new file is empty, so the first :meth:allocate_page returns 0 and the
+    caller must write the metadata page there.
+    """
+
+    def __init__(self, path: str | Path, *, repair_torn_page: bool = False) -> None:
+        self._path = Path(path)
+        self._lock = RLock()
+        self._fd: int | None = None
+        self._known_pages = 0
+        self._active = 0  # I/O calls currently using the fd
+        with _os_errors(self._path):
+            fd = os.open(self._path, os.O_RDWR | os.O_CREAT | _O_BINARY, 0o644)
+        try:
+            with _os_errors(self._path):
+                size = os.fstat(fd).st_size
+                tail = size % PAGE_SIZE
+                if tail:
+                    if not repair_torn_page:
+                        raise DiskManagerError(
+                            f"{self._path} is {size} bytes, not a multiple of "
+                            f"{PAGE_SIZE}; the last page is incomplete"
+                        )
+                    size -= tail
+                    os.ftruncate(fd, size)
+                    os.fsync(fd)
+                self._known_pages = size // PAGE_SIZE
+        except BaseException:
+            os.close(fd)
+            raise
+        self._fd = fd
+
+    @property
+    def path(self) -> Path:
+        """Path of the database file."""
+        return self._path
+
+    @property
+    def page_count(self) -> int:
+        """Number of whole pages in the file, read from the file size."""
+        with self._in_flight() as fd:
+            return self._refresh_page_count(fd)
+
+    def read_page(self, page_id: UInt32) -> bytes:
+        """Return the PAGE_SIZE`` bytes of page page_id.
+
+        Raises
+        ------
+        DiskManagerError
+            If page_id is out of range, the file is closed, or the file
+            ends before a full page could be read.
+        """
+        with self._in_flight() as fd:
+            offset = self._offset(fd, page_id)
+            buf = bytearray()
+            while len(buf) < PAGE_SIZE:
+                chunk = self._read_at(fd, PAGE_SIZE - len(buf), offset + len(buf))
+                if not chunk:
+                    raise DiskManagerError(
+                        f"unexpected end of file reading page {page_id}"
+                    )
+                buf += chunk
+            return bytes(buf)
+
+    def write_page(self, page_id: UInt32, data: bytes) -> None:
+        """Overwrite the existing page page_id with data.
+
+        Not fsynced; call :meth:sync when the write must be durable.
+
+        Raises
+        ------
+        DiskManagerError
+            If data is not exactly PAGE_SIZE bytes, page_id has not
+            been allocated, or the file is closed.
+        """
+        if len(data) != PAGE_SIZE:
+            raise DiskManagerError(
+                f"page data must be {PAGE_SIZE} bytes, got {len(data)}"
+            )
+        with self._in_flight() as fd:
+            self._write_all(fd, data, self._offset(fd, page_id))
+
+    def allocate_page(self) -> UInt32:
+        """Allocate a new page at the end of the file and return its ID.
+
+        The page is zero-filled. Reusing freed pages is the caller's job.
+        Not fsynced; call :meth:`sync` before relying on the new size after
+        a crash.
+
+        Raises
+        ------
+        DiskManagerError
+            If the file is closed or page IDs would exceed the UInt32 range.
+        """
+        with self._in_flight() as fd, self._lock:
+            count = self._refresh_page_count(fd)
+            if count > UInt32.MAX:
+                raise DiskManagerError("page ID would exceed the UInt32 range")
+            self._write_all(fd, bytes(PAGE_SIZE), count * PAGE_SIZE)
+            self._known_pages = count + 1
+            return UInt32(count)
+
+    def sync(self) -> None:
+        """Flush file contents and size to stable storage (``fsync``).
+
+        Raises
+        ------
+        DiskManagerError
+            If the file is closed or the flush fails.
+        """
+        with self._in_flight() as fd:
+            os.fsync(fd)
+
+    def close(self) -> None:
+        """Close the file. Safe to call more than once.
+
+        Raises
+        ------
+        DiskManagerError
+            If a read, write, allocation or sync is still running, or the
+            file cannot be closed.
+        """
+        with self._lock:
+            if self._active:
+                raise DiskManagerError("cannot close while I/O is in progress")
+            if self._fd is not None:
+                fd, self._fd = self._fd, None
+                with _os_errors(self._path):
+                    os.close(fd)
+
+    def __enter__(self) -> Self:
+        """Return the manager for use in a ``with`` block."""
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        """Close the file when leaving the ``with`` block."""
+        self.close()
+
+    @contextmanager
+    def _in_flight(self) -> Iterator[int]:
+        # Counts the call as running so close() cannot free the fd under it.
+        with self._lock:
+            fd = self._require_open()
+            self._active += 1
+        try:
+            with _os_errors(self._path):
+                yield fd
+        finally:
+            with self._lock:
+                self._active -= 1
+
+    def _require_open(self) -> int:
+        if self._fd is None:
+            raise DiskManagerError("the database file is closed")
+        return self._fd
+
+    def _refresh_page_count(self, fd: int) -> int:
+        count = os.fstat(fd).st_size // PAGE_SIZE
+        self._known_pages = max(self._known_pages, count)
+        return count
+
+    def _offset(self, fd: int, page_id: UInt32) -> int:
+        # The file only grows while open, so the cached count is a lower
+        # bound: re-read the file size only when an ID is beyond it, which
+        # also picks up pages another handle allocated.
+        if page_id >= self._known_pages:
+            self._refresh_page_count(fd)
+        if not 0 <= page_id < self._known_pages:
+            raise DiskManagerError(
+                f"page {page_id} is out of range (0..{self._known_pages - 1})"
+            )
+        return page_id * PAGE_SIZE
+
+    def _read_at(self, fd: int, size: int, offset: int) -> bytes:
+        if _pread is not None:
+            return _pread(fd, size, offset)
+        with self._lock:  # seek + read must be atomic without pread
+            os.lseek(fd, offset, os.SEEK_SET)
+            return os.read(fd, size)
+
+    def _write_all(self, fd: int, data: bytes, offset: int) -> None:
+        view = memoryview(data)
+        written = 0
+        while written < len(view):
+            if _pwrite is not None:
+                n = _pwrite(fd, view[written:], offset + written)
+            else:
+                with self._lock:
+                    os.lseek(fd, offset + written, os.SEEK_SET)
+                    n = os.write(fd, view[written:])
+            if n <= 0:
+                raise DiskManagerError("write made no progress")
+            written += n
diff --git a/src/seriousdb/exceptions.py b/src/seriousdb/exceptions.py
index d0cba68..2be7743 100644
--- a/src/seriousdb/exceptions.py
+++ b/src/seriousdb/exceptions.py
@@ -41,3 +41,9 @@ class ServiceUnavailableError(ApplicationError):
     """A dependency the application needs is currently not usable."""
 
     default_detail = "The service is temporarily unavailable"
+
+
+class DiskManagerError(ApplicationError):
+    """A raw page read, write or allocation on the database file failed."""
+
+    default_detail = "The database file could not be accessed as fixed-size pages"
diff --git a/tests/test_disk_manager.py b/tests/test_disk_manager.py
new file mode 100644
index 0000000..aed5f10
--- /dev/null
+++ b/tests/test_disk_manager.py
@@ -0,0 +1,244 @@
+from threading import Barrier, Thread
+
+import pytest
+
+from seriousdb import disk_manager
+from seriousdb.disk_manager import DiskManager
+from seriousdb.exceptions import DiskManagerError
+from seriousdb.storage_format import PAGE_SIZE
+from seriousdb.types import UInt32
+
+
+@pytest.fixture
+def path(tmp_path):
+    return tmp_path / "test.sdb"
+
+
+def page(byte: int) -> bytes:
+    return bytes([byte]) * PAGE_SIZE
+
+
+def test_new_file_is_empty_and_first_append_is_page_zero(path):
+    with DiskManager(path) as dm:
+        assert dm.page_count == 0
+        assert dm.allocate_page() == 0
+        assert dm.allocate_page() == 1
+        assert dm.page_count == 2
+        assert path.stat().st_size == 2 * PAGE_SIZE
+
+
+def test_round_trip_and_reopen(path):
+    with DiskManager(path) as dm:
+        for i in range(3):
+            dm.write_page(dm.allocate_page(), page(i + 1))
+        dm.sync()
+    with DiskManager(path) as dm:
+        assert dm.page_count == 3
+        assert [dm.read_page(UInt32(i)) for i in range(3)] == [
+            page(1),
+            page(2),
+            page(3),
+        ]
+
+
+def test_fresh_page_is_zeroed(path):
+    with DiskManager(path) as dm:
+        pid = dm.allocate_page()
+        assert dm.read_page(pid) == bytes(PAGE_SIZE)
+
+
+def test_misaligned_file_rejected(path):
+    path.write_bytes(bytes(PAGE_SIZE + 10))
+    with pytest.raises(DiskManagerError):
+        DiskManager(path)
+    assert path.stat().st_size == PAGE_SIZE + 10  # untouched
+
+
+def test_repair_truncates_only_the_torn_tail(path):
+    path.write_bytes(page(7) + page(8) + b"x" * 100)
+    with DiskManager(path, repair_torn_page=True) as dm:
+        assert dm.page_count == 2
+        assert dm.read_page(UInt32(0)) == page(7)
+        assert dm.read_page(UInt32(1)) == page(8)
+    assert path.stat().st_size == 2 * PAGE_SIZE
+
+
+@pytest.mark.parametrize("size", [PAGE_SIZE - 1, PAGE_SIZE + 1, 0])
+def test_wrong_size_write_rejected(path, size):
+    with DiskManager(path) as dm:
+        dm.allocate_page()
+        with pytest.raises(DiskManagerError):
+            dm.write_page(UInt32(0), bytes(size))
+
+
+@pytest.mark.parametrize("bad", [UInt32(1), UInt32(99), UInt32(UInt32.MAX)])
+def test_out_of_range_page_id_rejected(path, bad):
+    with DiskManager(path) as dm:
+        dm.allocate_page()
+        with pytest.raises(DiskManagerError):
+            dm.read_page(bad)
+        with pytest.raises(DiskManagerError):
+            dm.write_page(bad, bytes(PAGE_SIZE))
+
+
+def test_write_never_grows_file(path):
+    with DiskManager(path) as dm:
+        with pytest.raises(DiskManagerError):
+            dm.write_page(UInt32(0), bytes(PAGE_SIZE))
+        assert dm.page_count == 0
+
+
+def test_in_range_io_does_not_call_fstat(path, monkeypatch):
+    with DiskManager(path) as dm:
+        dm.write_page(dm.allocate_page(), page(1))
+        calls: list[int] = []
+        real = disk_manager.os.fstat
+        monkeypatch.setattr(
+            disk_manager.os, "fstat", lambda fd: calls.append(fd) or real(fd)
+        )
+        dm.read_page(UInt32(0))
+        dm.write_page(UInt32(0), page(2))
+        assert calls == []
+
+
+def test_page_allocated_by_another_handle_is_readable(path):
+    with DiskManager(path) as a, DiskManager(path) as b:
+        a.write_page(a.allocate_page(), page(4))
+        assert b.read_page(UInt32(0)) == page(4)
+        with pytest.raises(DiskManagerError):
+            b.read_page(UInt32(1))
+
+
+def test_close_during_io_raises(path, monkeypatch):
+    with DiskManager(path) as dm:
+        dm.write_page(dm.allocate_page(), page(1))
+        real = dm._read_at
+
+        def read_at(fd, size, offset):
+            with pytest.raises(DiskManagerError):
+                dm.close()
+            return real(fd, size, offset)
+
+        monkeypatch.setattr(dm, "_read_at", read_at)
+        assert dm.read_page(UInt32(0)) == page(1)
+    # The with-block closed it, so no I/O call leaked its in-flight count.
+    with pytest.raises(DiskManagerError):
+        dm.read_page(UInt32(0))
+
+
+def test_failed_io_does_not_block_close(path):
+    dm = DiskManager(path)
+    with pytest.raises(DiskManagerError):
+        dm.read_page(UInt32(5))
+    dm.close()
+
+
+def test_open_failure_is_wrapped(tmp_path):
+    with pytest.raises(DiskManagerError) as info:
+        DiskManager(tmp_path / "missing-dir" / "test.sdb")
+    assert isinstance(info.value.__cause__, OSError)
+
+
+def test_repair_failure_is_wrapped(path, monkeypatch):
+    path.write_bytes(page(1) + b"x")
+
+    def boom(*args):
+        raise OSError("disk on fire")
+
+    monkeypatch.setattr(disk_manager.os, "ftruncate", boom)
+    with pytest.raises(DiskManagerError) as info:
+        DiskManager(path, repair_torn_page=True)
+    assert isinstance(info.value.__cause__, OSError)
+
+
+def test_read_write_sync_os_errors_are_wrapped(path, monkeypatch):
+    def boom(*args):
+        raise OSError("disk on fire")
+
+    with DiskManager(path) as dm:
+        dm.write_page(dm.allocate_page(), page(1))
+        with monkeypatch.context() as m:
+            m.setattr(dm, "_read_at", boom)
+            with pytest.raises(DiskManagerError) as info:
+                dm.read_page(UInt32(0))
+            assert isinstance(info.value.__cause__, OSError)
+        with monkeypatch.context() as m:
+            m.setattr(dm, "_write_all", boom)
+            with pytest.raises(DiskManagerError):
+                dm.write_page(UInt32(0), page(2))
+            with pytest.raises(DiskManagerError):
+                dm.allocate_page()
+        with monkeypatch.context() as m:
+            m.setattr(disk_manager.os, "fsync", boom)
+            with pytest.raises(DiskManagerError):
+                dm.sync()
+        # Failed calls must not leave the in-flight count raised.
+    assert dm.path == path
+
+
+def test_use_after_close_raises(path):
+    dm = DiskManager(path)
+    dm.close()
+    dm.close()  # idempotent
+    for call in (dm.allocate_page, dm.sync, lambda: dm.read_page(UInt32(0))):
+        with pytest.raises(DiskManagerError):
+            call()
+
+
+def test_page_count_sees_other_handles_appends(path):
+    with DiskManager(path) as a, DiskManager(path) as b:
+        a.allocate_page()
+        assert b.page_count == 1
+        assert b.read_page(UInt32(0)) == bytes(PAGE_SIZE)
+
+
+@pytest.mark.skipif(disk_manager._pread is None, reason="needs os.pread")
+def test_short_reads_are_retried(path, monkeypatch):
+    real = disk_manager._pread
+    assert real is not None
+    monkeypatch.setattr(
+        disk_manager, "_pread", lambda fd, n, off: real(fd, min(n, 1000), off)
+    )
+    with DiskManager(path) as dm:
+        dm.write_page(dm.allocate_page(), page(5))
+        assert dm.read_page(UInt32(0)) == page(5)
+
+
+@pytest.mark.skipif(disk_manager._pwrite is None, reason="needs os.pwrite")
+def test_short_writes_are_retried(path, monkeypatch):
+    real = disk_manager._pwrite
+    assert real is not None
+    monkeypatch.setattr(
+        disk_manager, "_pwrite", lambda fd, data, off: real(fd, bytes(data[:1000]), off)
+    )
+    with DiskManager(path) as dm:
+        dm.write_page(dm.allocate_page(), page(9))
+        assert dm.read_page(UInt32(0)) == page(9)
+    assert path.read_bytes() == page(9)
+
+
+@pytest.mark.skipif(disk_manager._pread is None, reason="needs os.pread")
+def test_eof_mid_page_raises(path, monkeypatch):
+    with DiskManager(path) as dm:
+        dm.allocate_page()
+        monkeypatch.setattr(disk_manager, "_pread", lambda fd, n, off: b"")
+        with pytest.raises(DiskManagerError):
+            dm.read_page(UInt32(0))
+
+
+def test_concurrent_allocations_get_unique_ids(path):
+    with DiskManager(path) as dm:
+        ids: list[int] = []
+        start = Barrier(8)
+
+        def work():
+            start.wait()
+            ids.extend(int(dm.allocate_page()) for _ in range(25))
+
+        threads = [Thread(target=work) for _ in range(8)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        assert sorted(ids) == list(range(200))
+        assert dm.page_count == 200