Backend: bundled-lite engine mode (SQLite + inline queue) for the desktop app

The desktop app now runs the backend as a single embedded process: no
Postgres, no Redis, no Docker. Server deployments are unaffected
(defaults unchanged: postgres + arq).

- dialect-neutral schema: JSON renders JSONB on PG / JSON on SQLite;
  new UTCDT type returns tz-aware datetimes from SQLite (naive rows
  crashed utcnow() comparisons); PublicIdMixin.updated_at gains an
  insert default (PG had silently tolerated NULLs); partial indexes get
  sqlite_where; tsvector migration is a PG-only no-op elsewhere
- SQLite connections: PRAGMA foreign_keys=ON, WAL, busy_timeout
- queue_backend=inline: in-process asyncio consumer replaces arq
  (shonar/services/inline_queue.py). DB rows remain the queue of record;
  startup sweep re-runs pending jobs; transient failures retry with
  backoff up to MAX_TRIES, one job at a time (faster-whisper memory)
- SHONAR_AUTO_MIGRATE=1: alembic upgrade head at startup (the desktop
  owns its SQLite file end-to-end)
- tests run on either dialect via SHONAR_TEST_DATABASE_URL; 2 new
  inline-queue tests (67 green on SQLite AND Postgres, ruff clean)
- verified live: bundled engine on SQLite + inline queue + faster-whisper
  base transcribed a real upload end-to-end (queued -> succeeded ->
  transcript v1 -> completed)

backend/migrations/versions/8d51af959ae4_initial_schema.py

@@ -19,6 +19,13 @@ branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 
 
+def _json_type() -> sa.types.TypeEngine:
+    """JSONB on PostgreSQL, plain JSON on SQLite (desktop bundled engine)."""
+    if op.get_context().dialect.name == "postgresql":
+        return postgresql.JSONB(astext_type=sa.Text())
+    return sa.JSON()
+
+
 def upgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
     op.create_table('users',
@@ -81,7 +88,7 @@ def upgrade() -> None:
     )
     op.create_index(op.f('ix_recordings_client_recording_id'), 'recordings', ['client_recording_id'], unique=False)
     op.create_index(op.f('ix_recordings_processing_status'), 'recordings', ['processing_status'], unique=False)
-    op.create_index('ix_recordings_user_client_id', 'recordings', ['user_id', 'client_recording_id'], unique=True, postgresql_where='client_recording_id IS NOT NULL')
+    op.create_index('ix_recordings_user_client_id', 'recordings', ['user_id', 'client_recording_id'], unique=True, postgresql_where='client_recording_id IS NOT NULL', sqlite_where=sa.text('client_recording_id IS NOT NULL'))
     op.create_index(op.f('ix_recordings_user_id'), 'recordings', ['user_id'], unique=False)
     op.create_index('ix_recordings_user_recorded', 'recordings', ['user_id', 'recorded_at'], unique=False)
     op.create_table('refresh_tokens',
@@ -120,7 +127,7 @@ def upgrade() -> None:
     )
     op.create_index(op.f('ix_assets_recording_id'), 'assets', ['recording_id'], unique=False)
     op.create_index(op.f('ix_assets_user_id'), 'assets', ['user_id'], unique=False)
-    op.create_index('uq_assets_one_original', 'assets', ['recording_id'], unique=True, postgresql_where="kind = 'original'")
+    op.create_index('uq_assets_one_original', 'assets', ['recording_id'], unique=True, postgresql_where="kind = 'original'", sqlite_where=sa.text("kind = 'original'"))
     op.create_table('processing_jobs',
     sa.Column('recording_id', sa.Uuid(), nullable=False),
     sa.Column('job_type', sa.Enum('normalize_audio', 'transcribe', 'summarize', name='job_type'), nullable=False),
@@ -153,7 +160,7 @@ def upgrade() -> None:
     sa.Column('superseded_at', sa.DateTime(timezone=True), nullable=True),
     sa.Column('provider', sa.String(length=80), nullable=False),
     sa.Column('model', sa.String(length=120), nullable=True),
-    sa.Column('content', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('content', _json_type(), nullable=False),
     sa.Column('edited_by_user', sa.Boolean(), nullable=False),
     sa.Column('id', sa.Uuid(), nullable=False),
     sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
@@ -170,7 +177,7 @@ def upgrade() -> None:
     sa.Column('provider', sa.String(length=80), nullable=False),
     sa.Column('model', sa.String(length=120), nullable=True),
     sa.Column('text', sa.Text(), nullable=False),
-    sa.Column('segments', postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+    sa.Column('segments', _json_type(), nullable=True),
     sa.Column('edited_by_user', sa.Boolean(), nullable=False),
     sa.Column('id', sa.Uuid(), nullable=False),
     sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),

backend/migrations/versions/fts0000000001_fts_columns.py

@@ -20,6 +20,11 @@ depends_on: str | Sequence[str] | None = None
 
 
 def upgrade() -> None:
+    # tsvector is PostgreSQL-only. On SQLite (desktop bundled engine) full-
+    # text search is handled outside the DB (the app scans report files),
+    # so this migration is a no-op.
+    if op.get_context().dialect.name != "postgresql":
+        return
     # recordings: title + notes + tag names are indexed; transcript/summary
     # contribute via their own tables (joined at query time).
     op.execute(
@@ -74,6 +79,8 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
+    if op.get_context().dialect.name != "postgresql":
+        return
     for table in ("tags", "summaries", "transcripts", "recordings"):
         op.execute(f"DROP INDEX IF EXISTS ix_{table}_search;")
         op.execute(f"ALTER TABLE {table} DROP COLUMN IF EXISTS search_vector;")

backend/migrations/versions/m8models000001_model_support.py

@@ -15,12 +15,19 @@ import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 
-revision: str = "m8models000001"
+revision: str = 'm8models000001'
-down_revision: str | None = "fts0000000001"
+down_revision: str | None = 'fts0000000001'
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 
 
+def _json_type() -> sa.types.TypeEngine:
+    """JSONB on PostgreSQL, plain JSON on SQLite (desktop bundled engine)."""
+    if op.get_context().dialect.name == "postgresql":
+        return postgresql.JSONB(astext_type=sa.Text())
+    return sa.JSON()
+
+
 def upgrade() -> None:
     op.add_column(
         "recordings",
@@ -43,9 +50,8 @@ def upgrade() -> None:
         sa.Column("key", sa.String(length=120), nullable=False),
         sa.Column(
             "value",
-            postgresql.JSONB(astext_type=sa.Text()),
+            _json_type(),
             nullable=False,
-            server_default=sa.text("'{}'::jsonb"),
         ),
         sa.Column("id", sa.Uuid(), nullable=False),
         sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),

backend/shonar/core/config.py

@@ -64,6 +64,14 @@ class Settings(BaseSettings):
 
     # --- Worker / queue -------------------------------------------------------
     redis_url: str = "redis://localhost:6379/0"
+    # "arq" = Redis transport (server deployments). "inline" = in-process
+    # asyncio runner (desktop bundled-lite engine: no Redis, one job at a
+    # time). DB rows are the queue of record either way.
+    queue_backend: str = "arq"  # arq | inline
+    # Apply 'alembic upgrade head' at startup. The desktop bundled engine
+    # owns its SQLite file and must self-migrate; server deployments run
+    # migrations in their deploy flow instead.
+    auto_migrate: bool = False
 
     # --- Audio processing -----------------------------------------------------
     # Optional server-side conversion via FFmpeg. Off by default.

backend/shonar/db/base.py

@@ -5,10 +5,27 @@ from __future__ import annotations
 import uuid
 from datetime import UTC, datetime
 
-from sqlalchemy import DateTime, Uuid
+from sqlalchemy import DateTime, TypeDecorator, Uuid
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
 
+class UTCDT(TypeDecorator):
+    """Timezone-aware datetime that survives SQLite round-trips.
+
+    Postgres returns aware values (impl is a no-op there); SQLite has no
+    tz info and hands back naive datetimes, which crash comparisons with
+    ``utcnow()``. Attach UTC on read whenever the driver lost it.
+    """
+
+    impl = DateTime(timezone=True)
+    cache_ok = True
+
+    def process_result_value(self, value, dialect):  # noqa: ANN001, ANN201
+        if value is not None and value.tzinfo is None:
+            return value.replace(tzinfo=UTC)
+        return value
+
+
 def utcnow() -> datetime:
     return datetime.now(UTC)
 
@@ -25,9 +42,9 @@ class PublicIdMixin:
     """Gives every table a UUID public id; integer PKs stay internal."""
 
     id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=new_uuid)
-    created_at: Mapped[datetime] = mapped_column(
+    created_at: Mapped[datetime] = mapped_column(UTCDT(), default=utcnow, nullable=False)
-        DateTime(timezone=True), default=utcnow, nullable=False
+    # default= as well as onupdate=: SQLite enforces NOT NULL on insert
-    )
+    # (Postgres silently stored NULL for rows that never got an UPDATE).
     updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), default=utcnow, onupdate=utcnow, nullable=False
+        UTCDT(), default=utcnow, onupdate=utcnow, nullable=False
     )

backend/shonar/db/migrate.py

@@ -0,0 +1,33 @@
+"""Programmatic schema migrations for the bundled desktop engine.
+
+Server deployments run ``alembic upgrade head`` in their deploy flow;
+the desktop app owns its SQLite file end-to-end, so the API applies
+migrations itself at startup (``SHONAR_AUTO_MIGRATE=1``).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from pathlib import Path
+
+from alembic import command
+from alembic.config import Config
+
+logger = logging.getLogger("shonar.migrate")
+
+# backend/ dir: migrations/ + alembic.ini live here relative to this file.
+_BACKEND_DIR = Path(__file__).resolve().parents[2]
+
+
+async def upgrade_head(database_url: str) -> None:
+    """Run 'alembic upgrade head' for `database_url` without blocking the loop."""
+
+    def _run() -> None:
+        cfg = Config(str(_BACKEND_DIR / "alembic.ini"))
+        cfg.set_main_option("script_location", str(_BACKEND_DIR / "migrations"))
+        cfg.set_main_option("sqlalchemy.url", database_url)
+        command.upgrade(cfg, "head")
+
+    await asyncio.to_thread(_run)
+    logger.info("database schema migrated to head")

backend/shonar/db/models.py

@@ -13,10 +13,12 @@ import enum
 import uuid
 from datetime import datetime
 
+# JSON that renders JSONB on PostgreSQL and plain JSON on SQLite (the
+# desktop bundled-lite engine runs on SQLite; JSONB does not compile there).
+from sqlalchemy import JSON as _JSON
 from sqlalchemy import (
     BigInteger,
     Boolean,
-    DateTime,
     Enum,
     Float,
     ForeignKey,
@@ -25,17 +27,12 @@ from sqlalchemy import (
     Text,
     UniqueConstraint,
 )
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
-from shonar.db.base import Base, PublicIdMixin, utcnow
+from shonar.db.base import UTCDT, Base, PublicIdMixin, utcnow
 
-# JSON type that uses JSONB on PostgreSQL and plain JSON elsewhere (tests on
+JSONType = _JSON().with_variant(JSONB(), "postgresql")
-# SQLite would need JSONB emulation; we run tests against Postgres, so JSONB
-# is fine, with a generic JSON fallback for ad-hoc SQLite use).
-try:  # pragma: no cover
-    from sqlalchemy.dialects.postgresql import JSONB as JSONType
-except ImportError:  # pragma: no cover
-    from sqlalchemy import JSON as JSONType  # type: ignore[attr-defined,no-redef]
 
 
 # ---------------------------------------------------------------------------
@@ -53,7 +50,7 @@ class User(Base, PublicIdMixin):
     # Feature switches the user controls (mirror of app settings, server truth
     # for e.g. whether location metadata may be stored at all).
     location_storage_enabled: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
-    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    deleted_at: Mapped[datetime | None] = mapped_column(UTCDT())
 
     devices: Mapped[list[Device]] = relationship(back_populates="user")
     recordings: Mapped[list[Recording]] = relationship(back_populates="user")
@@ -67,8 +64,8 @@ class Device(Base, PublicIdMixin):
     )
     name: Mapped[str] = mapped_column(String(120), nullable=False)
     platform: Mapped[str] = mapped_column(String(40), default="android", nullable=False)
-    last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
+    last_seen_at: Mapped[datetime] = mapped_column(UTCDT(), default=utcnow)
-    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    revoked_at: Mapped[datetime | None] = mapped_column(UTCDT())
 
     user: Mapped[User] = relationship(back_populates="devices")
 
@@ -88,8 +85,8 @@ class RefreshToken(Base, PublicIdMixin):
     device_id: Mapped[uuid.UUID | None] = mapped_column(
         ForeignKey("devices.id", ondelete="SET NULL")
     )
-    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    expires_at: Mapped[datetime] = mapped_column(UTCDT(), nullable=False)
-    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    revoked_at: Mapped[datetime | None] = mapped_column(UTCDT())
     replaced_by: Mapped[uuid.UUID | None] = mapped_column(
         ForeignKey("refresh_tokens.id", ondelete="SET NULL")
     )
@@ -124,7 +121,7 @@ class Recording(Base, PublicIdMixin):
     client_recording_id: Mapped[str | None] = mapped_column(String(64), index=True)
 
     title: Mapped[str] = mapped_column(String(300), nullable=False, default="Untitled recording")
-    recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    recorded_at: Mapped[datetime] = mapped_column(UTCDT(), nullable=False)
     duration_seconds: Mapped[float] = mapped_column(Float, default=0.0, nullable=False)
     notes: Mapped[str | None] = mapped_column(Text)
     # Location is stored ONLY when the user has explicitly enabled it.
@@ -151,7 +148,7 @@ class Recording(Base, PublicIdMixin):
     # The original audio is the Asset row with kind=original for this
     # recording (at most one, enforced by a partial unique index). Keeping
     # the pointer one-directional avoids a recordings<->assets FK cycle.
-    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    deleted_at: Mapped[datetime | None] = mapped_column(UTCDT())
 
     user: Mapped[User] = relationship(back_populates="recordings")
     assets: Mapped[list[Asset]] = relationship(
@@ -255,7 +252,7 @@ class UploadSession(Base, PublicIdMixin):
         default=UploadSessionStatus.open,
         nullable=False,
     )
-    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    expires_at: Mapped[datetime] = mapped_column(UTCDT(), nullable=False)
     completed_asset_id: Mapped[uuid.UUID | None] = mapped_column(
         ForeignKey("assets.id", ondelete="SET NULL")
     )
@@ -276,7 +273,7 @@ class UploadChunk(Base):
     size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False)
     checksum_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
     storage_key: Mapped[str] = mapped_column(String(500), nullable=False)
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
+    created_at: Mapped[datetime] = mapped_column(UTCDT(), default=utcnow)
 
     session: Mapped[UploadSession] = relationship(back_populates="chunks")
 
@@ -297,7 +294,7 @@ class Transcript(Base, PublicIdMixin):
     # Versioned: regenerated transcripts supersede older rows; the newest
     # non-superseded row is authoritative. ``edited_by_user`` rows win.
     version: Mapped[int] = mapped_column(default=1, nullable=False)
-    superseded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    superseded_at: Mapped[datetime | None] = mapped_column(UTCDT())
     language: Mapped[str | None] = mapped_column(String(16))
     provider: Mapped[str] = mapped_column(String(80), nullable=False, default="manual")
     model: Mapped[str | None] = mapped_column(String(120))
@@ -332,7 +329,7 @@ class Summary(Base, PublicIdMixin):
         ForeignKey("recordings.id", ondelete="CASCADE"), index=True, nullable=False
     )
     version: Mapped[int] = mapped_column(default=1, nullable=False)
-    superseded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    superseded_at: Mapped[datetime | None] = mapped_column(UTCDT())
     provider: Mapped[str] = mapped_column(String(80), nullable=False, default="manual")
     model: Mapped[str | None] = mapped_column(String(120))
     content: Mapped[dict] = mapped_column(JSONType, nullable=False, default=dict)
@@ -414,8 +411,8 @@ class ProcessingJob(Base, PublicIdMixin):
     stage: Mapped[str | None] = mapped_column(String(32))
     # 0-100 work estimate within the current stage, when known.
     progress: Mapped[int | None] = mapped_column()
-    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    started_at: Mapped[datetime | None] = mapped_column(UTCDT())
-    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    finished_at: Mapped[datetime | None] = mapped_column(UTCDT())
     # Opaque arq task handle for observability.
     task_handle: Mapped[str | None] = mapped_column(String(120))
 

backend/shonar/db/session.py

@@ -18,16 +18,38 @@ def get_engine():
         settings = get_settings()
         kwargs: dict = {"pool_pre_ping": True}
         url = settings.database_url
-        # SQLite (tests) does not support the pg pool sizing kwargs.
+        # SQLite (tests, desktop bundled engine) does not support the pg
+        # pool sizing kwargs.
         if url.startswith("sqlite"):
             kwargs = {}
         else:
             kwargs.update(pool_size=settings.db_pool_size, max_overflow=settings.db_max_overflow)
         _engine = create_async_engine(url, **kwargs)
+        if url.startswith("sqlite"):
+            _configure_sqlite(_engine)
         _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
     return _engine
 
 
+def _configure_sqlite(engine) -> None:  # noqa: ANN001
+    """Per-connection SQLite pragmas.
+
+    foreign_keys is OFF by default in SQLite; the schema leans on
+    ON DELETE CASCADE, so every connection must enable it. WAL lets the
+    inline-queue writer and API readers coexist without SQLITE_BUSY
+    storms on the desktop box.
+    """
+    from sqlalchemy import event
+
+    @event.listens_for(engine.sync_engine, "connect")
+    def _pragmas(dbapi_conn, _record):  # noqa: ANN001
+        cur = dbapi_conn.cursor()
+        cur.execute("PRAGMA foreign_keys=ON")
+        cur.execute("PRAGMA journal_mode=WAL")
+        cur.execute("PRAGMA busy_timeout=5000")
+        cur.close()
+
+
 async def dispose_engine() -> None:
     global _engine, _session_factory
     if _engine is not None:

backend/shonar/main.py

@@ -29,7 +29,20 @@ async def lifespan(app: FastAPI):
     get_engine()  # validate URL parses; connections are lazy
     for warning in settings.validate_production():
         logger.warning("CONFIG: %s", warning)
+    if settings.auto_migrate:
+        from shonar.db.migrate import upgrade_head
+
+        await upgrade_head(settings.database_url)
+    inline = settings.queue_backend.strip().lower() == "inline"
+    if inline:
+        from shonar.services import inline_queue
+
+        await inline_queue.start()
     yield
+    if inline:
+        from shonar.services import inline_queue
+
+        await inline_queue.stop()
     await dispose_engine()
 
 

backend/shonar/services/inline_queue.py

@@ -0,0 +1,92 @@
+"""In-process job runner (desktop bundled-lite engine).
+
+``queue_backend=inline`` replaces the arq/Redis transport with a single
+asyncio consumer inside the uvicorn process: DB rows stay the source of
+truth (ProcessingJob), this just runs the work. One job at a time —
+local faster-whisper is multi-GB per pass, mirroring the worker's
+``max_jobs=1`` rule.
+
+Transient failures retry in-process with backoff up to MAX_TRIES (the
+same budget arq gives via ``max_tries``); the final failure is recorded
+by the task body itself (see processing._fail).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+
+from shonar.db.models import JobType
+from shonar.services import processing
+
+logger = logging.getLogger("shonar.inline_queue")
+
+RETRY_DELAY_SECONDS = 5.0
+
+_queue: asyncio.Queue[tuple[str, str, int]] | None = None
+_consumer: asyncio.Task | None = None
+
+
+def _get_queue() -> asyncio.Queue[tuple[str, str, int]]:
+    global _queue
+    if _queue is None:
+        _queue = asyncio.Queue()
+    return _queue
+
+
+async def start() -> None:
+    """Start the consumer and re-run anything the DB says is pending."""
+    global _consumer
+    _get_queue()
+    if _consumer is None or _consumer.done():
+        _consumer = asyncio.create_task(_consume(), name="shonar-inline-queue")
+    # Crash recovery: queued rows (and orphaned running rows requeued by
+    # the sweep) go back on the in-process queue.
+    count = await processing.sweep_stale()
+    if count:
+        logger.info("inline queue startup sweep requeued %d jobs", count)
+
+
+async def stop() -> None:
+    global _consumer
+    if _consumer is not None:
+        _consumer.cancel()
+        with contextlib.suppress(BaseException):  # noqa: BLE001 — shutdown is best-effort
+            await _consumer
+        _consumer = None
+
+
+async def enqueue(job_type: JobType, recording_id: str) -> None:
+    await _get_queue().put((job_type.value, str(recording_id), 1))
+
+
+async def _consume() -> None:
+    q = _get_queue()
+    while True:
+        job_value, recording_id, attempt = await q.get()
+        ctx = {"job_try": attempt}
+        try:
+            if job_value == JobType.transcribe.value:
+                await processing.run_transcribe(ctx, recording_id)
+            else:
+                await processing.run_summarize(ctx, recording_id)
+        except processing.ProviderTransientError as e:
+            if attempt < processing.MAX_TRIES:
+                logger.warning(
+                    "inline job %s:%s transient failure (%s); retry %d/%d",
+                    job_value, recording_id, e, attempt + 1, processing.MAX_TRIES,
+                )
+                await asyncio.sleep(RETRY_DELAY_SECONDS)
+                await q.put((job_value, recording_id, attempt + 1))
+            else:
+                logger.error(
+                    "inline job %s:%s failed after %d tries: %s",
+                    job_value, recording_id, attempt, e,
+                )
+        except asyncio.CancelledError:
+            raise
+        except Exception:  # noqa: BLE001 — consumer must survive any task crash
+            logger.exception("inline job %s:%s crashed", job_value, recording_id)
+        finally:
+            q.task_done()

backend/shonar/services/processing.py

@@ -144,8 +144,14 @@ async def latest_transcript_text(
 
 
 async def transport_enqueue(job_type: JobType, recording_id: uuid.UUID) -> None:
-    """Best-effort arq trigger. Failure only logs: the DB rows are the real
+    """Best-effort trigger for the configured backend ('arq' or 'inline').
-    queue and the worker sweep picks up anything the transport missed."""
+    Failure only logs: the DB rows are the real queue and the sweep picks
+    up anything the transport missed."""
+    if get_settings().queue_backend.strip().lower() == "inline":
+        from shonar.services import inline_queue
+
+        await inline_queue.enqueue(job_type, str(recording_id))
+        return
     from arq import create_pool
     from arq.connections import RedisSettings
     from arq.constants import result_key_prefix

backend/tests/conftest.py

@@ -48,20 +48,26 @@ async def _setup_db() -> AsyncIterator[None]:
 async def clean_db(_setup_db: None) -> AsyncIterator[None]:
     """Truncate between tests for isolation."""
     yield
-    from sqlalchemy import text
+    from sqlalchemy import delete, text
 
-    from shonar.db.session import _session_factory  # type: ignore[attr-defined]
+    from shonar.db.base import Base
+    from shonar.db.session import _session_factory, get_engine  # type: ignore[attr-defined]
 
     assert _session_factory is not None
     async with _session_factory() as s:
-        await s.execute(
+        if get_engine().dialect.name == "postgresql":
-            text(
+            await s.execute(
-                "TRUNCATE users, devices, refresh_tokens, recordings, assets, "
+                text(
-                "upload_sessions, upload_chunks, transcripts, summaries, tags, "
+                    "TRUNCATE users, devices, refresh_tokens, recordings, assets, "
-                "recording_tags, processing_jobs, export_jobs, app_settings "
+                    "upload_sessions, upload_chunks, transcripts, summaries, tags, "
-                "RESTART IDENTITY CASCADE"
+                    "recording_tags, processing_jobs, export_jobs, app_settings "
+                    "RESTART IDENTITY CASCADE"
+                )
             )
-        )
+        else:
+            # SQLite: delete child-first (reverse dependency order).
+            for table in reversed(Base.metadata.sorted_tables):
+                await s.execute(delete(table))
         await s.commit()
 
 

backend/tests/test_inline_queue.py

@@ -0,0 +1,101 @@
+"""Inline queue tests (desktop bundled-lite engine): with
+``queue_backend=inline`` an upload runs transcribe→summarize automatically
+in-process — no arq, no Redis. Providers are fakes (see test_ai_pipeline).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+
+from sqlalchemy import select
+
+from shonar.core.config import get_settings
+from shonar.db import session as db_session
+from shonar.db.models import JobStatus, JobType, ProcessingJob
+from shonar.services import inline_queue, processing
+
+from .test_ai_pipeline import FakeLlm, FakeTranscriber, upload_recording, use_fakes, user_tokens
+
+
+def use_inline(monkeypatch):
+    """Force queue_backend=inline for everything resolved via processing."""
+    inline_settings = get_settings().model_copy(update={"queue_backend": "inline"})
+    monkeypatch.setattr(processing, "get_settings", lambda: inline_settings)
+
+
+async def _wait_terminal(recording_id: str, want: int = 2, timeout: float = 20.0):
+    """Poll the DB until `want` jobs reach a terminal state."""
+    deadline = asyncio.get_running_loop().time() + timeout
+    rows: list[ProcessingJob] = []
+    while asyncio.get_running_loop().time() < deadline:
+        async with db_session._session_factory() as s:
+            rows = list(
+                (
+                    await s.scalars(
+                        select(ProcessingJob).where(
+                            ProcessingJob.recording_id == uuid.UUID(recording_id)
+                        )
+                    )
+                ).all()
+            )
+            if sum(
+                1
+                for j in rows
+                if j.status in (JobStatus.succeeded, JobStatus.failed, JobStatus.skipped)
+            ) >= want:
+                return {j.job_type: j.status for j in rows}
+        await asyncio.sleep(0.1)
+    raise AssertionError(
+        f"jobs did not reach terminal state in {timeout}s: "
+        f"{[(j.job_type, j.status) for j in rows]}"
+    )
+
+
+async def test_inline_backend_processes_upload_without_arq(client, monkeypatch):
+    use_fakes(monkeypatch, FakeTranscriber(), FakeLlm())
+    use_inline(monkeypatch)
+    await inline_queue.start()
+    try:
+        token = await user_tokens(client, email="inline@shonar.dev")
+        rec = await upload_recording(client, token, client_id="inline-1")
+
+        statuses = await _wait_terminal(rec["id"], want=2)
+        assert statuses[JobType.transcribe] == JobStatus.succeeded
+        assert statuses[JobType.summarize] == JobStatus.succeeded
+
+        h = {"Authorization": f"Bearer {token}"}
+        r = await client.get(f"/api/v1/recordings/{rec['id']}", headers=h)
+        assert r.json()["processing_status"] == "completed"
+        t = await client.get(f"/api/v1/recordings/{rec['id']}/transcript", headers=h)
+        assert t.json()["text"] == "hello world from the meeting"
+        s = await client.get(f"/api/v1/recordings/{rec['id']}/summary", headers=h)
+        assert s.json()["content"]["short"] == "Standup happened."
+    finally:
+        await inline_queue.stop()
+
+
+async def test_inline_retries_transient_failure(client, monkeypatch):
+    stt = FakeTranscriber()
+    state = {"tries": 0}
+    original = stt.transcribe
+
+    async def flaky(audio, mime, *, language_hint=None):
+        state["tries"] += 1
+        if state["tries"] == 1:
+            raise processing.ProviderTransientError("temporary hiccup")
+        return await original(audio, mime, language_hint=language_hint)
+
+    stt.transcribe = flaky
+    use_fakes(monkeypatch, stt, FakeLlm())
+    use_inline(monkeypatch)
+    monkeypatch.setattr(inline_queue, "RETRY_DELAY_SECONDS", 0.05)
+    await inline_queue.start()
+    try:
+        token = await user_tokens(client, email="inline2@shonar.dev")
+        rec = await upload_recording(client, token, client_id="inline-2")
+        statuses = await _wait_terminal(rec["id"], want=2)
+        assert statuses[JobType.transcribe] == JobStatus.succeeded
+        assert state["tries"] >= 2  # the retry actually happened
+    finally:
+        await inline_queue.stop()