feat(embeddings): batch-tolerant import embeddings via pg-llm-batch submodule

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "vendor/pg-llm-batch"]
+	path = vendor/pg-llm-batch
+	url = https://github.com/ContextualWisdomLab/pg-llm-batch.git

README.md

@@ -609,6 +609,29 @@ vector counts, unsupported embedding model names, fake quality totals, or inert
 permanent ready-soon controls; use source-backed rows or explicit pending
 states.
 
+## Batch-tolerant LLM embeddings (optional submodule)
+
+`vendor/pg-llm-batch` is a git **submodule** (the org's standalone
+`pg_tiktoken` Postgres batch engine, Apache-2.0). It powers batch-tolerant
+hotspots such as bulk email-import embeddings. It is fully optional: naruon runs
+normally with the submodule uninitialized — the batch path degrades to the
+existing per-item embedding path whenever it is unconfigured or unavailable.
+
+To enable it (human actions):
+
+1. Initialize the submodule: `git submodule update --init vendor/pg-llm-batch`
+   and install it on the backend path (`pip install -e vendor/pg-llm-batch`).
+2. Bring up the batch Postgres:
+   `docker compose -f docker-compose.yml -f docker-compose.pg-llm-batch.yml up`.
+3. Seed per-tenant batch config in the Fernet DB (`tenant_configs`, never via
+   `os.getenv`): set `batch_embedding_enabled = true` and the Fernet-encrypted
+   `batch_embedding_dsn` (plus `batch_embedding_endpoint` / `batch_embedding_model`).
+   Provider credentials continue to resolve through `resolve_runtime_llm_provider`.
+
+Batch runs are recorded in the `llm_batch_jobs` / `llm_batch_items` control-plane
+tables (migration `0010_llm_batch_embedding`). The submodule can also run
+standalone — see `vendor/pg-llm-batch/README.md` and its own `docker-compose.yml`.
+
 ## Operations and release docs
 
 - `docs/operations/release-deployment-architecture.md`: release, CI, GHCR, and

backend/alembic/versions/0010_llm_batch_embedding.py

@@ -0,0 +1,128 @@
+"""add llm batch embedding job/item tables and tenant batch config
+
+Revision ID: 0010_llm_batch_embedding
+Revises: 0009_project_graph_projection
+Create Date: 2026-07-08 00:00:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0010_llm_batch_embedding"
+down_revision = "0009_project_graph_projection"
+
+_JOBS_TABLE = "llm_batch_jobs"
+_ITEMS_TABLE = "llm_batch_items"
+_TENANT_TABLE = "tenant_configs"
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+
+    if not inspector.has_table(_JOBS_TABLE):
+        op.create_table(
+            _JOBS_TABLE,
+            sa.Column("batch_job_uid", sa.String(), nullable=False),
+            sa.Column("organization_id", sa.String(), nullable=False),
+            sa.Column("user_id", sa.String(), nullable=False),
+            sa.Column("job_status", sa.String(), nullable=False),
+            sa.Column("model_name", sa.String(), nullable=False),
+            sa.Column("endpoint_alias", sa.String(), nullable=True),
+            sa.Column("total_items", sa.Integer(), nullable=False),
+            sa.Column("completed_items", sa.Integer(), nullable=False),
+            sa.Column("failed_items", sa.Integer(), nullable=False),
+            sa.Column("total_tokens", sa.Integer(), nullable=False),
+            sa.Column("part_count", sa.Integer(), nullable=False),
+            sa.Column("error_code", sa.String(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("batch_job_uid"),
+        )
+
+    if not inspector.has_table(_ITEMS_TABLE):
+        op.create_table(
+            _ITEMS_TABLE,
+            sa.Column("batch_item_uid", sa.String(), nullable=False),
+            sa.Column("batch_job_uid", sa.String(), nullable=False),
+            sa.Column("sequence_no", sa.Integer(), nullable=False),
+            sa.Column("part_index", sa.Integer(), nullable=False),
+            sa.Column("token_count", sa.Integer(), nullable=False),
+            sa.Column("item_status", sa.String(), nullable=False),
+            sa.Column("error_code", sa.String(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["batch_job_uid"],
+                [f"{_JOBS_TABLE}.batch_job_uid"],
+                ondelete="CASCADE",
+            ),
+            sa.PrimaryKeyConstraint("batch_item_uid"),
+        )
+
+    for table_name, index_name, columns in _batch_indexes():
+        op.create_index(index_name, table_name, columns, if_not_exists=True)
+
+    if inspector.has_table(_TENANT_TABLE):
+        for column in _tenant_batch_columns():
+            if not _has_column(inspector, _TENANT_TABLE, column.name):
+                op.add_column(_TENANT_TABLE, column)
+
+
+def downgrade() -> None:
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+
+    if inspector.has_table(_TENANT_TABLE):
+        for column in reversed(_tenant_batch_columns()):
+            if _has_column(inspector, _TENANT_TABLE, column.name):
+                op.drop_column(_TENANT_TABLE, column.name)
+
+    for table_name, index_name, _columns in reversed(_batch_indexes()):
+        if inspector.has_table(table_name):
+            op.drop_index(index_name, table_name=table_name, if_exists=True)
+
+    if inspector.has_table(_ITEMS_TABLE):
+        op.drop_table(_ITEMS_TABLE)
+    if inspector.has_table(_JOBS_TABLE):
+        op.drop_table(_JOBS_TABLE)
+
+
+def _batch_indexes() -> list[tuple[str, str, list[str]]]:
+    return [
+        (_JOBS_TABLE, "ix_llm_batch_jobs_organization_id", ["organization_id"]),
+        (_JOBS_TABLE, "ix_llm_batch_jobs_user_id", ["user_id"]),
+        (_JOBS_TABLE, "ix_llm_batch_jobs_job_status", ["job_status"]),
+        (
+            _JOBS_TABLE,
+            "ix_llm_batch_jobs_scope_status",
+            ["organization_id", "user_id", "job_status"],
+        ),
+        (_ITEMS_TABLE, "ix_llm_batch_items_batch_job_uid", ["batch_job_uid"]),
+        (_ITEMS_TABLE, "ix_llm_batch_items_item_status", ["item_status"]),
+        (
+            _ITEMS_TABLE,
+            "ix_llm_batch_items_job_sequence",
+            ["batch_job_uid", "sequence_no"],
+        ),
+    ]
+
+
+def _tenant_batch_columns() -> list["sa.Column"]:
+    return [
+        sa.Column(
+            "batch_embedding_enabled",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.false(),
+        ),
+        sa.Column("batch_embedding_dsn", sa.String(), nullable=True),
+        sa.Column("batch_embedding_endpoint", sa.String(), nullable=True),
+        sa.Column("batch_embedding_model", sa.String(), nullable=True),
+    ]
+
+
+def _has_column(inspector, table_name: str, column_name: str) -> bool:
+    return any(
+        column["name"] == column_name for column in inspector.get_columns(table_name)
+    )

backend/db/models.py

@@ -274,6 +274,116 @@ class ProviderWritebackRetryItem(Base):
     )
 
 
+class LlmBatchJob(Base):
+    """A batch-tolerant embedding/completion job routed via pg-llm-batch.
+
+    naruon-side control-plane mirror of the component's ``llm_batches`` table.
+    Records one job per bulk embedding run (e.g. an email import batch) so the
+    batched work has a durable audit trail even though the JSONL assembly lives
+    in the batch engine's own Postgres. Modeled on
+    :class:`ProviderWritebackRetryItem` (string uid PK, scope indexes).
+    """
+
+    __tablename__ = "llm_batch_jobs"
+
+    batch_job_uid: Mapped[str] = mapped_column(
+        String,
+        primary_key=True,
+        default=lambda: f"llm_batch_{uuid.uuid4().hex}",
+    )
+    organization_id: Mapped[str] = mapped_column(String, index=True, nullable=False)
+    user_id: Mapped[str] = mapped_column(String, index=True, nullable=False)
+    job_status: Mapped[str] = mapped_column(
+        String,
+        index=True,
+        default="preparing",
+        nullable=False,
+    )
+    model_name: Mapped[str] = mapped_column(String, nullable=False)
+    endpoint_alias: Mapped[str | None] = mapped_column(String, nullable=True)
+    total_items: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    completed_items: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    failed_items: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    total_tokens: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    part_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    error_code: Mapped[str | None] = mapped_column(String, nullable=True)
+    created_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.datetime.now(datetime.timezone.utc),
+        nullable=False,
+    )
+    updated_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.datetime.now(datetime.timezone.utc),
+        onupdate=lambda: datetime.datetime.now(datetime.timezone.utc),
+        nullable=False,
+    )
+    items: Mapped[list["LlmBatchItem"]] = relationship(
+        back_populates="job",
+        cascade="all, delete-orphan",
+    )
+    __table_args__ = (
+        Index(
+            "ix_llm_batch_jobs_scope_status",
+            "organization_id",
+            "user_id",
+            "job_status",
+        ),
+    )
+
+
+class LlmBatchItem(Base):
+    """A single request within an :class:`LlmBatchJob`.
+
+    naruon-side mirror of the component's ``llm_requests`` rows. One item per
+    input text, carrying its token count and the partition (batch file part) it
+    was assigned to by the engine's token/byte/record accumulator.
+    """
+
+    __tablename__ = "llm_batch_items"
+
+    batch_item_uid: Mapped[str] = mapped_column(
+        String,
+        primary_key=True,
+        default=lambda: f"llm_batch_item_{uuid.uuid4().hex}",
+    )
+    batch_job_uid: Mapped[str] = mapped_column(
+        String,
+        ForeignKey("llm_batch_jobs.batch_job_uid", ondelete="CASCADE"),
+        index=True,
+        nullable=False,
+    )
+    sequence_no: Mapped[int] = mapped_column(Integer, nullable=False)
+    part_index: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    token_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    item_status: Mapped[str] = mapped_column(
+        String,
+        index=True,
+        default="queued",
+        nullable=False,
+    )
+    error_code: Mapped[str | None] = mapped_column(String, nullable=True)
+    created_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.datetime.now(datetime.timezone.utc),
+        nullable=False,
+    )
+    updated_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=lambda: datetime.datetime.now(datetime.timezone.utc),
+        onupdate=lambda: datetime.datetime.now(datetime.timezone.utc),
+        nullable=False,
+    )
+    job: Mapped["LlmBatchJob"] = relationship(back_populates="items")
+    __table_args__ = (
+        Index(
+            "ix_llm_batch_items_job_sequence",
+            "batch_job_uid",
+            "sequence_no",
+        ),
+    )
+
+
 class Organization(Base):
     __tablename__ = "organization_entities"
 
@@ -1042,6 +1152,20 @@ class TenantConfig(Base):
         EncryptedString, nullable=True
     )
 
+    # Batch-tolerant embedding routing (pg-llm-batch submodule). All config here
+    # lives in the Fernet DB, never in os.getenv. The batch Postgres DSN is a
+    # connection secret, so it is stored EncryptedString (Fernet at rest).
+    batch_embedding_enabled: Mapped[bool] = mapped_column(
+        Boolean, default=False, nullable=False
+    )
+    batch_embedding_dsn: Mapped[str | None] = mapped_column(
+        EncryptedString, nullable=True
+    )
+    batch_embedding_endpoint: Mapped[str | None] = mapped_column(
+        String, nullable=True
+    )
+    batch_embedding_model: Mapped[str | None] = mapped_column(String, nullable=True)
+
     def __repr__(self) -> str:
         return (
             f"<TenantConfig(id={self.id}, user_id='{self.user_id}', "