clean up

.github/workflows/update-flake-lock.yml

@@ -8,7 +8,6 @@ jobs:
   lockfile:
     runs-on: self-hosted
     permissions:
-      actions: write
       contents: write
       pull-requests: write
     steps:

flake.lock

@@ -8,11 +8,11 @@
       },
       "locked": {
         "dir": "pkgs/firefox-addons",
-        "lastModified": 1781270820,
-        "narHash": "sha256-KuDdN/p4UtqUb1wnjo8a/YougmRXEaKvUoGHkYnuAq0=",
+        "lastModified": 1780733803,
+        "narHash": "sha256-QBJPq12P1DAXFGezoEJaSO/xPUrPlnaI3ddSaMG2JpM=",
         "owner": "rycee",
         "repo": "nur-expressions",
-        "rev": "5f4259c0c832a93e13c5ec481d3318ce1394a8f9",
+        "rev": "c80b0aa94392c5f3612ac797108f6d952752036d",
         "type": "gitlab"
       },
       "original": {
@@ -29,11 +29,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1781305496,
-        "narHash": "sha256-g8Vv4Qfc7n+lgov97REu3X6BeJtvYY0hlSUZR1GrGQQ=",
+        "lastModified": 1780679734,
+        "narHash": "sha256-KmRNvpNOb7QEORa06bVgjW9kITcx0VhsI7w0vhmZyD8=",
         "owner": "nix-community",
         "repo": "home-manager",
-        "rev": "c87a39aa979acc4848016d2220c6238390d84779",
+        "rev": "b2b7db486e06e098711dc291bb25db82850e1d16",
         "type": "github"
       },
       "original": {
@@ -47,11 +47,11 @@
         "nixpkgs": "nixpkgs"
       },
       "locked": {
-        "lastModified": 1781168557,
-        "narHash": "sha256-LOnLQ2tpYF9gqIDDr3+j3DbpJJr/QCH6zPRT2GzEUOE=",
+        "lastModified": 1780310866,
+        "narHash": "sha256-fPBRVf6A5xlACYcOI59shGrjURuvwu0lRsDoSCEXt/I=",
         "owner": "nixos",
         "repo": "nixos-hardware",
-        "rev": "6358ff76821101c178e3ab4919a62799bfe3652e",
+        "rev": "4ed851c979641e28597a05086332d75cdc9e395f",
         "type": "github"
       },
       "original": {
@@ -76,11 +76,11 @@
     },
     "nixpkgs-master": {
       "locked": {
-        "lastModified": 1781306963,
-        "narHash": "sha256-qWZ+XEwsf8pN4DnYyyLbmAyj1a74gOO/VNCff44MuB4=",
+        "lastModified": 1780798858,
+        "narHash": "sha256-4KLc5ZMjfMQosXA2JasUgZTk3i+c/i1zMH4custtmI0=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "1eee1a273e2e3fdc3db5b1a4a66f06cd5749db69",
+        "rev": "92840095e65b9970125843175f4be974b71a92ad",
         "type": "github"
       },
       "original": {
@@ -108,11 +108,11 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1781074563,
-        "narHash": "sha256-md8WlXOlfnIeHeOScMTTHFyf2d6iaTwPl2apR5EQ3P4=",
+        "lastModified": 1780243769,
+        "narHash": "sha256-x5UQuRsH3MqI0U9afaXSNqzTPSeZlRLvFAav2Ux1pNw=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "9ae611a455b90cf061d8f332b977e387bda8e1ca",
+        "rev": "331800de5053fcebacf6813adb5db9c9dca22a0c",
         "type": "github"
       },
       "original": {

frontend/.gitignore

@@ -1,24 +0,0 @@
-# Logs
-logs
-*.log
-npm-debug.log*
-yarn-debug.log*
-yarn-error.log*
-pnpm-debug.log*
-lerna-debug.log*
-
-node_modules
-dist
-dist-ssr
-*.local
-
-# Editor directories and files
-.vscode/*
-!.vscode/extensions.json
-.idea
-.DS_Store
-*.suo
-*.ntvs*
-*.njsproj
-*.sln
-*.sw?

python/alembic/richie/versions/2026_06_03-adding_audiobook_libreary_metadata_d7864d1ffc17.py

@@ -0,0 +1,93 @@
+"""adding audiobook libreary metadata.
+
+Revision ID: d7864d1ffc17
+Revises: c8a794340928
+Create Date: 2026-06-03 20:24:09.200837
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+
+from python.orm import RichieBase
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "d7864d1ffc17"
+down_revision: str | None = "c8a794340928"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+schema = RichieBase.schema_name
+
+
+def upgrade() -> None:
+    """Upgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "audiobook_author",
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_audiobook_author")),
+        sa.UniqueConstraint("name", name=op.f("uq_audiobook_author_name")),
+        schema=schema,
+    )
+    op.create_table(
+        "audiobook_series",
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("author_id", sa.Integer(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["author_id"],
+            [f"{schema}.audiobook_author.id"],
+            name=op.f("fk_audiobook_series_author_id_audiobook_author"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_audiobook_series")),
+        sa.UniqueConstraint("author_id", "name", name=op.f("uq_audiobook_series_author_id")),
+        schema=schema,
+    )
+    op.create_table(
+        "audiobook",
+        sa.Column("title", sa.String(), nullable=False),
+        sa.Column("author_id", sa.Integer(), nullable=False),
+        sa.Column("series_id", sa.Integer(), nullable=True),
+        sa.Column("series_index", sa.Integer(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["author_id"],
+            [f"{schema}.audiobook_author.id"],
+            name=op.f("fk_audiobook_author_id_audiobook_author"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["series_id"],
+            [f"{schema}.audiobook_series.id"],
+            name=op.f("fk_audiobook_series_id_audiobook_series"),
+            ondelete="SET NULL",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_audiobook")),
+        schema=schema,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("audiobook", schema=schema)
+    op.drop_table("audiobook_series", schema=schema)
+    op.drop_table("audiobook_author", schema=schema)
+    # ### end Alembic commands ###

python/alembic/richie/versions/2026_06_10-add_ebook_search_tables_2db132cace1a.py

@@ -0,0 +1,200 @@
+"""add ebook search tables.
+
+Revision ID: 2db132cace1a
+Revises: b3c60cc5beb5
+Create Date: 2026-06-10 22:10:54.379159
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pgvector
+import sqlalchemy as sa
+from alembic import op
+
+from python.orm import RichieBase
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "2db132cace1a"
+down_revision: str | None = "b3c60cc5beb5"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+schema = RichieBase.schema_name
+
+
+def upgrade() -> None:
+    """Upgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "ebook_embedding_model",
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("dimension", sa.Integer(), nullable=False),
+        sa.Column("is_default", sa.Boolean(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_ebook_embedding_model")),
+        sa.UniqueConstraint("name", name=op.f("uq_ebook_embedding_model_name")),
+        schema=schema,
+    )
+    op.create_table(
+        "ebook_source",
+        sa.Column("title", sa.String(), nullable=False),
+        sa.Column("author", sa.String(), nullable=True),
+        sa.Column("language", sa.String(), nullable=True),
+        sa.Column("publisher", sa.String(), nullable=True),
+        sa.Column("identifier", sa.String(), nullable=True),
+        sa.Column("file_path", sa.String(), nullable=False),
+        sa.Column("file_sha256", sa.String(length=64), nullable=False),
+        sa.Column("file_mtime", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("file_size", sa.BigInteger(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_ebook_source")),
+        sa.UniqueConstraint("file_path", name=op.f("uq_ebook_source_file_path")),
+        sa.UniqueConstraint("file_sha256", name=op.f("uq_ebook_source_file_sha256")),
+        schema=schema,
+    )
+    op.create_table(
+        "ebook_chapter",
+        sa.Column("source_id", sa.Integer(), nullable=False),
+        sa.Column("spine_index", sa.Integer(), nullable=False),
+        sa.Column("title", sa.String(), nullable=True),
+        sa.Column("href", sa.String(), nullable=True),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["source_id"],
+            [f"{schema}.ebook_source.id"],
+            name=op.f("fk_ebook_chapter_source_id_ebook_source"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_ebook_chapter")),
+        sa.UniqueConstraint("source_id", "spine_index", name=op.f("uq_ebook_chapter_source_id")),
+        schema=schema,
+    )
+    op.create_table(
+        "ebook_chunk",
+        sa.Column("source_id", sa.Integer(), nullable=False),
+        sa.Column("chapter_id", sa.Integer(), nullable=True),
+        sa.Column("chunk_index", sa.Integer(), nullable=False),
+        sa.Column("text", sa.String(), nullable=False),
+        sa.Column("token_start", sa.Integer(), nullable=False),
+        sa.Column("token_count", sa.Integer(), nullable=False),
+        sa.Column("page_label", sa.String(), nullable=True),
+        sa.Column("content_sha256", sa.String(length=64), nullable=False),
+        sa.Column("search_text", sa.String(), nullable=False),
+        sa.Column("id", sa.BigInteger(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["chapter_id"],
+            [f"{schema}.ebook_chapter.id"],
+            name=op.f("fk_ebook_chunk_chapter_id_ebook_chapter"),
+            ondelete="SET NULL",
+        ),
+        sa.ForeignKeyConstraint(
+            ["source_id"],
+            [f"{schema}.ebook_source.id"],
+            name=op.f("fk_ebook_chunk_source_id_ebook_source"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_ebook_chunk")),
+        sa.UniqueConstraint("source_id", "chunk_index", name="uq_ebook_chunk_source_id_chunk_index"),
+        sa.UniqueConstraint("source_id", "content_sha256", name="uq_ebook_chunk_source_id_content_sha256"),
+        schema=schema,
+    )
+    op.create_table(
+        "ebook_chunk_embedding_1024",
+        sa.Column("chunk_id", sa.BigInteger(), nullable=False),
+        sa.Column("model_id", sa.Integer(), nullable=False),
+        sa.Column("embedding", pgvector.sqlalchemy.vector.VECTOR(dim=1024), nullable=False),
+        sa.Column("id", sa.BigInteger(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["chunk_id"],
+            [f"{schema}.ebook_chunk.id"],
+            name=op.f("fk_ebook_chunk_embedding_1024_chunk_id_ebook_chunk"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["model_id"],
+            [f"{schema}.ebook_embedding_model.id"],
+            name=op.f("fk_ebook_chunk_embedding_1024_model_id_ebook_embedding_model"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_ebook_chunk_embedding_1024")),
+        sa.UniqueConstraint("chunk_id", "model_id", name=op.f("uq_ebook_chunk_embedding_1024_chunk_id")),
+        schema=schema,
+    )
+    op.create_table(
+        "ebook_chunk_embedding_2560",
+        sa.Column("chunk_id", sa.BigInteger(), nullable=False),
+        sa.Column("model_id", sa.Integer(), nullable=False),
+        sa.Column("embedding", pgvector.sqlalchemy.vector.VECTOR(dim=2560), nullable=False),
+        sa.Column("id", sa.BigInteger(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["chunk_id"],
+            [f"{schema}.ebook_chunk.id"],
+            name=op.f("fk_ebook_chunk_embedding_2560_chunk_id_ebook_chunk"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["model_id"],
+            [f"{schema}.ebook_embedding_model.id"],
+            name=op.f("fk_ebook_chunk_embedding_2560_model_id_ebook_embedding_model"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_ebook_chunk_embedding_2560")),
+        sa.UniqueConstraint("chunk_id", "model_id", name=op.f("uq_ebook_chunk_embedding_2560_chunk_id")),
+        schema=schema,
+    )
+    op.create_table(
+        "ebook_chunk_embedding_4096",
+        sa.Column("chunk_id", sa.BigInteger(), nullable=False),
+        sa.Column("model_id", sa.Integer(), nullable=False),
+        sa.Column("embedding", pgvector.sqlalchemy.vector.VECTOR(dim=4096), nullable=False),
+        sa.Column("id", sa.BigInteger(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["chunk_id"],
+            [f"{schema}.ebook_chunk.id"],
+            name=op.f("fk_ebook_chunk_embedding_4096_chunk_id_ebook_chunk"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["model_id"],
+            [f"{schema}.ebook_embedding_model.id"],
+            name=op.f("fk_ebook_chunk_embedding_4096_model_id_ebook_embedding_model"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_ebook_chunk_embedding_4096")),
+        sa.UniqueConstraint("chunk_id", "model_id", name=op.f("uq_ebook_chunk_embedding_4096_chunk_id")),
+        schema=schema,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("ebook_chunk_embedding_4096", schema=schema)
+    op.drop_table("ebook_chunk_embedding_2560", schema=schema)
+    op.drop_table("ebook_chunk_embedding_1024", schema=schema)
+    op.drop_table("ebook_chunk", schema=schema)
+    op.drop_table("ebook_chapter", schema=schema)
+    op.drop_table("ebook_source", schema=schema)
+    op.drop_table("ebook_embedding_model", schema=schema)
+    # ### end Alembic commands ###

python/alembic/richie/versions/2026_06_10-updated_series_index_to_float_and_added__b3c60cc5beb5.py

@@ -0,0 +1,63 @@
+"""updated series_index to float and added UniqueConstraint to audiobook and audiobook_author.
+
+Revision ID: b3c60cc5beb5
+Revises: d7864d1ffc17
+Create Date: 2026-06-10 20:02:43.073725
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+
+from python.orm import RichieBase
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "b3c60cc5beb5"
+down_revision: str | None = "d7864d1ffc17"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+schema = RichieBase.schema_name
+
+
+def upgrade() -> None:
+    """Upgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column(
+        "audiobook",
+        "series_index",
+        existing_type=sa.INTEGER(),
+        type_=sa.Float(),
+        existing_nullable=False,
+        schema=schema,
+    )
+    op.create_unique_constraint(
+        op.f("uq_audiobook_author_id"),
+        "audiobook",
+        ["author_id", "series_id", "title"],
+        schema=schema,
+        postgresql_nulls_not_distinct=True,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(op.f("uq_audiobook_author_id"), "audiobook", schema=schema, type_="unique")
+    op.alter_column(
+        "audiobook",
+        "series_index",
+        existing_type=sa.Float(),
+        type_=sa.INTEGER(),
+        existing_nullable=False,
+        schema=schema,
+    )
+    # ### end Alembic commands ###

python/orm/richie/__init__.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from python.orm.richie.audiobook import Audiobook, AudiobookAuthor, AudiobookSeries
 from python.orm.richie.base import RichieBase, TableBase, TableBaseBig, TableBaseSmall
 from python.orm.richie.contact import (
     Contact,
@@ -10,11 +11,30 @@ from python.orm.richie.contact import (
     Need,
     RelationshipType,
 )
+from python.orm.richie.ebook import (
+    EbookChapter,
+    EbookChunk,
+    EbookChunkEmbedding1024,
+    EbookChunkEmbedding2560,
+    EbookChunkEmbedding4096,
+    EbookEmbeddingModel,
+    EbookSource,
+)
 
 __all__ = [
+    "Audiobook",
+    "AudiobookAuthor",
+    "AudiobookSeries",
     "Contact",
     "ContactNeed",
     "ContactRelationship",
+    "EbookChapter",
+    "EbookChunk",
+    "EbookChunkEmbedding1024",
+    "EbookChunkEmbedding2560",
+    "EbookChunkEmbedding4096",
+    "EbookEmbeddingModel",
+    "EbookSource",
     "Need",
     "RelationshipType",
     "RichieBase",

python/orm/richie/audiobook.py

@@ -0,0 +1,55 @@
+"""Audiobook catalog models."""
+
+from __future__ import annotations
+
+from sqlalchemy import ForeignKey, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from python.orm.richie.base import TableBase
+
+
+class AudiobookAuthor(TableBase):
+    """Canonical audiobook author."""
+
+    __tablename__ = "audiobook_author"
+    __table_args__ = (UniqueConstraint("name"),)
+
+    name: Mapped[str] = mapped_column(String, unique=True)
+
+    books: Mapped[list[Audiobook]] = relationship("Audiobook", back_populates="author")
+    series: Mapped[list[AudiobookSeries]] = relationship("AudiobookSeries", back_populates="author")
+
+
+class AudiobookSeries(TableBase):
+    """Canonical audiobook series."""
+
+    __tablename__ = "audiobook_series"
+    __table_args__ = (UniqueConstraint("author_id", "name"),)
+
+    name: Mapped[str] = mapped_column(String)
+    author_id: Mapped[int] = mapped_column(ForeignKey("main.audiobook_author.id", ondelete="CASCADE"))
+
+    author: Mapped[AudiobookAuthor] = relationship("AudiobookAuthor", back_populates="series")
+    books: Mapped[list[Audiobook]] = relationship("Audiobook", back_populates="series")
+
+
+class Audiobook(TableBase):
+    """Canonical audiobook title."""
+
+    __tablename__ = "audiobook"
+    __table_args__ = (
+        UniqueConstraint(
+            "author_id",
+            "series_id",
+            "title",
+            postgresql_nulls_not_distinct=True,
+        ),
+    )
+
+    title: Mapped[str] = mapped_column(String)
+    author_id: Mapped[int] = mapped_column(ForeignKey("main.audiobook_author.id", ondelete="CASCADE"))
+    series_id: Mapped[int | None] = mapped_column(ForeignKey("main.audiobook_series.id", ondelete="SET NULL"))
+    series_index: Mapped[float] = mapped_column(default=0.0)
+
+    author: Mapped[AudiobookAuthor] = relationship("AudiobookAuthor", back_populates="books")
+    series: Mapped[AudiobookSeries | None] = relationship("AudiobookSeries", back_populates="books")

python/orm/richie/ebook.py

@@ -0,0 +1,130 @@
+"""EPUB search models."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pgvector.sqlalchemy import Vector
+from sqlalchemy import BigInteger, Boolean, DateTime, ForeignKey, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from python.orm.richie.base import TableBase, TableBaseBig
+
+
+class EbookSource(TableBase):
+    """One indexed EPUB file."""
+
+    __tablename__ = "ebook_source"
+    __table_args__ = (
+        UniqueConstraint("file_path"),
+        UniqueConstraint("file_sha256"),
+    )
+
+    title: Mapped[str]
+    author: Mapped[str | None]
+    language: Mapped[str | None]
+    publisher: Mapped[str | None]
+    identifier: Mapped[str | None]
+    file_path: Mapped[str]
+    file_sha256: Mapped[str] = mapped_column(String(64))
+    file_mtime: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    file_size: Mapped[int] = mapped_column(BigInteger)
+
+    chapters: Mapped[list[EbookChapter]] = relationship(
+        "EbookChapter",
+        back_populates="source",
+        cascade="all, delete-orphan",
+        passive_deletes=True,
+    )
+    chunks: Mapped[list[EbookChunk]] = relationship(
+        "EbookChunk",
+        back_populates="source",
+        cascade="all, delete-orphan",
+        passive_deletes=True,
+    )
+
+
+class EbookChapter(TableBase):
+    """A chapter or spine document inside an EPUB."""
+
+    __tablename__ = "ebook_chapter"
+    __table_args__ = (UniqueConstraint("source_id", "spine_index"),)
+
+    source_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_source.id", ondelete="CASCADE"))
+    spine_index: Mapped[int]
+    title: Mapped[str | None]
+    href: Mapped[str | None]
+
+    source: Mapped[EbookSource] = relationship("EbookSource", back_populates="chapters")
+    chunks: Mapped[list[EbookChunk]] = relationship(
+        "EbookChunk",
+        back_populates="chapter",
+        cascade="all, delete-orphan",
+        passive_deletes=True,
+    )
+
+
+class EbookChunk(TableBaseBig):
+    """A searchable text chunk."""
+
+    __tablename__ = "ebook_chunk"
+    __table_args__ = (
+        UniqueConstraint("source_id", "chunk_index", name="uq_ebook_chunk_source_id_chunk_index"),
+        UniqueConstraint("source_id", "content_sha256", name="uq_ebook_chunk_source_id_content_sha256"),
+    )
+
+    source_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_source.id", ondelete="CASCADE"))
+    chapter_id: Mapped[int | None] = mapped_column(ForeignKey("main.ebook_chapter.id", ondelete="SET NULL"))
+    chunk_index: Mapped[int]
+    text: Mapped[str]
+    token_start: Mapped[int]
+    token_count: Mapped[int]
+    page_label: Mapped[str | None]
+    content_sha256: Mapped[str] = mapped_column(String(64))
+    search_text: Mapped[str]
+
+    source: Mapped[EbookSource] = relationship("EbookSource", back_populates="chunks")
+    chapter: Mapped[EbookChapter | None] = relationship("EbookChapter", back_populates="chunks")
+
+
+class EbookEmbeddingModel(TableBase):
+    """A supported embedding model."""
+
+    __tablename__ = "ebook_embedding_model"
+
+    name: Mapped[str] = mapped_column(String, unique=True)
+    dimension: Mapped[int]
+    is_default: Mapped[bool] = mapped_column(Boolean, default=False)
+
+
+class EbookChunkEmbedding1024(TableBaseBig):
+    """1024-dimensional chunk embedding."""
+
+    __tablename__ = "ebook_chunk_embedding_1024"
+    __table_args__ = (UniqueConstraint("chunk_id", "model_id"),)
+
+    chunk_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_chunk.id", ondelete="CASCADE"))
+    model_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_embedding_model.id", ondelete="CASCADE"))
+    embedding: Mapped[list[float]] = mapped_column(Vector(1024))
+
+
+class EbookChunkEmbedding2560(TableBaseBig):
+    """2560-dimensional chunk embedding."""
+
+    __tablename__ = "ebook_chunk_embedding_2560"
+    __table_args__ = (UniqueConstraint("chunk_id", "model_id"),)
+
+    chunk_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_chunk.id", ondelete="CASCADE"))
+    model_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_embedding_model.id", ondelete="CASCADE"))
+    embedding: Mapped[list[float]] = mapped_column(Vector(2560))
+
+
+class EbookChunkEmbedding4096(TableBaseBig):
+    """4096-dimensional chunk embedding."""
+
+    __tablename__ = "ebook_chunk_embedding_4096"
+    __table_args__ = (UniqueConstraint("chunk_id", "model_id"),)
+
+    chunk_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_chunk.id", ondelete="CASCADE"))
+    model_id: Mapped[int] = mapped_column(ForeignKey("main.ebook_embedding_model.id", ondelete="CASCADE"))
+    embedding: Mapped[list[float]] = mapped_column(Vector(4096))

python/tools/audiobook/__init__.py

@@ -0,0 +1 @@
+"""Audiobook tools."""

python/tools/audiobook/audible_convert.py

@@ -0,0 +1,471 @@
+"""Convert Audible AAX downloads into Audiobookshelf-friendly M4B files."""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import shutil
+import subprocess
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import asdict, dataclass
+from os import getenv
+from pathlib import Path  # noqa: TC003 This is required for the typer CLI
+from typing import TYPE_CHECKING, Annotated, Any
+from uuid import uuid7
+
+import typer
+
+from python.common import configure_logger
+from python.orm.common import get_postgres_engine
+from python.tools.audiobook.metadata_agent import (
+    AgentConfig,
+    StandardBookMetadata,
+    standard_book_metadata,
+    write_agent_log,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.engine import Engine
+
+logger = logging.getLogger(__name__)
+
+SENSITIVE_COMMAND_ARGUMENTS = {"-activation_bytes"}
+BOOK_RANGE_PATTERN = re.compile(r"(?:^|-)books?-(?P<start>[1-9]\d*)-(?P<end>[1-9]\d*)(?:-|$)")
+
+
+@dataclass(frozen=True)
+class ConversionConfig:
+    """Runtime settings for one conversion command."""
+
+    resolved_output: Path
+    ollama_api_key: str
+    agent_config: AgentConfig
+    engine: Engine
+    activation_bytes: str | None
+    dry_run: bool
+    overwrite: bool
+    work_directory_name: str = ".audible_convert"
+    dry_run_directory_name: str = "dry-run"
+    temp_directory_name: str = "tmp"
+    log_directory_name: str = "logs"
+    review_directory_name: str = "review"
+
+
+@dataclass(frozen=True)
+class ConcurrentConversionResult:
+    """Result from running ffmpeg and metadata resolution together."""
+
+    metadata: StandardBookMetadata | None
+    conversion_error: Exception | None
+    metadata_error: Exception | None
+
+
+class CommandExecutionError(RuntimeError):
+    """Command failed without exposing sensitive arguments."""
+
+    def __init__(self, arguments: list[str], returncode: int) -> None:
+        """Create a redacted command failure."""
+        self.arguments = tuple(arguments)
+        self.returncode = returncode
+        command = " ".join(redact_command_arguments(arguments))
+        super().__init__(f"Command failed with exit code {returncode}: {command}")
+
+
+def main(
+    input_directory: Annotated[Path, typer.Argument(help="Directory audible-cli downloads AAX files into.")],
+    output_directory: Annotated[Path, typer.Argument(help="Audiobook output directory.")],
+    *,
+    dry_run: Annotated[
+        bool,
+        typer.Option("--dry-run", help="Print planned output files and write marker files without converting."),
+    ] = False,
+    overwrite: Annotated[bool, typer.Option("--overwrite", help="Overwrite existing M4B files.")] = False,
+) -> None:
+    """Convert AAX files from a download directory into M4B files."""
+    configure_logger()
+    resolved_input = input_directory.resolve(strict=True)
+    resolved_output = output_directory.resolve()
+    if not dry_run:
+        resolved_output.mkdir(parents=True, exist_ok=True)
+
+    ollama_api_key = getenv("OLLAMA_API_KEY")
+    if not ollama_api_key:
+        msg = "OLLAMA_API_KEY is required for audiobook metadata resolution"
+        raise RuntimeError(msg)
+
+    config = ConversionConfig(
+        resolved_output=resolved_output,
+        ollama_api_key=ollama_api_key,
+        agent_config=AgentConfig(),
+        engine=get_postgres_engine(name="RICHIE"),
+        activation_bytes=getenv("AUDIBLE_ACTIVATION_BYTES"),
+        dry_run=dry_run,
+        overwrite=overwrite,
+    )
+
+    aax_files = sorted(resolved_input.glob("*.aax"))
+    if not aax_files:
+        logger.info("No AAX files found in %s", resolved_input)
+        return
+    for aax_file in aax_files:
+        logger.info("Converting %s", aax_file)
+        convert_aax_file_with_agent(aax_file, config)
+
+
+def run_command(arguments: list[str], *, capture: bool = False) -> subprocess.CompletedProcess[str]:
+    """Run a command and return the completed process.
+
+    Args:
+        arguments: Command and arguments to run.
+        capture: Whether to capture stdout and stderr.
+
+    Returns:
+        The completed process.
+    """
+    logger.debug("%s", " ".join(redact_command_arguments(arguments)))
+    try:
+        return subprocess.run(arguments, check=True, capture_output=capture, text=True)
+    except subprocess.CalledProcessError as error:
+        raise CommandExecutionError(arguments, error.returncode) from error
+
+
+def redact_command_arguments(arguments: list[str]) -> list[str]:
+    """Return command arguments with sensitive values redacted."""
+    redacted = []
+    redact_next = False
+    for argument in arguments:
+        if redact_next:
+            redacted.append("<redacted>")
+            redact_next = False
+            continue
+
+        redacted.append(argument)
+        redact_next = argument in SENSITIVE_COMMAND_ARGUMENTS
+    return redacted
+
+
+def read_metadata(aax_file: Path) -> dict[str, str]:
+    """Read ffprobe format tags from an AAX file.
+
+    Args:
+        aax_file: AAX file to inspect.
+
+    Returns:
+        Lower-cased metadata tag names mapped to their values.
+    """
+    completed = run_command(
+        [
+            "ffprobe",
+            "-v",
+            "quiet",
+            "-print_format",
+            "json",
+            "-show_format",
+            str(aax_file),
+        ],
+        capture=True,
+    )
+    ffprobe_data: dict[str, Any] = json.loads(completed.stdout)
+    tags = ffprobe_data.get("format", {}).get("tags", {})
+    return {str(key).lower(): str(value) for key, value in tags.items()}
+
+
+def output_stem(metadata: StandardBookMetadata) -> str:
+    """Build the output stem for a book.
+
+    Args:
+        metadata: Book metadata.
+
+    Returns:
+        Output stem in author-series_01-title form.
+    """
+    index_slug = series_index_slug(metadata.series_index, metadata.title)
+    return f"{metadata.author}-{metadata.series}_{index_slug}-{metadata.title}"
+
+
+def series_index_slug(series_index: float, title: str = "") -> str:
+    """Return a filename-safe series index."""
+    if title_range := title_series_range_slug(series_index, title):
+        return title_range
+    index = float(series_index)
+    if index.is_integer():
+        return f"{int(index):02}"
+    return f"{int(index):02}.5"
+
+
+def title_series_range_slug(series_index: float, title: str) -> str | None:
+    """Return a series range slug found in an omnibus title."""
+    index = float(series_index)
+    if not index.is_integer():
+        return None
+    first_index = int(index)
+    for match in BOOK_RANGE_PATTERN.finditer(title):
+        start = int(match.group("start"))
+        end = int(match.group("end"))
+        if start == first_index and end > start:
+            return f"{start:02}-{end:02}"
+    return None
+
+
+def metadata_output_path(output_directory: Path, metadata: StandardBookMetadata) -> Path:
+    """Build the final M4B path from resolved metadata."""
+    stem = output_stem(metadata)
+    return output_directory / stem / f"{stem}.m4b"
+
+
+def convert_aax_file(
+    aax_file: Path,
+    destination: Path,
+    activation_bytes: str | None,
+    *,
+    overwrite: bool,
+) -> None:
+    """Convert an AAX file into an M4B file.
+
+    Args:
+        aax_file: Source AAX file.
+        destination: Destination M4B file.
+        activation_bytes: Optional Audible activation bytes for ffmpeg.
+        overwrite: Whether to overwrite an existing M4B.
+    """
+    if destination.exists() and not overwrite:
+        logger.info("Skipping existing file %s", destination)
+        return
+
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    arguments = ["ffmpeg", "-hide_banner", "-y" if overwrite else "-n"]
+    if activation_bytes:
+        arguments.extend(["-activation_bytes", activation_bytes])
+    arguments.extend(["-i", str(aax_file), "-map_metadata", "0", "-c", "copy", str(destination)])
+    run_command(arguments)
+
+
+def write_review_file(
+    *,
+    destination: Path | None,
+    ffprobe_metadata: dict[str, str],
+    log_file: Path,
+    metadata: StandardBookMetadata | None,
+    reason: str,
+    review_file: Path,
+    source: Path,
+    temp_file: Path | None,
+) -> None:
+    """Write a manual review file for an unresolved conversion."""
+    review_file.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "destination": str(destination) if destination else None,
+        "ffprobe_metadata": ffprobe_metadata,
+        "metadata": asdict(metadata) if metadata else None,
+        "reason": reason,
+        "source": str(source),
+        "temp_file": str(temp_file) if temp_file else None,
+    }
+    review_file.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+    write_agent_log(log_file, "review_written", path=str(review_file), reason=reason)
+
+
+def cleanup_temp_output(temp_file: Path) -> None:
+    """Remove a run's temporary output directory."""
+    shutil.rmtree(temp_file.parent, ignore_errors=True)
+
+
+def dry_run_aax_file_with_agent(
+    aax_file: Path,
+    ffprobe_metadata: dict[str, str],
+    engine: Engine,
+    config: ConversionConfig,
+    log_file: Path,
+    review_file: Path,
+) -> None:
+    """Resolve and print the planned output path without converting."""
+    metadata = standard_book_metadata(
+        aax_file.name,
+        ffprobe_metadata,
+        engine,
+        log_file,
+        config.ollama_api_key,
+        config.agent_config,
+    )
+    destination = None if metadata.needs_review else metadata_output_path(config.resolved_output, metadata)
+    if metadata.needs_review:
+        write_review_file(
+            destination=destination,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=metadata,
+            reason="metadata_needs_review",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=None,
+        )
+        typer.echo(f"{aax_file} -> REVIEW {review_file}")
+    else:
+        stem = output_stem(metadata)
+        dry_run_file = (
+            config.resolved_output / config.work_directory_name / config.dry_run_directory_name / stem / f"{stem}.m4b"
+        )
+        dry_run_file.parent.mkdir(parents=True, exist_ok=True)
+        dry_run_file.write_text(f"{destination}\n", encoding="utf-8")
+        write_agent_log(
+            log_file,
+            "dry_run_file_written",
+            destination=str(destination),
+            path=str(dry_run_file),
+        )
+        typer.echo(f"{aax_file} -> {destination}")
+
+
+def convert_temp_file_and_resolve_metadata(
+    aax_file: Path,
+    temp_file: Path,
+    ffprobe_metadata: dict[str, str],
+    config: ConversionConfig,
+    log_file: Path,
+) -> ConcurrentConversionResult:
+    """Run ffmpeg and metadata resolution in parallel."""
+    conversion_error: Exception | None = None
+    metadata_error: Exception | None = None
+    metadata: StandardBookMetadata | None = None
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        conversion_future = executor.submit(
+            convert_aax_file,
+            aax_file,
+            temp_file,
+            config.activation_bytes,
+            overwrite=True,
+        )
+        metadata_future = executor.submit(
+            standard_book_metadata,
+            aax_file.name,
+            ffprobe_metadata,
+            config.engine,
+            log_file,
+            config.ollama_api_key,
+            config.agent_config,
+        )
+
+        conversion_error = conversion_future.exception()
+        if conversion_error is None:
+            conversion_future.result()
+
+        metadata_error = metadata_future.exception()
+        if metadata_error is None:
+            metadata = metadata_future.result()
+
+    return ConcurrentConversionResult(
+        metadata=metadata,
+        conversion_error=conversion_error,
+        metadata_error=metadata_error,
+    )
+
+
+def convert_aax_file_with_agent(aax_file: Path, config: ConversionConfig) -> None:
+    """Convert one AAX file using the metadata agent for the final path."""
+    run_id = uuid7().hex
+    log_file = config.resolved_output / config.work_directory_name / config.log_directory_name / f"{run_id}.jsonl"
+    review_file = config.resolved_output / config.work_directory_name / config.review_directory_name / f"{run_id}.json"
+    write_agent_log(log_file, "conversion_start", source=str(aax_file), dry_run=config.dry_run)
+    try:
+        ffprobe_metadata = read_metadata(aax_file)
+    except Exception as error:
+        logger.exception("ffprobe failed")
+        write_review_file(
+            destination=None,
+            ffprobe_metadata={},
+            log_file=log_file,
+            metadata=None,
+            reason=f"ffprobe_failed: {error}",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=None,
+        )
+        return
+
+    if config.dry_run:
+        dry_run_aax_file_with_agent(
+            aax_file,
+            ffprobe_metadata,
+            config.engine,
+            config,
+            log_file,
+            review_file,
+        )
+        return
+
+    temp_file = (
+        config.resolved_output / config.work_directory_name / config.temp_directory_name / run_id / "converted.m4b"
+    )
+    temp_file.parent.mkdir(parents=True, exist_ok=True)
+
+    result = convert_temp_file_and_resolve_metadata(aax_file, temp_file, ffprobe_metadata, config, log_file)
+
+    if result.conversion_error:
+        reason = f"ffmpeg_failed: {result.conversion_error}"
+        write_review_file(
+            destination=None,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=result.metadata,
+            reason=reason,
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file if temp_file.exists() else None,
+        )
+        return
+
+    if result.metadata_error:
+        write_review_file(
+            destination=None,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=None,
+            reason=f"metadata_failed: {result.metadata_error}",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file,
+        )
+        return
+
+    if result.metadata is None or result.metadata.needs_review:
+        write_review_file(
+            destination=None,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=result.metadata,
+            reason="metadata_needs_review",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file,
+        )
+        return
+
+    destination = metadata_output_path(config.resolved_output, result.metadata)
+    if destination.exists() and not config.overwrite:
+        write_agent_log(log_file, "destination_exists", destination=str(destination))
+        cleanup_temp_output(temp_file)
+        return
+
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        temp_file.replace(destination)
+    except Exception as error:  # noqa: BLE001
+        write_review_file(
+            destination=destination,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=result.metadata,
+            reason=f"rename_failed: {error}",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file if temp_file.exists() else None,
+        )
+    else:
+        cleanup_temp_output(temp_file)
+        write_agent_log(log_file, "conversion_complete", destination=str(destination))
+
+
+if __name__ == "__main__":
+    typer.run(main)

python/tools/audiobook/catalog.py

@@ -0,0 +1,176 @@
+"""Import audiobook catalog authors and series from CSV files."""
+
+from __future__ import annotations
+
+import csv
+import logging
+from pathlib import Path  # noqa: TC003 This is required for the typer CLI
+from typing import Annotated
+
+import typer
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from python.common import configure_logger
+from python.orm.common import get_postgres_engine
+from python.orm.richie import AudiobookAuthor, AudiobookSeries
+
+logger = logging.getLogger(__name__)
+
+AUTHOR_NAME_COLUMN = "author_name"
+ID_COLUMN = "id"
+NAME_COLUMN = "name"
+
+
+class CatalogImportError(ValueError):
+    """CSV catalog import failed validation."""
+
+
+def main(
+    authors_csv: Annotated[Path, typer.Argument(help="CSV with name and optional id.")],
+    series_csv: Annotated[Path, typer.Argument(help="CSV with name, author_name, and optional id.")],
+) -> None:
+    """Upsert audiobook authors and series from CSV files."""
+    configure_logger()
+    try:
+        engine = get_postgres_engine(name="RICHIE")
+        with Session(engine) as session:
+            author_count = upsert_authors_from_csv(session, authors_csv)
+            series_count = upsert_series_from_csv(session, series_csv)
+            session.commit()
+    except CatalogImportError as error:
+        typer.echo(str(error), err=True)
+        raise typer.Exit(code=1) from error
+
+    logger.info("Upserted %s authors and %s series", author_count, series_count)
+
+
+def upsert_authors_from_csv(session: Session, authors_csv: Path) -> int:
+    """Upsert authors from a CSV file."""
+    count = 0
+    for row_number, row in csv_rows(authors_csv):
+        name = required_csv_value(row, authors_csv, row_number, NAME_COLUMN)
+        upsert_author(session, name, csv_id(row, authors_csv, row_number))
+        count += 1
+    return count
+
+
+def upsert_series_from_csv(session: Session, series_csv: Path) -> int:
+    """Upsert series from a CSV file."""
+    count = 0
+    for row_number, row in csv_rows(series_csv):
+        series_name = required_csv_value(row, series_csv, row_number, NAME_COLUMN)
+        author_name = required_csv_value(row, series_csv, row_number, AUTHOR_NAME_COLUMN)
+        author = find_author_by_name(session, author_name)
+        if author is None:
+            msg = f"{series_csv}:{row_number}: author not found: {author_name}"
+            raise CatalogImportError(msg)
+        upsert_series(session, series_name, author, csv_id(row, series_csv, row_number))
+        count += 1
+    return count
+
+
+def upsert_author(session: Session, name: str, author_id: int | None) -> AudiobookAuthor:
+    """Upsert one author by id or exact name."""
+    if author_id is not None:
+        author = session.get(AudiobookAuthor, author_id)
+        if author is None:
+            author = AudiobookAuthor(id=author_id, name=name)
+            session.add(author)
+        else:
+            author.name = name
+        session.flush()
+        return author
+
+    author = find_author_by_name(session, name)
+    if author is None:
+        author = AudiobookAuthor(name=name)
+        session.add(author)
+        session.flush()
+    return author
+
+
+def upsert_series(
+    session: Session,
+    name: str,
+    author: AudiobookAuthor,
+    series_id: int | None,
+) -> AudiobookSeries:
+    """Upsert one series by id or exact author/name match."""
+    if series_id is not None:
+        series = session.get(AudiobookSeries, series_id)
+        if series is None:
+            series = AudiobookSeries(id=series_id, name=name, author=author)
+            session.add(series)
+        else:
+            series.name = name
+            series.author = author
+        session.flush()
+        return series
+
+    series = find_series_by_name_and_author(session, name, author.id)
+    if series is None:
+        series = AudiobookSeries(name=name, author=author)
+        session.add(series)
+        session.flush()
+    return series
+
+
+def find_author_by_name(session: Session, name: str) -> AudiobookAuthor | None:
+    """Find one author by exact name."""
+    return session.scalar(select(AudiobookAuthor).where(AudiobookAuthor.name == name))
+
+
+def find_series_by_name_and_author(
+    session: Session,
+    name: str,
+    author_id: int,
+) -> AudiobookSeries | None:
+    """Find one series by exact name and author."""
+    return session.scalar(
+        select(AudiobookSeries).where(
+            AudiobookSeries.name == name,
+            AudiobookSeries.author_id == author_id,
+        ),
+    )
+
+
+def csv_rows(csv_path: Path) -> list[tuple[int, dict[str, str | None]]]:
+    """Read a CSV file as numbered rows."""
+    with csv_path.open(newline="", encoding="utf-8") as file:
+        reader = csv.DictReader(file)
+        if reader.fieldnames is None:
+            msg = f"{csv_path}: missing CSV header"
+            raise CatalogImportError(msg)
+        return [(row_number, row) for row_number, row in enumerate(reader, start=2)]
+
+
+def required_csv_value(
+    row: dict[str, str | None],
+    csv_path: Path,
+    row_number: int,
+    column: str,
+) -> str:
+    """Read a required CSV value."""
+    value = row.get(column)
+    if value and value.strip():
+        return value.strip()
+    msg = f"{csv_path}:{row_number}: missing required column value: {column}"
+    raise CatalogImportError(msg)
+
+
+def csv_id(row: dict[str, str | None], csv_path: Path, row_number: int) -> int | None:
+    """Read an optional id field from a CSV row."""
+    value = row.get(ID_COLUMN)
+    if value is None or not value.strip():
+        return None
+    try:
+        return int(value)
+    except ValueError as error:
+        msg = f"{csv_path}:{row_number}: id must be an integer: {value}"
+        raise CatalogImportError(msg) from error
+    return None
+
+
+if __name__ == "__main__":
+    typer.run(main)

python/tools/audiobook/llm_tool_calling.py

@@ -0,0 +1,599 @@
+"""LLM tool calling support for audiobook metadata resolution."""
+
+from __future__ import annotations
+
+import json
+import re
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from sqlalchemy import or_, select
+
+from python.orm.richie import Audiobook, AudiobookAuthor, AudiobookSeries
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from sqlalchemy.orm import Session
+
+    from python.tools.audiobook.metadata_agent import AgentConfig
+
+CATALOG_SLUG_PATTERN = re.compile(r"^[a-z0-9]+(?:_[a-z0-9]+)*$")
+TITLE_SLUG_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
+
+LogWriter = Callable[..., None]
+
+
+class MetadataResolutionError(ValueError):
+    """Metadata resolution failed validation."""
+
+
+@dataclass(frozen=True)
+class EnsuredBook:
+    """Book row plus whether it was created."""
+
+    book: Audiobook
+    action: str
+
+
+class CatalogToolRegistry:
+    """Controlled catalog tools exposed to the metadata model."""
+
+    def __init__(
+        self,
+        session: Session,
+        log_path: Path,
+        config: AgentConfig,
+        write_log: LogWriter,
+    ) -> None:
+        """Create a registry bound to one database session and audit log."""
+        self.session = session
+        self.log_path = log_path
+        self.config = config
+        self.write_log = write_log
+        self.seen_author_ids: set[int] = set()
+        self.seen_series_ids: set[int] = set()
+        self.seen_book_ids: set[int] = set()
+        self.created_author_ids: set[int] = set()
+        self.created_series_ids: set[int] = set()
+        self.created_book_ids: set[int] = set()
+
+    def tool_schemas(self) -> list[dict[str, object]]:
+        """Return Ollama tool schemas."""
+        schemas = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "search_authors",
+                    "description": "Search canonical audiobook authors by slug or noisy source text.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"query": {"type": "string"}},
+                        "required": ["query"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "search_series",
+                    "description": "Search canonical audiobook series by slug or noisy source text.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "query": {"type": "string"},
+                            "author_id": {"type": ["integer", "null"]},
+                        },
+                        "required": ["query"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "search_books",
+                    "description": "Search canonical audiobook titles with optional author and series filters.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "query": {"type": "string"},
+                            "author_id": {"type": ["integer", "null"]},
+                            "series_id": {"type": ["integer", "null"]},
+                        },
+                        "required": ["query"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "ensure_author",
+                    "description": "Normalize an author name to a catalog slug, then return or create that author.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"name": {"type": "string"}},
+                        "required": ["name"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "ensure_series",
+                    "description": "Normalize a series name to a catalog slug, then return or create it for an author.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "author_id": {"type": "integer"},
+                        },
+                        "required": ["name", "author_id"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "ensure_book",
+                    "description": "Normalize a title to a book slug, then return or create it for an author/series.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "title": {"type": "string"},
+                            "author_id": {"type": "integer"},
+                            "series_id": {"type": ["integer", "null"]},
+                            "series_index": {"type": "number", "multipleOf": 0.5},
+                        },
+                        "required": ["title", "author_id", "series_id", "series_index"],
+                    },
+                },
+            },
+        ]
+        enabled_tool_names = set(self.config.tool_names)
+        return [schema for schema in schemas if schema["function"]["name"] in enabled_tool_names]
+
+    def run(self, name: str, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Run one catalog tool and audit the call."""
+        handlers = {
+            "search_authors": self.run_search_authors,
+            "search_series": self.run_search_series,
+            "search_books": self.run_search_books,
+            "ensure_author": self.run_ensure_author,
+            "ensure_series": self.run_ensure_series,
+            "ensure_book": self.run_ensure_book,
+        }
+        handler = handlers.get(name)
+        if handler is None:
+            self.write_log(self.log_path, "tool_error", tool=name, arguments=arguments, error="unknown_tool")
+            msg = f"Unknown audiobook metadata tool: {name}"
+            raise MetadataResolutionError(msg)
+        if name not in self.config.tool_names:
+            self.write_log(self.log_path, "tool_error", tool=name, arguments=arguments, error="tool_not_enabled")
+            msg = f"Audiobook metadata tool is not enabled: {name}"
+            raise MetadataResolutionError(msg)
+
+        started = time.perf_counter()
+        self.write_log(self.log_path, "tool_call", tool=name, arguments=arguments)
+        result = handler(arguments)
+        duration_ms = round((time.perf_counter() - started) * 1000, 3)
+        self.write_log(
+            self.log_path,
+            "tool_result",
+            tool=name,
+            duration_ms=duration_ms,
+            result_count=len(result),
+            preview=result[:3],
+        )
+        return result
+
+    def get_author(self, author_id: int) -> AudiobookAuthor | None:
+        """Return an author by id."""
+        return self.session.get(AudiobookAuthor, author_id)
+
+    def get_book(self, book_id: int) -> Audiobook | None:
+        """Return a book by id."""
+        return self.session.get(Audiobook, book_id)
+
+    def get_series(self, series_id: int) -> AudiobookSeries | None:
+        """Return a series by id."""
+        return self.session.get(AudiobookSeries, series_id)
+
+    def prune_unused_created_rows(self, *, author_id: int, book_id: int | None, series_id: int | None) -> None:
+        """Remove catalog rows created during this run but not used by final metadata."""
+        used_book_ids = {book_id} if book_id is not None else set()
+        for created_book_id in self.created_book_ids - used_book_ids:
+            if book := self.get_book(created_book_id):
+                self.session.delete(book)
+
+        self.session.flush()
+        used_series_ids = {series_id} if series_id is not None else set()
+        for created_series_id in self.created_series_ids - used_series_ids:
+            series = self.get_series(created_series_id)
+            if series and not series.books:
+                self.session.delete(series)
+
+        self.session.flush()
+        for created_author_id in self.created_author_ids - {author_id}:
+            author = self.get_author(created_author_id)
+            if author and not author.books and not author.series:
+                self.session.delete(author)
+
+    def run_search_authors(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Search authors from tool arguments and remember returned ids."""
+        query = required_string(arguments, "query")
+        statement = select(AudiobookAuthor).order_by(AudiobookAuthor.name).limit(self.config.max_tool_results)
+        if terms := query_terms(query):
+            statement = statement.where(or_(*(AudiobookAuthor.name.ilike(f"%{term}%") for term in terms)))
+
+        authors = self.session.scalars(statement).all()
+        self.seen_author_ids.update(author.id for author in authors)
+        return [{"id": author.id, "name": author.name} for author in authors]
+
+    def run_search_series(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Search series from tool arguments and remember returned ids."""
+        query = required_string(arguments, "query")
+        author_id = optional_int(arguments.get("author_id"), "author_id")
+        statement = select(AudiobookSeries).order_by(AudiobookSeries.name).limit(self.config.max_tool_results)
+        if terms := query_terms(query):
+            statement = statement.where(or_(*(AudiobookSeries.name.ilike(f"%{term}%") for term in terms)))
+        if author_id is not None:
+            statement = statement.where(AudiobookSeries.author_id == author_id)
+
+        series_rows = self.session.scalars(statement).all()
+        self.seen_series_ids.update(series.id for series in series_rows)
+        self.seen_author_ids.update(series.author_id for series in series_rows)
+        return [
+            {
+                "id": series.id,
+                "name": series.name,
+                "author_id": series.author_id,
+                "author": series.author.name,
+            }
+            for series in series_rows
+        ]
+
+    def run_search_books(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Search books from tool arguments and remember returned ids."""
+        query = required_string(arguments, "query")
+        author_id = optional_int(arguments.get("author_id"), "author_id")
+        series_id = optional_int(arguments.get("series_id"), "series_id")
+        statement = select(Audiobook).order_by(Audiobook.title).limit(self.config.max_tool_results)
+        if terms := query_terms(query):
+            statement = statement.where(or_(*(Audiobook.title.ilike(f"%{term}%") for term in terms)))
+        if author_id is not None:
+            statement = statement.where(Audiobook.author_id == author_id)
+        if series_id is not None:
+            statement = statement.where(Audiobook.series_id == series_id)
+
+        books = self.session.scalars(statement).all()
+        self.seen_book_ids.update(book.id for book in books)
+        self.seen_author_ids.update(book.author_id for book in books)
+        self.seen_series_ids.update(book.series_id for book in books if book.series_id is not None)
+        return [
+            {
+                "id": book.id,
+                "title": book.title,
+                "author_id": book.author_id,
+                "author": book.author.name,
+                "series_id": book.series_id,
+                "series": book.series.name if book.series else self.config.standalone_series,
+                "series_index": book.series_index,
+            }
+            for book in books
+        ]
+
+    def run_ensure_author(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Ensure an author from tool arguments and return a tool result."""
+        name = normalize_catalog_slug(required_string(arguments, "name"))
+        validate_catalog_slug(name, "author")
+        author = self.session.scalar(select(AudiobookAuthor).where(AudiobookAuthor.name == name))
+        action = "existing"
+        if author is None:
+            author = AudiobookAuthor(name=name)
+            self.session.add(author)
+            self.session.flush()
+            self.created_author_ids.add(author.id)
+            action = "created"
+
+        self.seen_author_ids.add(author.id)
+        return [{"id": author.id, "name": author.name, "action": action}]
+
+    def run_ensure_series(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Ensure a series from tool arguments and return a tool result."""
+        name = normalize_catalog_slug(required_string(arguments, "name"))
+        author_id = required_int(arguments, "author_id")
+        validate_catalog_slug(name, "series")
+        author = self.required_author(author_id)
+        series = self.find_series_by_catalog_slug(name, author.id)
+        action = "existing"
+        if series is None:
+            series = AudiobookSeries(name=name, author=author)
+            self.session.add(series)
+            self.session.flush()
+            self.created_series_ids.add(series.id)
+            action = "created"
+
+        self.seen_author_ids.add(author.id)
+        self.seen_series_ids.add(series.id)
+        return [self.series_result(series, action)]
+
+    def run_ensure_book(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Ensure a book from tool arguments and return a tool result."""
+        title = required_string(arguments, "title")
+        author_id = required_int(arguments, "author_id")
+        series_id = optional_int(arguments.get("series_id"), "series_id")
+        series_index = required_series_index(arguments, "series_index")
+        ensured = self.ensure_book(title, author_id, series_id, series_index)
+        return [self.book_result(ensured.book, ensured.action)]
+
+    def ensure_book(
+        self,
+        title: str,
+        author_id: int,
+        series_id: int | None,
+        series_index: float,
+    ) -> EnsuredBook:
+        """Return an existing book row, or create it after validating ownership."""
+        title = normalize_title_slug(title)
+        validate_title_slug(title)
+        author = self.required_author(author_id)
+        series = None
+        if series_id is None:
+            if series_index != 0:
+                msg = "standalone books must use series_index 0"
+                raise MetadataResolutionError(msg)
+        else:
+            series = self.required_series(series_id)
+            if series.author_id != author.id:
+                msg = f"series_id {series_id} does not belong to author_id {author_id}"
+                raise MetadataResolutionError(msg)
+            if series_index <= 0:
+                msg = "series books must use a positive series_index"
+                raise MetadataResolutionError(msg)
+
+        statement = select(Audiobook).where(
+            Audiobook.title == title,
+            Audiobook.author_id == author.id,
+        )
+        if series is None:
+            statement = statement.where(Audiobook.series_id.is_(None))
+        else:
+            statement = statement.where(Audiobook.series_id == series.id)
+        book = self.session.scalar(statement)
+        if book is None:
+            book = Audiobook(title=title, author=author, series=series, series_index=series_index)
+            self.session.add(book)
+            self.session.flush()
+            self.created_book_ids.add(book.id)
+            action = "created"
+        else:
+            action = "existing"
+
+        self.seen_book_ids.add(book.id)
+        self.seen_author_ids.add(author.id)
+        if book.series_id is not None:
+            self.seen_series_ids.add(book.series_id)
+        return EnsuredBook(book=book, action=action)
+
+    def required_author(self, author_id: int) -> AudiobookAuthor:
+        """Return an author or fail metadata resolution."""
+        author = self.get_author(author_id)
+        if author is None:
+            msg = f"author_id {author_id} does not exist"
+            raise MetadataResolutionError(msg)
+        return author
+
+    def required_series(self, series_id: int) -> AudiobookSeries:
+        """Return a series or fail metadata resolution."""
+        series = self.get_series(series_id)
+        if series is None:
+            msg = f"series_id {series_id} does not exist"
+            raise MetadataResolutionError(msg)
+        return series
+
+    def find_series_by_catalog_slug(self, name: str, author_id: int) -> AudiobookSeries | None:
+        """Return a series by exact slug or underscore-insensitive slug."""
+        exact = self.session.scalar(
+            select(AudiobookSeries).where(
+                AudiobookSeries.name == name,
+                AudiobookSeries.author_id == author_id,
+            ),
+        )
+        if exact is not None:
+            return exact
+
+        compact_name = compact_catalog_slug(name)
+        series_rows = self.session.scalars(
+            select(AudiobookSeries).where(AudiobookSeries.author_id == author_id).order_by(AudiobookSeries.name),
+        ).all()
+        for series in series_rows:
+            if compact_catalog_slug(series.name) == compact_name:
+                return series
+        return None
+
+    def series_result(self, series: AudiobookSeries, action: str) -> dict[str, object]:
+        """Build a normalized series tool result."""
+        return {
+            "id": series.id,
+            "name": series.name,
+            "author_id": series.author_id,
+            "author": series.author.name,
+            "action": action,
+        }
+
+    def book_result(self, book: Audiobook, action: str) -> dict[str, object]:
+        """Build a normalized book tool result."""
+        return {
+            "id": book.id,
+            "title": book.title,
+            "author_id": book.author_id,
+            "author": book.author.name,
+            "series_id": book.series_id,
+            "series": book.series.name if book.series else self.config.standalone_series,
+            "series_index": book.series_index,
+            "action": action,
+        }
+
+
+def run_tool_calls(
+    messages: list[dict[str, object]],
+    message: dict[str, object],
+    tool_calls: list[tuple[str, dict[str, object]]],
+    registry: CatalogToolRegistry,
+    log_path: Path,
+    write_log: LogWriter,
+) -> str | None:
+    """Run tool calls, append tool messages, and return fatal error text when stopped."""
+    messages.append(message)
+    for tool_name, arguments in tool_calls:
+        try:
+            tool_result = registry.run(tool_name, arguments)
+        except MetadataResolutionError as error:
+            if is_fatal_tool_error(error):
+                return str(error)
+            write_log(log_path, "tool_error", tool=tool_name, arguments=arguments, error=str(error))
+            messages.append(
+                {
+                    "role": "tool",
+                    "tool_name": tool_name,
+                    "content": json.dumps({"error": str(error)}, sort_keys=True),
+                },
+            )
+            continue
+        messages.append(
+            {
+                "role": "tool",
+                "tool_name": tool_name,
+                "content": json.dumps(tool_result, sort_keys=True),
+            },
+        )
+    return None
+
+
+def parse_tool_calls(message: dict[str, object]) -> list[tuple[str, dict[str, object]]]:
+    """Parse Ollama tool calls from a response message."""
+    raw_tool_calls = message.get("tool_calls") or []
+    if not isinstance(raw_tool_calls, list):
+        msg = "tool_calls must be a list"
+        raise MetadataResolutionError(msg)
+
+    tool_calls = []
+    for raw_call in raw_tool_calls:
+        if not isinstance(raw_call, dict):
+            msg = "tool call must be an object"
+            raise MetadataResolutionError(msg)
+        function = raw_call.get("function")
+        if not isinstance(function, dict):
+            msg = "tool call is missing function"
+            raise MetadataResolutionError(msg)
+        name = function.get("name")
+        if not isinstance(name, str) or not name:
+            msg = "tool call is missing function name"
+            raise MetadataResolutionError(msg)
+        arguments = parse_tool_arguments(function.get("arguments", {}))
+        tool_calls.append((name, arguments))
+    return tool_calls
+
+
+def parse_tool_arguments(raw_arguments: object) -> dict[str, object]:
+    """Parse tool call arguments returned by Ollama."""
+    if isinstance(raw_arguments, dict):
+        return {str(key): value for key, value in raw_arguments.items()}
+    if isinstance(raw_arguments, str):
+        parsed = json.loads(raw_arguments) if raw_arguments else {}
+        if isinstance(parsed, dict):
+            return {str(key): value for key, value in parsed.items()}
+    msg = "tool arguments must be an object"
+    raise MetadataResolutionError(msg)
+
+
+def validate_title_slug(title: str) -> None:
+    """Validate a canonical book title slug."""
+    if not TITLE_SLUG_PATTERN.fullmatch(title):
+        msg = f"title slug is invalid: {title}"
+        raise MetadataResolutionError(msg)
+
+
+def validate_catalog_slug(value: str, label: str) -> None:
+    """Validate a canonical catalog slug."""
+    if not CATALOG_SLUG_PATTERN.fullmatch(value):
+        msg = f"{label} slug is invalid: {value}"
+        raise MetadataResolutionError(msg)
+
+
+def normalize_catalog_slug(value: str) -> str:
+    """Normalize noisy catalog names into lower snake-case slugs."""
+    return re.sub(r"[^a-z0-9]+", "_", value.strip().casefold()).strip("_")
+
+
+def compact_catalog_slug(value: str) -> str:
+    """Return a catalog slug comparison key that ignores underscores."""
+    return normalize_catalog_slug(value).replace("_", "")
+
+
+def normalize_title_slug(value: str) -> str:
+    """Normalize noisy book titles into lower kebab-case slugs."""
+    return re.sub(r"[^a-z0-9]+", "-", value.strip().casefold()).strip("-")
+
+
+def is_fatal_tool_error(error: MetadataResolutionError) -> bool:
+    """Return whether a tool error should stop the agent immediately."""
+    message = str(error)
+    return message.startswith(
+        (
+            "Unknown audiobook metadata tool",
+            "Audiobook metadata tool is not enabled",
+        ),
+    )
+
+
+def query_terms(query: str) -> tuple[str, ...]:
+    """Return text variants useful for matching noisy audiobook metadata."""
+    normalized = query.strip().casefold()
+    underscore_slug = normalize_catalog_slug(normalized)
+    compact_slug = compact_catalog_slug(normalized)
+    hyphen_slug = normalize_title_slug(normalized)
+    return tuple(dict.fromkeys(term for term in (normalized, underscore_slug, compact_slug, hyphen_slug) if term))
+
+
+def required_string(data: dict[str, object], key: str) -> str:
+    """Read a required string field."""
+    value = data.get(key)
+    if not isinstance(value, str) or not value.strip():
+        msg = f"{key} must be a non-empty string"
+        raise MetadataResolutionError(msg)
+    return value.strip()
+
+
+def required_int(data: dict[str, object], key: str) -> int:
+    """Read a required integer field."""
+    value = data.get(key)
+    if isinstance(value, bool) or not isinstance(value, int):
+        msg = f"{key} must be an integer"
+        raise MetadataResolutionError(msg)
+    return value
+
+
+def required_series_index(data: dict[str, object], key: str) -> float:
+    """Read a required whole-number or half-number series index."""
+    value = data.get(key)
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        msg = f"{key} must be a number"
+        raise MetadataResolutionError(msg)
+    series_index = float(value)
+    if not (series_index * 2).is_integer():
+        msg = f"{key} must be a whole number or .5 increment"
+        raise MetadataResolutionError(msg)
+    return series_index
+
+
+def optional_int(value: object, key: str) -> int | None:
+    """Read an optional integer field."""
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, int):
+        msg = f"{key} must be an integer or null"
+        raise MetadataResolutionError(msg)
+    return value

python/tools/audiobook/metadata_agent.py

@@ -0,0 +1,575 @@
+"""Resolve audiobook metadata with a controlled Ollama tool loop."""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import asdict, dataclass, is_dataclass, replace
+from os import PathLike
+from typing import TYPE_CHECKING
+
+import httpx
+from sqlalchemy.orm import Session
+
+from python.common import utcnow
+from python.tools.audiobook.llm_tool_calling import (
+    CatalogToolRegistry,
+    MetadataResolutionError,
+    normalize_title_slug,
+    optional_int,
+    parse_tool_calls,
+    required_int,
+    required_series_index,
+    required_string,
+    run_tool_calls,
+    validate_catalog_slug,
+    validate_title_slug,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from sqlalchemy.engine import Engine
+
+    from python.orm.richie import AudiobookAuthor
+
+FENCED_JSON_PATTERN = re.compile(r"^```(?:json)?\s*(?P<json>.*?)\s*```$", re.IGNORECASE | re.DOTALL)
+
+
+@dataclass(frozen=True)
+class AgentConfig:
+    """Runtime settings for the audiobook metadata agent."""
+
+    model: str = "deepseek-v4-flash:cloud"
+    ollama_chat_url: str = "https://ollama.com/api/chat"
+    http_timeout_seconds: int = 300
+    max_agent_turns: int = 8
+    max_tool_results: int = 10
+    min_confidence: float = 0.85
+    invalid_final_retries: int = 1
+    standalone_series: str = "standalone"
+    tool_names: tuple[str, ...] = (
+        "search_authors",
+        "search_series",
+        "search_books",
+        "ensure_author",
+        "ensure_series",
+        "ensure_book",
+    )
+
+
+@dataclass(frozen=True)
+class StandardBookMetadata:
+    """Canonical metadata for the final audiobook path."""
+
+    author_id: int
+    author: str
+    book_id: int | None
+    title: str
+    series_id: int | None
+    series: str
+    series_index: float
+    confidence: float
+    needs_review: bool
+    evidence: list[str]
+
+
+@dataclass(frozen=True)
+class FinalMetadataFields:
+    """Raw model fields after schema validation."""
+
+    author_id: int
+    book_id: int | None
+    title: str
+    series_id: int | None
+    series_index: float
+    confidence: float
+    evidence: list[str]
+
+
+@dataclass(frozen=True)
+class ResolvedBookFields:
+    """Book fields after optional catalog book resolution."""
+
+    book_id: int | None
+    title: str
+    series_id: int | None
+    series_index: float
+
+
+@dataclass(frozen=True)
+class AgentStepResult:
+    """Outcome from one model response."""
+
+    metadata: StandardBookMetadata | None
+    invalid_final_count: int
+    should_continue: bool
+
+
+def standard_book_metadata(
+    aax_file_name: str,
+    aax_metadata_from_ffprobe: dict[str, str],
+    engine: Engine,
+    log_path: Path,
+    ollama_api_key: str,
+    config: AgentConfig,
+) -> StandardBookMetadata:
+    """Resolve canonical audiobook metadata with the configured Ollama Cloud model."""
+    with Session(engine) as session:
+        registry = CatalogToolRegistry(session, log_path, config, write_agent_log)
+        agent = AudiobookMetadataAgent(
+            registry=registry, log_path=log_path, ollama_api_key=ollama_api_key, config=config
+        )
+        metadata = agent.run(aax_file_name, aax_metadata_from_ffprobe)
+        if metadata.needs_review:
+            session.rollback()
+        else:
+            registry.prune_unused_created_rows(
+                author_id=metadata.author_id,
+                book_id=metadata.book_id,
+                series_id=metadata.series_id,
+            )
+            session.commit()
+        return metadata
+
+
+class AudiobookMetadataAgent:
+    """Ollama-backed metadata resolver with a fixed local tool registry."""
+
+    def __init__(
+        self,
+        *,
+        registry: CatalogToolRegistry,
+        log_path: Path,
+        ollama_api_key: str,
+        config: AgentConfig,
+    ) -> None:
+        """Create an Ollama metadata agent."""
+        self._registry = registry
+        self._log_path = log_path
+        self._ollama_api_key = ollama_api_key
+        self._config = config
+
+    def run(self, aax_file_name: str, aax_metadata_from_ffprobe: dict[str, str]) -> StandardBookMetadata:
+        """Resolve metadata for one AAX file."""
+        messages = [
+            {"role": "system", "content": system_prompt()},
+            {"role": "user", "content": user_prompt(aax_file_name, aax_metadata_from_ffprobe)},
+        ]
+        invalid_final_count = 0
+        result: StandardBookMetadata | None = None
+
+        for turn in range(1, self._config.max_agent_turns + 1):
+            step = self.run_step(messages, turn, invalid_final_count)
+            invalid_final_count = step.invalid_final_count
+            if step.should_continue:
+                continue
+            result = step.metadata
+            break
+
+        if result is None:
+            return self.force_final_response(messages)
+        return result
+
+    def run_step(
+        self,
+        messages: list[dict[str, object]],
+        turn: int,
+        invalid_final_count: int,
+    ) -> AgentStepResult:
+        """Run one model turn and return the next agent-loop action."""
+        data = self.chat(messages, turn)
+        message = data.get("message")
+        if not isinstance(message, dict):
+            return AgentStepResult(
+                metadata=review_metadata("Ollama response did not include a message", self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+
+        try:
+            tool_calls = parse_tool_calls(message)
+        except (json.JSONDecodeError, MetadataResolutionError) as error:
+            return AgentStepResult(
+                metadata=review_metadata(str(error), self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+        if tool_calls:
+            fatal_error = run_tool_calls(messages, message, tool_calls, self._registry, self._log_path, write_agent_log)
+            if fatal_error is not None:
+                return AgentStepResult(
+                    metadata=review_metadata(fatal_error, self._config),
+                    invalid_final_count=invalid_final_count,
+                    should_continue=False,
+                )
+            return AgentStepResult(metadata=None, invalid_final_count=invalid_final_count, should_continue=True)
+        return self.handle_final_message(messages, message, invalid_final_count)
+
+    def handle_final_message(
+        self,
+        messages: list[dict[str, object]],
+        message: dict[str, object],
+        invalid_final_count: int,
+    ) -> AgentStepResult:
+        """Validate a final model message or request one retry."""
+        content = message.get("content")
+        if not isinstance(content, str):
+            return AgentStepResult(
+                metadata=review_metadata("Ollama final response did not include string content", self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+
+        try:
+            resolved = self.validate_final(parse_final_json_content(content))
+        except (json.JSONDecodeError, MetadataResolutionError) as error:
+            return self.handle_invalid_final(messages, error, invalid_final_count)
+
+        write_agent_log(self._log_path, "final_metadata", metadata=resolved)
+        return AgentStepResult(metadata=resolved, invalid_final_count=invalid_final_count, should_continue=False)
+
+    def handle_invalid_final(
+        self,
+        messages: list[dict[str, object]],
+        error: json.JSONDecodeError | MetadataResolutionError,
+        invalid_final_count: int,
+    ) -> AgentStepResult:
+        """Log invalid final JSON and either retry or return review metadata."""
+        invalid_final_count += 1
+        write_agent_log(
+            self._log_path,
+            "final_validation_error",
+            error=str(error),
+            invalid_final_count=invalid_final_count,
+        )
+        if invalid_final_count > self._config.invalid_final_retries:
+            return AgentStepResult(
+                metadata=review_metadata(str(error), self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+        messages.append(
+            {
+                "role": "user",
+                "content": (
+                    "Your previous final answer was invalid. Return only valid JSON matching the required "
+                    f"schema. Validation error: {error}"
+                ),
+            },
+        )
+        return AgentStepResult(metadata=None, invalid_final_count=invalid_final_count, should_continue=True)
+
+    def force_final_response(self, messages: list[dict[str, object]]) -> StandardBookMetadata:
+        """Request a no-tool final answer after the normal turn limit."""
+        messages.append({"role": "user", "content": forced_final_prompt()})
+        write_agent_log(self._log_path, "forced_final_request", reason="max_turns")
+        data = self.chat(messages, self._config.max_agent_turns + 1, tools_enabled=False)
+        message = data.get("message")
+        if not isinstance(message, dict):
+            return review_metadata("Ollama forced final response did not include a message", self._config)
+        content = message.get("content")
+        if not isinstance(content, str):
+            return review_metadata("Ollama forced final response did not include string content", self._config)
+        try:
+            resolved = self.validate_final(parse_final_json_content(content))
+        except (json.JSONDecodeError, MetadataResolutionError) as error:
+            return review_metadata(f"Ollama forced final response was invalid: {error}", self._config)
+        write_agent_log(self._log_path, "final_metadata", metadata=resolved)
+        return resolved
+
+    def chat(self, messages: list[dict[str, object]], turn: int, *, tools_enabled: bool = True) -> dict[str, object]:
+        """Send one chat request to Ollama and log the request and response."""
+        payload = {
+            "model": self._config.model,
+            "messages": messages,
+            "stream": False,
+            "options": {"temperature": 0.1},
+        }
+        tool_names = []
+        if tools_enabled:
+            payload["tools"] = self._registry.tool_schemas()
+            tool_names = self._config.tool_names
+        write_agent_log(
+            self._log_path,
+            "model_request",
+            model=self._config.model,
+            turn=turn,
+            message_count=len(messages),
+            tool_names=tool_names,
+            tools_enabled=tools_enabled,
+        )
+        write_agent_log(
+            self._log_path,
+            "llm_messages_sent",
+            model=self._config.model,
+            turn=turn,
+            messages=messages,
+            tools_enabled=tools_enabled,
+        )
+        response = httpx.post(
+            self._config.ollama_chat_url,
+            headers={"Authorization": f"Bearer {self._ollama_api_key}"},
+            json=payload,
+            timeout=self._config.http_timeout_seconds,
+        )
+        response.raise_for_status()
+        raw_data = response.json()
+        if not isinstance(raw_data, dict):
+            return {}
+        data = {str(key): value for key, value in raw_data.items()}
+        message = data.get("message", {})
+        content = message.get("content") if isinstance(message, dict) else ""
+        write_agent_log(
+            self._log_path,
+            "llm_message_received",
+            model=self._config.model,
+            turn=turn,
+            message=message,
+        )
+        write_agent_log(
+            self._log_path,
+            "model_response",
+            model=self._config.model,
+            turn=turn,
+            has_tool_calls=bool(isinstance(message, dict) and message.get("tool_calls")),
+            content_chars=len(content) if isinstance(content, str) else 0,
+        )
+        return data
+
+    def validate_final(self, raw_metadata: object) -> StandardBookMetadata:
+        """Validate final model metadata against catalog rows."""
+        fields = parse_final_metadata_fields(raw_metadata)
+        fields = replace(fields, title=normalize_title_slug(fields.title))
+        author = self.validate_author(fields.author_id)
+        validate_title_slug(fields.title)
+        book_fields = self.resolve_book_fields(fields)
+        series = self.validate_series(fields.author_id, book_fields.series_id, book_fields.series_index)
+
+        return StandardBookMetadata(
+            author_id=fields.author_id,
+            author=author.name,
+            book_id=book_fields.book_id,
+            title=book_fields.title,
+            series_id=book_fields.series_id,
+            series=series,
+            series_index=book_fields.series_index,
+            confidence=fields.confidence,
+            needs_review=fields.confidence < self._config.min_confidence,
+            evidence=fields.evidence,
+        )
+
+    def validate_author(self, author_id: int) -> AudiobookAuthor:
+        """Validate that an author id was seen and exists."""
+        if author_id not in self._registry.seen_author_ids:
+            msg = f"author_id {author_id} was not returned by search_authors"
+            raise MetadataResolutionError(msg)
+        author = self._registry.get_author(author_id)
+        if author is None:
+            msg = f"author_id {author_id} does not exist"
+            raise MetadataResolutionError(msg)
+        validate_catalog_slug(author.name, "author")
+        return author
+
+    def resolve_book_fields(self, fields: FinalMetadataFields) -> ResolvedBookFields:
+        """Resolve final book fields from a seen book id or created book."""
+        if fields.book_id is None:
+            ensured = self._registry.ensure_book(
+                fields.title,
+                fields.author_id,
+                fields.series_id,
+                fields.series_index,
+            )
+            return ResolvedBookFields(
+                book_id=ensured.book.id,
+                title=ensured.book.title,
+                series_id=ensured.book.series_id,
+                series_index=ensured.book.series_index,
+            )
+
+        if fields.book_id not in self._registry.seen_book_ids:
+            msg = f"book_id {fields.book_id} was not returned by search_books"
+            raise MetadataResolutionError(msg)
+        book = self._registry.get_book(fields.book_id)
+        if book is None:
+            msg = f"book_id {fields.book_id} does not exist"
+            raise MetadataResolutionError(msg)
+        if book.author_id != fields.author_id:
+            msg = f"book_id {fields.book_id} does not belong to author_id {fields.author_id}"
+            raise MetadataResolutionError(msg)
+        return ResolvedBookFields(
+            book_id=fields.book_id,
+            title=book.title,
+            series_id=book.series_id,
+            series_index=book.series_index,
+        )
+
+    def validate_series(self, author_id: int, series_id: int | None, series_index: float) -> str:
+        """Validate final series fields and return the canonical series slug."""
+        if series_id is None:
+            if series_index != 0:
+                msg = "standalone books must use series_index 0"
+                raise MetadataResolutionError(msg)
+            return self._config.standalone_series
+
+        if series_id not in self._registry.seen_series_ids:
+            msg = f"series_id {series_id} was not returned by search_series"
+            raise MetadataResolutionError(msg)
+        series = self._registry.get_series(series_id)
+        if series is None:
+            msg = f"series_id {series_id} does not exist"
+            raise MetadataResolutionError(msg)
+        if series.author_id != author_id:
+            msg = f"series_id {series_id} does not belong to author_id {author_id}"
+            raise MetadataResolutionError(msg)
+        if series_index <= 0:
+            msg = "series books must use a positive series_index"
+            raise MetadataResolutionError(msg)
+        validate_catalog_slug(series.name, "series")
+        return series.name
+
+
+def write_agent_log(log_path: Path, event: str, **fields: object) -> None:
+    """Append one JSONL audit event."""
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    record = {
+        "created": utcnow().isoformat(),
+        "event": event,
+        **{key: json_log_value(value) for key, value in fields.items()},
+    }
+    with log_path.open("a", encoding="utf-8") as file:
+        file.write(json.dumps(record, sort_keys=True))
+        file.write("\n")
+
+
+def json_log_value(value: object) -> object:
+    """Return a JSON-serializable value for audit logs."""
+    if is_dataclass(value) and not isinstance(value, type):
+        return json_log_value(asdict(value))
+    if isinstance(value, dict):
+        return {str(key): json_log_value(item) for key, item in value.items()}
+    if isinstance(value, list | tuple):
+        return [json_log_value(item) for item in value]
+    if isinstance(value, set):
+        return [json_log_value(item) for item in sorted(value, key=str)]
+    if isinstance(value, PathLike):
+        return str(value)
+    return value
+
+
+def system_prompt() -> str:
+    """Return the stable system prompt."""
+    return """You standardize Audible audiobook metadata against a private catalog.
+
+Rules:
+- You must use the provided tools before returning final metadata.
+- Only use author_id, series_id, or book_id values returned by tools.
+- Return final metadata as JSON only. Do not wrap it in Markdown.
+- The final JSON object must contain author_id, book_id, title, series_id, series_index, confidence, and evidence.
+- title must be a canonical title slug using lower-case words separated by hyphens.
+- Use series_id null and series_index 0 for standalone books.
+- If you use a series_id, series_index must be a whole number or .5 value greater than 0.
+- Treat series slugs that differ only by underscores as the same series. Prefer the existing catalog row instead of
+  creating a new series.
+- Detect omnibus or box-set editions that contain multiple numbered novels, books, or novellas.
+- For an omnibus, make a best-effort range from the filename, tags, and catalog rows. Keep series_index as the
+  first covered book number and include the range in the title when the source title includes it, for example
+  books-1-3.
+- Be careful with omnibuses of novels or novellas later published as one book: keep the omnibus as the audiobook's
+  book record unless catalog rows clearly identify a better match.
+- Do not create publisher collections or author collections as series unless the book metadata clearly gives a
+  numbered series.
+- Series belong to authors. Use a series_id only when it belongs to the selected author_id.
+- Always search for the author before creating one. If no exact author slug exists, call ensure_author.
+- Always search for a series with author_id before creating one. If no exact series slug exists, call ensure_series.
+- Always search for a book before creating one. If no exact title slug exists, call ensure_book.
+- If a tool returns an error, correct your tool arguments or final metadata before continuing.
+- confidence must be a number from 0 to 1.
+- evidence must be a short list of strings explaining which filename, tags, and catalog rows support the answer."""
+
+
+def forced_final_prompt() -> str:
+    """Return the no-tools finalization prompt."""
+    return (
+        "Stop calling tools. Return final metadata as JSON only using the tool results already provided. "
+        "If search_books returned no matching rows but author and series are known, use book_id null and resolve "
+        "the title slug from the AAX filename and ffprobe tags. The validator will create the missing book. "
+        "Use only author_id and series_id values returned by earlier tool results."
+    )
+
+
+def user_prompt(aax_file_name: str, metadata: dict[str, str]) -> str:
+    """Build the user prompt from source metadata."""
+    return (
+        "Resolve this Audible audiobook.\n\n"
+        f"AAX file name: {aax_file_name}\n\n"
+        "ffprobe format tags:\n"
+        f"{json.dumps(metadata, indent=2, sort_keys=True)}"
+    )
+
+
+def parse_final_json_content(content: str) -> object:
+    """Parse final model content, accepting bare or fenced JSON."""
+    stripped = content.strip()
+    if match := FENCED_JSON_PATTERN.fullmatch(stripped):
+        stripped = match.group("json").strip()
+    return json.loads(stripped)
+
+
+def parse_final_metadata_fields(raw_metadata: object) -> FinalMetadataFields:
+    """Parse the model's final JSON object into typed fields."""
+    if not isinstance(raw_metadata, dict):
+        msg = "Final metadata must be a JSON object"
+        raise MetadataResolutionError(msg)
+    data = {str(key): value for key, value in raw_metadata.items()}
+    return FinalMetadataFields(
+        author_id=required_int(data, "author_id"),
+        book_id=optional_int(data.get("book_id"), "book_id"),
+        title=required_string(data, "title"),
+        series_id=optional_int(data.get("series_id"), "series_id"),
+        series_index=required_series_index(data, "series_index"),
+        confidence=required_float(data, "confidence"),
+        evidence=required_string_list(data, "evidence"),
+    )
+
+
+def review_metadata(reason: str, config: AgentConfig) -> StandardBookMetadata:
+    """Return a metadata result that must be reviewed manually."""
+    return StandardBookMetadata(
+        author_id=0,
+        author="unknown_author",
+        book_id=None,
+        title="unknown-title",
+        series_id=None,
+        series=config.standalone_series,
+        series_index=0,
+        confidence=0,
+        needs_review=True,
+        evidence=[reason],
+    )
+
+
+def required_float(data: dict[str, object], key: str) -> float:
+    """Read a required float field."""
+    value = data.get(key)
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        msg = f"{key} must be a number"
+        raise MetadataResolutionError(msg)
+    confidence = float(value)
+    if confidence < 0 or confidence > 1:
+        msg = f"{key} must be between 0 and 1"
+        raise MetadataResolutionError(msg)
+    return confidence
+
+
+def required_string_list(data: dict[str, object], key: str) -> list[str]:
+    """Read a required list of strings."""
+    value = data.get(key)
+    if not isinstance(value, list) or not value or not all(isinstance(item, str) for item in value):
+        msg = f"{key} must be a non-empty list of strings"
+        raise MetadataResolutionError(msg)
+    strings = [item.strip() for item in value if item.strip()]
+    if not strings:
+        msg = f"{key} must include at least one non-empty string"
+        raise MetadataResolutionError(msg)
+    return strings

systems/bob/default.nix

@@ -28,12 +28,7 @@
   networking = {
     hostName = "bob";
     hostId = "7c678a41";
-    firewall = {
-      enable = true;
-      allowedTCPPorts = [
-        8000
-      ];
-    };
+    firewall.enable = true;
     networkmanager.enable = true;
   };
 

systems/bob/hardware.nix

@@ -30,11 +30,6 @@
         keyFile = "/dev/disk/by-id/usb-Samsung_Flash_Drive_FIT_0374620080067131-0:0";
       };
     };
-
-    zfs.extraPools = [
-      "storage"
-    ];
-
     kernelModules = [ "kvm-amd" ];
     extraModulePackages = [ ];
   };

systems/bob/llms.nix

@@ -4,7 +4,7 @@
     host = "0.0.0.0";
     enable = true;
 
-    syncModels = true;
+    syncModels = false;
     loadModels = [
       "codellama:7b"
       "deepscaler:1.5b"

systems/jeeves/services/llms.nix

@@ -6,7 +6,7 @@ in
     user = "ollama";
     enable = true;
     host = "0.0.0.0";
-    syncModels = true;
+    syncModels = false;
     loadModels = [
       "codellama:7b"
       "deepscaler:1.5b"
@@ -30,6 +30,9 @@ in
       "ministral-3:14b"
       "nemotron-3-nano:30b"
       "qwen3-coder:30b"
+      "qwen3-embedding:0.6b"
+      "qwen3-embedding:4b"
+      "qwen3-embedding:8b"
       "qwen3-vl:32b"
       "qwen3:14b"
       "qwen3.5:35b"

systems/jeeves/services/postgress.nix

@@ -38,9 +38,6 @@ in
       # signalbot
       local signalbot  signalbot   trust
 
-      # hedgedoc
-      local hedgedoc   hedgedoc    trust
-
       # math
       local postgres  math   trust
       host  postgres  math   127.0.0.1/32    trust
@@ -120,19 +117,11 @@ in
           login = true;
         };
       }
-      {
-        name = "hedgedoc";
-        ensureDBOwnership = true;
-        ensureClauses = {
-          login = true;
-        };
-      }
     ];
     ensureDatabases = [
       "data_science_dev"
       "hass"
       "gitea"
-      "hedgedoc"
       "math"
       "n8n"
       "richie"

tests/test_audiobook_catalog.py

@@ -0,0 +1,126 @@
+"""test_audiobook_catalog."""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import sessionmaker
+
+from python.orm.richie import AudiobookAuthor, AudiobookSeries, RichieBase
+from python.tools.audiobook import catalog
+
+
+@pytest.fixture
+def audiobook_session():
+    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+    RichieBase.metadata.create_all(engine)
+    with sessionmaker(bind=engine, expire_on_commit=False, future=True)() as session:
+        yield session
+    engine.dispose()
+
+
+def test_upsert_catalog_csv_inserts_and_updates_authors_and_series(tmp_path, audiobook_session) -> None:
+    audiobook_session.add_all(
+        [
+            AudiobookAuthor(id=10, name="old_author"),
+            AudiobookAuthor(id=11, name="craig_alanson"),
+        ],
+    )
+    audiobook_session.commit()
+    authors_csv = tmp_path / "authors.csv"
+    series_csv = tmp_path / "series.csv"
+    authors_csv.write_text(
+        "name,id\n"
+        "glynn_stewart,\n"
+        "craig_alanson,\n"
+        "updated_author,10\n",
+        encoding="utf-8",
+    )
+    series_csv.write_text(
+        "name,author_name,id\n"
+        "starships_mage,glynn_stewart,\n"
+        "expeditionary_force,craig_alanson,\n",
+        encoding="utf-8",
+    )
+
+    author_count = catalog.upsert_authors_from_csv(audiobook_session, authors_csv)
+    series_count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
+    audiobook_session.commit()
+
+    authors = audiobook_session.scalars(select(AudiobookAuthor).order_by(AudiobookAuthor.id)).all()
+    series = audiobook_session.scalars(select(AudiobookSeries).order_by(AudiobookSeries.name)).all()
+    assert author_count == 3
+    assert series_count == 2
+    assert [(author.id, author.name) for author in authors] == [
+        (10, "updated_author"),
+        (11, "craig_alanson"),
+        (12, "glynn_stewart"),
+    ]
+    assert [(row.name, row.author.name) for row in series] == [
+        ("expeditionary_force", "craig_alanson"),
+        ("starships_mage", "glynn_stewart"),
+    ]
+
+
+def test_upsert_series_csv_updates_series_by_id(tmp_path, audiobook_session) -> None:
+    author = AudiobookAuthor(id=1, name="glynn_stewart")
+    audiobook_session.add_all(
+        [
+            author,
+            AudiobookSeries(id=7, name="old_series", author=author),
+        ],
+    )
+    audiobook_session.commit()
+    series_csv = tmp_path / "series.csv"
+    series_csv.write_text(
+        "name,author_name,id\n"
+        "starships_mage,glynn_stewart,7\n",
+        encoding="utf-8",
+    )
+
+    count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
+    audiobook_session.commit()
+
+    series = audiobook_session.get(AudiobookSeries, 7)
+    assert count == 1
+    assert series.name == "starships_mage"
+    assert series.author.name == "glynn_stewart"
+
+
+def test_upsert_csv_allows_missing_id_column(tmp_path, audiobook_session) -> None:
+    authors_csv = tmp_path / "authors.csv"
+    series_csv = tmp_path / "series.csv"
+    authors_csv.write_text(
+        "name\n"
+        "glynn_stewart\n",
+        encoding="utf-8",
+    )
+    series_csv.write_text(
+        "name,author_name\n"
+        "starships_mage,glynn_stewart\n",
+        encoding="utf-8",
+    )
+
+    author_count = catalog.upsert_authors_from_csv(audiobook_session, authors_csv)
+    series_count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
+    audiobook_session.commit()
+
+    series = audiobook_session.scalar(select(AudiobookSeries))
+    assert author_count == 1
+    assert series_count == 1
+    assert series.name == "starships_mage"
+    assert series.author.name == "glynn_stewart"
+
+
+def test_upsert_series_csv_rejects_unknown_author(tmp_path, audiobook_session) -> None:
+    series_csv = tmp_path / "series.csv"
+    series_csv.write_text(
+        "name,author_name,id\n"
+        "starships_mage,glynn_stewart,\n",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(catalog.CatalogImportError) as error:
+        catalog.upsert_series_from_csv(audiobook_session, series_csv)
+
+    assert "author not found: glynn_stewart" in str(error.value)