diff --git a/README.md b/README.md
index 86db3e1..d69892c 100644
--- a/README.md
+++ b/README.md
@@ -94,6 +94,18 @@ await kanta.open(create=False)
 With `create=False`, open fails if the database file does not exist or is
 empty.
 
+Read-only mode opens an existing database without locking it or starting the
+background flush task. This is useful for readers that must not block the
+writer or modify the file:
+
+```python
+await kanta.open(readonly=True)
+```
+
+In read-only mode, records are replayed and migrations are applied in memory,
+but transactions and explicit flushes are rejected and the file is never
+created if missing.
+
 ## Fatal Error Handlers
 
 Fatal background write errors can be observed with a decorator:
diff --git a/docs/database.md b/docs/database.md
index 141c6db..daf8edc 100644
--- a/docs/database.md
+++ b/docs/database.md
@@ -118,6 +118,12 @@ reloads, while system operations such as migrations leave it unchanged.
 
 - `await kanta.open()` (default) creates the database file if missing.
 - `await kanta.open(create=False)` fails when the file is missing or empty.
+- `await kanta.open(readonly=True)` opens an existing database read-only.
+  - The file is opened without acquiring a lock and without a background flush
+    task.
+  - Existing records are replayed and migrations are still applied in memory.
+  - Transactions and explicit flushes are rejected.
+  - The file is never created if missing.
 
 ### Callbacks
 
diff --git a/kanta/callbacks.py b/kanta/callbacks.py
index 7dfaa7a..d10c177 100644
--- a/kanta/callbacks.py
+++ b/kanta/callbacks.py
@@ -12,6 +12,7 @@ and receive the value plus an optional ``path`` string.  They return
 from __future__ import annotations
 
 import inspect
+import types
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Annotated, Any, Union, get_args, get_origin
@@ -317,17 +318,13 @@ class CallbackRegistry:
                     f"Allowed: str path, {self._allowed_message('logfmt')}"
                 )
 
-        if sig.return_annotation is inspect.Signature.empty:
-            raise TypeError(
-                f"logfmt callback {callback.__name__} must annotate its "
-                f"return type as str | None"
-            )
-        return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
-        if not self._is_optional_str(return_ann):
-            raise TypeError(
-                f"logfmt callback {callback.__name__} must return str | None, "
-                f"got {return_ann!r}"
-            )
+        if sig.return_annotation is not inspect.Signature.empty:
+            return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
+            if not self._is_optional_str(return_ann):
+                raise TypeError(
+                    f"logfmt callback {callback.__name__} must return str | None, "
+                    f"got {return_ann!r}"
+                )
 
         return _LogFmtFunctionSpec(
             callback=callback,
@@ -416,19 +413,15 @@ class CallbackRegistry:
                 f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
             )
 
-        if resolve_sig.return_annotation is inspect.Signature.empty:
-            raise TypeError(
-                f"logfmt class {cls.__name__}.resolve must annotate its "
-                f"return type as str | None"
-            )
-        return_ann = self._resolve_raw_annotation(
-            resolve_sig.return_annotation, resolve
-        )
-        if not self._is_optional_str(return_ann):
-            raise TypeError(
-                f"logfmt class {cls.__name__}.resolve must return str | None, "
-                f"got {return_ann!r}"
+        if resolve_sig.return_annotation is not inspect.Signature.empty:
+            return_ann = self._resolve_raw_annotation(
+                resolve_sig.return_annotation, resolve
             )
+            if not self._is_optional_str(return_ann):
+                raise TypeError(
+                    f"logfmt class {cls.__name__}.resolve must return str | None, "
+                    f"got {return_ann!r}"
+                )
 
         return _LogFmtClassSpec(cls=cls, inject_params=inject_params, path=path)
 
@@ -516,7 +509,7 @@ class CallbackRegistry:
     @staticmethod
     def _unwrap_optional(ann: Any) -> Any:
         origin = get_origin(ann)
-        if origin is not Union:
+        if origin not in (Union, types.UnionType):
             return ann
         args = [arg for arg in get_args(ann) if arg is not type(None)]
         return args[0] if len(args) == 1 else ann
@@ -524,7 +517,7 @@ class CallbackRegistry:
     @staticmethod
     def _is_optional_str(ann: Any) -> bool:
         origin = get_origin(ann)
-        if origin is not Union:
+        if origin not in (Union, types.UnionType):
             return ann is str
         args = get_args(ann)
         return type(None) in args and any(arg is str for arg in args)
diff --git a/kanta/filelock.py b/kanta/filelock.py
index 4fd4d49..d719c19 100644
--- a/kanta/filelock.py
+++ b/kanta/filelock.py
@@ -34,6 +34,7 @@ if sys.platform == "win32":
     _GENERIC_READ = 0x80000000
     _GENERIC_WRITE = 0x40000000
     _FILE_SHARE_READ = 0x00000001
+    _FILE_SHARE_WRITE = 0x00000002
     _OPEN_EXISTING = 3
     _OPEN_ALWAYS = 4
     _FILE_ATTRIBUTE_NORMAL = 0x80
@@ -91,15 +92,16 @@ else:
 
 
 class LockedFile:
-    """A file opened with an exclusive write lock.
+    """A file opened for read+write with an optional exclusive lock.
 
     Usage::
 
         f = LockedFile()
-        f.open(path)          # open + lock (read+write)
-        content = f.read()    # read entire content
-        f.write(data)         # append data (seeks to end first)
-        f.close()             # release lock + close fd
+        f.open(path)                    # open + lock (read+write)
+        f.open(path, readonly=True)     # open read-only without locking
+        content = f.read()              # read entire content
+        f.write(data)                   # append data (seeks to end first)
+        f.close()                       # release lock + close fd
 
     Unix:    fcntl.flock (advisory) — read-only callers that don't flock are unaffected.
     Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers.
@@ -108,12 +110,13 @@ class LockedFile:
     def __init__(self) -> None:
         self._fd: int | None = None  # Unix fd or Windows HANDLE
 
-    def open(self, path: Path, *, create: bool = False) -> None:
-        """Open *path* for read+write with an exclusive lock.
+    def open(self, path: Path, *, create: bool = False, readonly: bool = False) -> None:
+        """Open *path* and optionally acquire an exclusive lock.
 
         Args:
             path:   File to open and lock.
             create: If True, create the file if it doesn't exist (bootstrap).
+            readonly: If True, open read-only without acquiring a lock.
 
         Raises:
             FileLockError: If the file is locked by another process or not found.
@@ -122,16 +125,16 @@ class LockedFile:
             return  # Already open (idempotent)
 
         if sys.platform == "win32":
-            self._open_win32(path, create)
+            self._open_win32(path, create, readonly)
         else:
-            self._open_unix(path, create)
+            self._open_unix(path, create, readonly)
 
-    def open_and_read(self, path: Path, create: bool = False) -> bytes:
-        """Open *path* with exclusive lock and read all content.
+    def open_and_read(self, path: Path, create: bool = False, readonly: bool = False) -> bytes:
+        """Open *path* and read all content.
 
         Combined operation for efficient use with asyncio.to_thread().
         """
-        self.open(path, create=create)
+        self.open(path, create=create, readonly=readonly)
         return self.read()
 
     def read(self) -> bytes:
@@ -188,20 +191,24 @@ class LockedFile:
 
     # -- Unix ----------------------------------------------------------------
 
-    def _open_unix(self, path: Path, create: bool) -> None:
-        flags = os.O_RDWR | (os.O_CREAT if create else 0)
+    def _open_unix(self, path: Path, create: bool, readonly: bool) -> None:
+        if readonly:
+            flags = os.O_RDONLY
+        else:
+            flags = os.O_RDWR | (os.O_CREAT if create else 0)
         try:
             fd = os.open(path, flags, 0o666)
         except FileNotFoundError:
             _fatal(f"Database file not found: {path.resolve()}", db_path=path)
-        try:
-            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except OSError:
-            os.close(fd)
-            _fatal(
-                f"{path.resolve()}: database already locked by another instance",
-                db_path=path,
-            )
+        if not readonly:
+            try:
+                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except OSError:
+                os.close(fd)
+                _fatal(
+                    f"{path.resolve()}: database already locked by another instance",
+                    db_path=path,
+                )
         self._fd = fd
 
     def _read_unix(self) -> bytes:
@@ -220,12 +227,19 @@ class LockedFile:
 
     # -- Windows -------------------------------------------------------------
 
-    def _open_win32(self, path: Path, create: bool) -> None:
-        disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
+    def _open_win32(self, path: Path, create: bool, readonly: bool) -> None:
+        if readonly:
+            disposition = _OPEN_EXISTING
+            access = _GENERIC_READ
+            share = _FILE_SHARE_READ | _FILE_SHARE_WRITE
+        else:
+            disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
+            access = _GENERIC_READ | _GENERIC_WRITE
+            share = _FILE_SHARE_READ
         handle = _kernel32.CreateFileW(
             str(path),
-            _GENERIC_READ | _GENERIC_WRITE,
-            _FILE_SHARE_READ,
+            access,
+            share,
             None,
             disposition,
             _FILE_ATTRIBUTE_NORMAL,
diff --git a/kanta/kanta.py b/kanta/kanta.py
index 3060359..9f1e80f 100644
--- a/kanta/kanta.py
+++ b/kanta/kanta.py
@@ -138,7 +138,7 @@ class Kanta(Generic[T]):
         """
         return self._impl.mtime
 
-    async def open(self, *, create: bool = True) -> None:
+    async def open(self, *, create: bool = True, readonly: bool = False) -> None:
         """Open the database file and start background persistence.
 
         This loads existing records, applies configured migrations, and starts
@@ -147,6 +147,9 @@ class Kanta(Generic[T]):
         Args:
             create: Whether to create the database file when missing.
                 If False, opening fails when the file does not exist or is empty.
+            readonly: If True, open the database read-only. No lock is acquired,
+                no background flush task is started, and transactions are
+                rejected. The file is not created if missing.
 
         Calling ``open`` more than once on the same instance is not allowed.
 
@@ -154,7 +157,7 @@ class Kanta(Generic[T]):
             kanta.exceptions.DatabaseError: If replay or decoding fails.
             kanta.exceptions.DataIntegrityError: If the instance is already open.
         """
-        await self._impl.open(create=create)
+        await self._impl.open(create=create, readonly=readonly)
 
     async def __aenter__(self) -> Kanta[T]:
         """Enter async context manager and open the database.
diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py
index a171fb7..eed0343 100644
--- a/kanta/kantaimpl.py
+++ b/kanta/kantaimpl.py
@@ -43,6 +43,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         self.in_transaction = False
         self.transaction_snapshot: dict[str, Any] | None = None
         self.opened = False
+        self.readonly = False
         self.bootstrap_action = "bootstrap"
         self.bootstrap_user: str | None = None
         self.bootstrap_mtime: bool | datetime = True
@@ -75,7 +76,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         """Register one transaction logfmt callback."""
         self.callback_registry.register("logfmt", callback, path=path)
 
-    async def open(self, *, create: bool = True) -> None:
+    async def open(self, *, create: bool = True, readonly: bool = False) -> None:
         """Open the database: load from disk, apply migrations, start background task."""
         if self.opened:
             raise DataIntegrityError(
@@ -84,12 +85,17 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 action="open",
             )
 
+        self.readonly = readonly
         existed_before_open = self.filename.exists()
 
+        # Read-only mode never creates the file.
+        open_create = create and not readonly
+
         content = await asyncio.to_thread(
             self.file.open_and_read,
             self.filename,
-            create=create,
+            create=open_create,
+            readonly=readonly,
         )
 
         if not create and (not existed_before_open or not content):
@@ -149,12 +155,22 @@ class KantaImpl(PersistenceMixin, Generic[T]):
             self.version = rr.version
             self.mtime = rr.m
             normalized = struct_to_dict(self.data, serializer=self.serializer)
-            self.queue_change("migrate:msgspec", normalized, mtime=False)
+            if self.readonly:
+                self.statedict = copy.deepcopy(normalized)
+            else:
+                self.queue_change("migrate:msgspec", normalized, mtime=False)
             self.snapshot.ts = (
                 datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
                 if rr.last_snapshot_mtime is not None
                 else None
             )
+        elif self.readonly:
+            self.file.close()
+            raise DataIntegrityError(
+                "Cannot open empty database in read-only mode",
+                db_path=self.filename,
+                action="open",
+            )
         elif self.callback_registry.has("bootstrap"):
             try:
                 await self.callback_registry.invoke(
@@ -179,7 +195,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
         self.opened = True
 
-        self.background_task = asyncio.create_task(self._background_loop())
+        if not self.readonly:
+            self.background_task = asyncio.create_task(self._background_loop())
 
     async def close(self) -> None:
         """Stop the background task, flush pending changes, and release the file lock."""
@@ -196,7 +213,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
         # Always run a final flush in case the background task never reached
         # its cancellation handler.
-        await self.flush()
+        if not self.readonly:
+            await self.flush()
 
         self.file.close()
         self.opened = False
diff --git a/kanta/persistence.py b/kanta/persistence.py
index 9b7d000..1feed89 100644
--- a/kanta/persistence.py
+++ b/kanta/persistence.py
@@ -39,6 +39,7 @@ class PersistenceMixin:
     flush_interval: float
     version: int
     opened: bool
+    readonly: bool
     mtime: datetime | None
 
     def __init__(self, **kwargs: Any) -> None:
@@ -68,6 +69,8 @@ class PersistenceMixin:
 
     async def _background_loop(self) -> None:
         """Background task that periodically flushes changes to disk."""
+        if self.readonly:
+            return
         while True:
             try:
                 await asyncio.sleep(self.flush_interval)
@@ -160,6 +163,13 @@ class PersistenceMixin:
                 action="flush_sync",
             )
 
+        if self.readonly:
+            raise DataIntegrityError(
+                "Cannot flush in read-only mode",
+                db_path=self.filename,
+                action="flush_sync",
+            )
+
         if self.flush_failed:
             return
 
@@ -207,6 +217,13 @@ class PersistenceMixin:
                 action="flush",
             )
 
+        if self.readonly:
+            raise DataIntegrityError(
+                "Cannot flush in read-only mode",
+                db_path=self.filename,
+                action="flush",
+            )
+
         if self.flush_failed:
             return
 
diff --git a/kanta/transaction.py b/kanta/transaction.py
index d5a9559..49fc509 100644
--- a/kanta/transaction.py
+++ b/kanta/transaction.py
@@ -24,6 +24,13 @@ def transaction(
     mtime: bool | datetime = True,
 ):
     """Wrap writes in a transaction and yield the live db object."""
+    if impl.readonly:
+        raise DataIntegrityError(
+            "Cannot start transaction in read-only mode",
+            db_path=impl.filename,
+            action=action,
+        )
+
     if impl.in_transaction:
         raise RuntimeError(
             "Nested or simultaneous transactions are not supported "
diff --git a/tests/test_callbacks.py b/tests/test_callbacks.py
index 9fe462d..014fc46 100644
--- a/tests/test_callbacks.py
+++ b/tests/test_callbacks.py
@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Optional, Union
 
 import pytest
 
@@ -49,16 +49,60 @@ def test_logfmt_requires_value_annotation(tmp_path, format_config):
             return None
 
 
-def test_logfmt_requires_return_annotation(tmp_path, format_config):
+def test_logfmt_allows_missing_return_annotation(tmp_path, format_config):
     kanta = make_kanta(tmp_path / "test.db", Data, format_config)
 
-    with pytest.raises(TypeError, match="must annotate its return"):
+    @kanta.logfmt
+    def resolve_names(value: str, current: DictPost):
+        return None
 
-        @kanta.logfmt
-        def resolve_names(value: str, current: DictPost):
+
+def test_logfmt_class_allows_missing_return_annotation(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    class UserLogFmt(LogFmt):
+        def resolve(self, value: str, path: str):
             return None
 
 
+# fmt: off
+def test_logfmt_accepts_optional_return_typing_forms(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    def resolve_optional(value: str) -> Optional[str]:  # noqa: UP007
+        return value
+
+    @kanta.logfmt
+    def resolve_union(value: str) -> Union[str, None]:  # noqa: UP007
+        return value
+
+    @kanta.logfmt
+    def resolve_pipe(value: "str") -> "str | None":
+        return value
+
+
+def test_logfmt_class_accepts_optional_return_typing_forms(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    class OptionalStyle(LogFmt):
+        def resolve(self, value: str, path: str) -> Optional[str]:  # noqa: UP007
+            return value
+
+    @kanta.logfmt
+    class UnionStyle(LogFmt):
+        def resolve(self, value: str, path: str) -> Union[str, None]:  # noqa: UP007
+            return value
+
+    @kanta.logfmt
+    class StringStyle(LogFmt):
+        def resolve(self, value: "str", path: "str") -> "str | None":
+            return value
+# fmt: on
+
+
 def test_logfmt_rejects_async_callback(tmp_path, format_config):
     kanta = make_kanta(tmp_path / "test.db", Data, format_config)
 
diff --git a/tests/test_readonly.py b/tests/test_readonly.py
new file mode 100644
index 0000000..b84ed8e
--- /dev/null
+++ b/tests/test_readonly.py
@@ -0,0 +1,153 @@
+"""Tests for Kanta read-only mode."""
+
+import pytest
+
+from kanta.exceptions import DataIntegrityError, FileLockError
+from kanta.serialization import struct_to_dict
+
+from .support import (
+    Data,
+    EvolvableDataV2,
+    fixed_change,
+    make_kanta,
+    make_migrations_module,
+    seed_single_change,
+)
+
+
+@pytest.mark.asyncio
+async def test_readonly_opens_existing_database(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 7}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    assert isinstance(kanta.data, Data)
+    assert kanta.data.counter == 7
+    assert kanta._impl.readonly is True
+    assert kanta._impl.background_task is None
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_missing_file_fails(tmp_path, format_config):
+    path = tmp_path / "missing.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(FileLockError):
+        await kanta.open(readonly=True)
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_readonly_empty_file_fails(tmp_path, format_config):
+    path = tmp_path / "empty.db"
+    path.touch()
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(DataIntegrityError, match="empty"):
+        await kanta.open(readonly=True)
+
+
+@pytest.mark.asyncio
+async def test_readonly_transaction_fails(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    with pytest.raises(DataIntegrityError, match="read-only"):
+        with kanta.transaction(action="inc") as data:
+            data.counter = 2
+
+    # In-memory state must remain unchanged.
+    assert kanta.data.counter == 1
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_flush_fails(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    with pytest.raises(DataIntegrityError, match="read-only"):
+        await kanta.flush()
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_create_true_does_not_create_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(FileLockError):
+        await kanta.open(create=True, readonly=True)
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_readonly_does_not_persist_changes(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+    original_content = path.read_bytes()
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+    await kanta.close()
+
+    assert path.read_bytes() == original_content
+
+
+@pytest.mark.asyncio
+async def test_readonly_runs_migrations(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(
+        path,
+        fixed_change("seed", {"counter": 1}, version=0),
+        format_config,
+    )
+
+    def migrate_v1(data, ctx):
+        data.setdefault("enabled", True)
+
+    migrations = make_migrations_module("readonly_migrations", "migrate_v1", migrate_v1)
+
+    kanta = make_kanta(path, EvolvableDataV2, format_config, migrations=migrations)
+    await kanta.open(readonly=True)
+
+    assert kanta.data.counter == 1
+    # Migration ran in memory even though no change was persisted.
+    assert struct_to_dict(kanta.data, serializer=kanta._impl.serializer) == {
+        "counter": 1,
+        "enabled": True,
+    }
+    assert not kanta._impl.pending_changes
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readwrite_and_readonly_can_open_together(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    rw = make_kanta(path, Data, format_config)
+    await rw.open()
+
+    ro = make_kanta(path, Data, format_config)
+    await ro.open(readonly=True)
+
+    assert rw.data.counter == 1
+    assert ro.data.counter == 1
+
+    await ro.close()
+    await rw.close()