diff --git a/trovedb/db.py b/trovedb/db.py
index 665e354..935501e 100644
--- a/trovedb/db.py
+++ b/trovedb/db.py
@@ -86,6 +86,13 @@ class Sqlite3Trove:
     # CRUD operations
     # ------------------------------------------------------------------
 
+    def get_object_type(self, object_id: int) -> str | None:
+        """Return the type column for an object, or None if not found."""
+        row = self._con.execute(
+            "SELECT type FROM objects WHERE id = ?", (object_id,)
+        ).fetchone()
+        return row["type"] if row else None
+
     def read_object(self, object_id: int) -> bytes | None:
         """Return raw data for a blob object, or None if not found."""
         row = self._con.execute(
@@ -104,6 +111,14 @@ class Sqlite3Trove:
             return None
         return bytes(row["value"]) if row["value"] is not None else b""
 
+    def write_metadata(self, object_id: int, key: str, value: bytes) -> None:
+        """Upsert a metadata row. db.py has no write_metadata, so we go direct."""
+        self._con.execute(
+            "INSERT OR REPLACE INTO metadata (id, key, value) VALUES (?, ?, ?)",
+            (object_id, key, value),
+        )
+        self._con.commit()
+
     def _write_object(self, data: bytes, dtype: str, object_id: int | None = None) -> int:
         """
         Insert or replace an object. Returns the id.
@@ -117,6 +132,7 @@ class Sqlite3Trove:
                 (dtype, data, modified)
             )
             self._con.commit()
+            assert cur.lastrowid is not None
             return cur.lastrowid
         else:
             self._con.execute(
@@ -126,12 +142,16 @@ class Sqlite3Trove:
             self._con.commit()
             return object_id
 
-    def write_blob(self, data: bytes, existing_id: int | None = None) -> int:
+    def write_blob(self, data: bytes, object_id: int | None = None) -> int:
         """
         Insert or replace a blob. Returns the id.
-        Pass existing_id to update an existing object.
+        Pass object_id to update an existing object.
         """
-        return self._write_object(data, "blob", existing_id)
+        return self._write_object(data, "blob", object_id)
+
+    def write_tree(self, data: bytes, object_id: int | None = None) -> int:
+        """Write a tree-typed object. Returns the assigned id."""
+        return self._write_object(data, "tree", object_id)
 
     def delete_object(self, object_id: int) -> bool:
         """
diff --git a/trovedb/fuse/__main__.py b/trovedb/fuse/__main__.py
new file mode 100644
index 0000000..f0c7381
--- /dev/null
+++ b/trovedb/fuse/__main__.py
@@ -0,0 +1,14 @@
+from . import server
+from trovedb import trovedb
+from argparse import ArgumentParser
+
+def main():
+    parser = ArgumentParser()
+    parser.add_argument("db", help="Path to the database file")
+    parser.add_argument("mountpoint", help="Path to the mount point")
+
+    args = parser.parse_args()
+    server.serve(trovedb.TroveImpl.open(args.db, create=True), args.mountpoint)
+
+if __name__ == '__main__':
+    main()
diff --git a/trovedb/fuse/server.py b/trovedb/fuse/server.py
new file mode 100644
index 0000000..8bcc9d6
--- /dev/null
+++ b/trovedb/fuse/server.py
@@ -0,0 +1,240 @@
+"""
+fuse.py — FUSE filesystem layer for Trove, backed by pyfuse3 + trio.
+
+Blob objects are exposed as regular files.
+Tree objects are exposed as directories.
+
+Inode numbers map 1:1 to Trove object IDs (NODE_ROOT_ID == pyfuse3 root inode == 1).
+
+Entry point: serve(trove, mountpoint)
+"""
+
+import errno
+import os
+import stat
+import time
+
+import pyfuse3
+import trio
+
+from trovedb.trove import Trove, Note, Tree as TroveTree, TreeNote, Blob as TroveBlob
+
+
+class TroveFuseOps(pyfuse3.Operations):
+
+    enable_writeback_cache = False
+
+    def __init__(self, trove: Trove):
+        super().__init__()
+        self._trove = trove
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    def _note_or_error(self, inode: int):
+        note = self._trove.get_raw_note(inode)
+        if note is None:
+            raise pyfuse3.FUSEError(errno.ENOENT)
+        return note
+
+    def _make_attr(self, inode: int, is_tree: bool, size: int = 0) -> pyfuse3.EntryAttributes:
+        attr = pyfuse3.EntryAttributes()
+        attr.st_ino = pyfuse3.InodeT(inode)
+        attr.st_nlink = 1
+        attr.st_uid = os.getuid()
+        attr.st_gid = os.getgid()
+        now_ns = int(time.time() * 1e9)
+        attr.st_atime_ns = now_ns
+        attr.st_mtime_ns = now_ns
+        attr.st_ctime_ns = now_ns
+        attr.generation = 0
+        attr.entry_timeout = 5.0
+        attr.attr_timeout = 5.0
+        if is_tree:
+            attr.st_mode = stat.S_IFDIR | 0o755
+            attr.st_size = 0
+            attr.st_blksize = 512
+            attr.st_blocks = 0
+        else:
+            attr.st_mode = stat.S_IFREG | 0o644
+            attr.st_size = size
+            attr.st_blksize = 512
+            attr.st_blocks = (size + 511) // 512
+        return attr
+
+    def _attr_for_note(self, note: Note) -> pyfuse3.EntryAttributes:
+        size = 0
+        is_tree = True
+        if isinstance(note, TroveBlob):
+            size = len(note.read())
+            is_tree = False
+        return self._make_attr(note.object_id, is_tree, size)
+
+    # ------------------------------------------------------------------
+    # Stat / lookup
+    # ------------------------------------------------------------------
+
+    async def getattr(self, inode: int, ctx=None) -> pyfuse3.EntryAttributes:
+        note = self._note_or_error(inode)
+        return self._attr_for_note(note)
+
+    async def lookup(self, parent_inode: int, name: bytes, ctx=None) -> pyfuse3.EntryAttributes:
+        parent = self._note_or_error(parent_inode)
+        if not isinstance(parent, TroveTree):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        entries = parent.list()
+        name_str = name.decode()
+        if name_str not in entries:
+            raise pyfuse3.FUSEError(errno.ENOENT)
+        child = self._trove.get_raw_note(entries[name_str])
+        if child is None:
+            raise pyfuse3.FUSEError(errno.ENOENT)
+        return self._attr_for_note(child)
+
+    async def setattr(self, inode: int, attr, fields, fh, ctx) -> pyfuse3.EntryAttributes:
+        note = self._note_or_error(inode)
+        if fields.update_size and not isinstance(note, TroveTree):
+            current = note.read()
+            new_size = attr.st_size
+            if new_size < len(current):
+                note.write(current[:new_size])
+            elif new_size > len(current):
+                note.write(current + b"\x00" * (new_size - len(current)))
+        return self._attr_for_note(note)
+
+    async def forget(self, inode_list) -> None:
+        pass
+
+    # ------------------------------------------------------------------
+    # Directory ops
+    # ------------------------------------------------------------------
+
+    async def opendir(self, inode: int, ctx) -> pyfuse3.FileHandleT:
+        note = self._note_or_error(inode)
+        if not isinstance(note, TroveTree):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        return pyfuse3.FileHandleT(inode)
+
+    async def readdir(self, fh: int, start_id: int, token) -> None:
+        note = self._note_or_error(fh)
+        if not isinstance(note, TroveTree):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        entries = list(note.list().items())  # [(name, object_id), ...]
+        for idx, (name, child_id) in enumerate(entries):
+            if idx < start_id:
+                continue
+            child = self._trove.get_raw_note(child_id)
+            if child is None:
+                continue
+            attr = self._attr_for_note(child)
+            if not pyfuse3.readdir_reply(token, name.encode(), attr, idx + 1):
+                break
+
+    async def releasedir(self, fh: int) -> None:
+        pass
+
+    async def mkdir(self, parent_inode: int, name: bytes, mode: int, ctx) -> pyfuse3.EntryAttributes:
+        parent = self._note_or_error(parent_inode)
+        if not isinstance(parent, TreeNote):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        name_str = name.decode()
+        if name_str in parent.list():
+            raise pyfuse3.FUSEError(errno.EEXIST)
+        new_tree: TreeNote = parent.mkdir(name_str)
+        return self._make_attr(new_tree.object_id, True, 0)
+
+    async def rmdir(self, parent_inode: int, name: bytes, ctx) -> None:
+        parent = self._note_or_error(parent_inode)
+        if not isinstance(parent, TroveTree):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        name_str = name.decode()
+        entries = parent.list()
+        if name_str not in entries:
+            raise pyfuse3.FUSEError(errno.ENOENT)
+        child = self._trove.get_raw_note(entries[name_str])
+        if child is None:
+            raise pyfuse3.FUSEError(errno.ENOENT)
+        if not isinstance(child, TroveTree):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        if child.list():
+            raise pyfuse3.FUSEError(errno.ENOTEMPTY)
+        parent.unlink(name_str)
+
+    # ------------------------------------------------------------------
+    # File ops
+    # ------------------------------------------------------------------
+
+    async def open(self, inode: int, flags, ctx) -> pyfuse3.FileInfo:
+        note = self._note_or_error(inode)
+        if isinstance(note, TroveTree):
+            raise pyfuse3.FUSEError(errno.EISDIR)
+        return pyfuse3.FileInfo(fh=pyfuse3.FileHandleT(inode))
+
+    async def create(self, parent_inode: int, name: bytes, mode: int, flags, ctx) -> tuple:
+        parent = self._note_or_error(parent_inode)
+        if not isinstance(parent, TroveTree):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        name_str = name.decode()
+        if name_str in parent.list():
+            raise pyfuse3.FUSEError(errno.EEXIST)
+        blob = self._trove.create_blob(b"")
+        parent.link(name_str, blob)
+        attr = self._make_attr(blob.object_id, False, 0)
+        return pyfuse3.FileInfo(fh=pyfuse3.FileHandleT(blob.object_id)), attr
+
+    async def read(self, fh: int, offset: int, length: int) -> bytes:
+        note = self._note_or_error(fh)
+        return note.read()[offset:offset + length]
+
+    async def write(self, fh: int, offset: int, data: bytes) -> int:
+        note = self._note_or_error(fh)
+        existing = note.read()
+        if offset > len(existing):
+            existing = existing + b"\x00" * (offset - len(existing))
+        note.write(existing[:offset] + data + existing[offset + len(data):])
+        return len(data)
+
+    async def release(self, fh: int) -> None:
+        pass
+
+    async def unlink(self, parent_inode: int, name: bytes, ctx) -> None:
+        parent = self._note_or_error(parent_inode)
+        if not isinstance(parent, TroveTree):
+            raise pyfuse3.FUSEError(errno.ENOTDIR)
+        name_str = name.decode()
+        entries = parent.list()
+        if name_str not in entries:
+            raise pyfuse3.FUSEError(errno.ENOENT)
+        child = self._trove.get_raw_note(entries[name_str])
+        if child is None:
+            raise pyfuse3.FUSEError(errno.ENOENT)
+        if isinstance(child, TroveTree):
+            raise pyfuse3.FUSEError(errno.EISDIR)
+        parent.unlink(name_str)
+
+
+# ------------------------------------------------------------------
+# Serve
+# ------------------------------------------------------------------
+
+async def _run(ops: TroveFuseOps, mountpoint: str) -> None:
+    options = set(pyfuse3.default_options)
+    options.add("fsname=trove")
+    pyfuse3.init(ops, mountpoint, options)
+    try:
+        await pyfuse3.main()
+    finally:
+        pyfuse3.close()
+
+
+def serve(trove: Trove, mountpoint: str) -> None:
+    """
+    Mount a Trove store at mountpoint and serve until KeyboardInterrupt.
+    Runs a trio event loop internally.
+    """
+    ops = TroveFuseOps(trove)
+    try:
+        trio.run(_run, ops, mountpoint)
+    except KeyboardInterrupt:
+        pass
diff --git a/trovedb/trove.py b/trovedb/trove.py
index 12fc70b..d358d74 100644
--- a/trovedb/trove.py
+++ b/trovedb/trove.py
@@ -47,6 +47,10 @@ class Tree(Protocol):
         """Create a new Tree with the given name."""
         ...
 
+    def list(self) -> dict[str, int]:
+        """Return all entries as {name: object_id}."""
+        ...
+
 class BlobNote(Note, Blob):
     """Blob Note"""
 
diff --git a/trovedb/trovedb.py b/trovedb/trovedb.py
new file mode 100644
index 0000000..0f57c7f
--- /dev/null
+++ b/trovedb/trovedb.py
@@ -0,0 +1,144 @@
+"""
+trovedb.py — Concrete implementation of Trove protocols backed by Sqlite3Trove.
+
+Implements BlobNote, TreeNote, and Trove protocols defined in trove.py.
+Depends on db.py (Sqlite3Trove) and tree.py (Tree) for storage and
+tree serialization respectively.
+"""
+
+from typing import Optional, Self
+from pathlib import Path
+
+from .db import Sqlite3Trove
+from .tree import Tree as TreeData
+from .trove import NODE_ROOT_ID, Note, Trove, TreeNote, BlobNote
+
+
+class NoteImpl(Note):
+    """Concrete not implementation"""
+
+    def __init__(self, db: Sqlite3Trove, object_id: int):
+        self._db = db
+        self._object_id = object_id
+
+    # Note protocol
+    @property
+    def object_id(self) -> int:
+        return self._object_id
+
+    def get_raw_metadata(self, key: str) -> Optional[bytes]:
+        return self._db.read_metadata(self._object_id, key)
+
+    def set_raw_metadata(self, key: str, value: bytes) -> None:
+        self._db.write_metadata(self._object_id, key, value)
+
+
+class BlobNoteImpl(NoteImpl, BlobNote):
+    """Concrete BlobNote: a blob object in the store with metadata access."""
+
+    # Blob protocol
+    def read(self) -> bytes:
+        data = self._db.read_object(self._object_id)
+        return data if data is not None else b""
+
+    def write(self, data: bytes) -> None:
+        self._db.write_blob(data, self._object_id)
+
+
+class TreeNoteImpl(NoteImpl, TreeNote):
+    """Concrete TreeNote: a tree object in the store with metadata access."""
+
+    def _read_tree(self) -> TreeData:
+        data = self._db.read_object(self._object_id)
+        return TreeData(data if data else None)
+
+    def _flush_tree(self, tree: TreeData) -> None:
+        self._db.write_tree(tree.serialize(), self._object_id)
+
+    # Tree protocol
+    def link(self, name: str, note: Note) -> None:
+        """Link name to an existing note (blob or tree)."""
+        tree = self._read_tree()
+        tree.set_entry(name, note.object_id)
+        self._flush_tree(tree)
+
+    def unlink(self, name: str) -> None:
+        """Remove an entry by name. Raises KeyError if not found."""
+        tree = self._read_tree()
+        tree.rm_entry(name)
+        self._flush_tree(tree)
+
+    def mkdir(self, name: str) -> 'TreeNoteImpl':
+        """Create a new empty tree, link it under name, and return it."""
+
+        # Create the new node
+        new_id = self._db.write_tree(TreeData().serialize())
+        tree = TreeNoteImpl(self._db, new_id)
+
+        # Update our node
+        self.link(name, tree)
+        return tree
+
+    def list(self) -> dict[str, int]:
+        """Return all entries as {name: object_id}."""
+        return self._read_tree().list()
+
+
+# ---------------------------------------------------------------------------
+# Trove
+# ---------------------------------------------------------------------------
+
+class TroveImpl:
+    """
+    Concrete Trove: top-level API backed by a Sqlite3Trove database.
+
+    Use TroveImpl.open() to get an instance.
+    """
+
+    def __init__(self, db: Sqlite3Trove):
+        self._db = db
+
+    @classmethod
+    def open(cls, path: str | Path, create: bool = False) -> "TroveImpl":
+        db = Sqlite3Trove.open(path, create=create)
+        trove = cls(db)
+        if create:
+            # Root was written as a blob by Sqlite3Trove.open(); fix its type.
+            db._con.execute(
+                "UPDATE objects SET type = 'tree' WHERE id = ?", (NODE_ROOT_ID,)
+            )
+            db._con.commit()
+        return trove
+
+    def close(self) -> None:
+        self._db.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        self.close()
+
+    # Trove protocol
+    def get_raw_note(self, note_id: int) -> Optional[Note]:
+        """Return a BlobNote or TreeNote for the given id, or None if not found."""
+        ot = self._db.get_object_type(note_id)
+        if ot is None:
+            return None
+        if ot == "blob":
+            return BlobNoteImpl(self._db, note_id)
+        if ot == "tree":
+            return TreeNoteImpl(self._db, note_id)
+        raise ValueError(f"Unknown object type '{ot}' for id {note_id}")
+
+    def create_blob(self, data: bytes | None = None) -> BlobNote:
+        """Create a new blob object and return a BlobNote for it."""
+        obj_id = self._db.write_blob(data or b"")
+        return BlobNoteImpl(self._db, obj_id)
+
+    def get_root(self) -> TreeNote:
+        """Return the root TreeNote (always id=NODE_ROOT_ID)."""
+        return TreeNoteImpl(self._db, NODE_ROOT_ID)
+
+def open_trove(path: str | Path, create: bool = False) -> Trove:
+    return TroveImpl.open(path, create=create)